1 项目概述

CyclicAgent 是一个专为创建 LLM 驱动的完全自主 AI Agent 而设计的框架。该框架的核心创新在于将 Agent 抽象为有限状态机（FSM），采用状态设计模式实现。在每个状态中，Agent 根据内部状态属性（如内存、元提示等）以及外部信号来推断下一个状态，并与外部环境进行交互。

资料来源：README.md:1-10

2 核心设计理念

2.1 有限状态机架构

CyclicAgent 的核心理念基于有限状态机（FSM）设计。框架中的所有状态都实现了一个状态转换函数，该函数返回另一个状态对象。这种设计允许状态转换操作无限链接，从而使 Agent 变得“循环”。

graph TD
    A[初始状态] --> B[State.next]
    B --> C{状态转换}
    C -->|State A| D[执行动作]
    C -->|State B| E[执行动作]
    D --> B
    E --> B

2.2 状态设计模式

每个状态（State）都包含一个 next() 方法，该方法负责确定下一个要转换到的状态。状态可以携带内部属性，如内存、提示词等，这些属性在状态转换过程中会被保留和传递。

资料来源：cyclic_agent/state.py:6-9

3 核心组件

3.1 State 基类

State 是框架的基础类，定义在 cyclic_agent/state.py 中。所有自定义状态都必须继承此类并实现 next() 方法。

class State[SigT](BaseModel):
    @abstractmethod
    def next(self, signal: SigT | None = None) -> State[SigT]:
        """Transition to the next state."""
        raise NotImplementedError

State 类使用 Pydantic 的 BaseModel 作为基类，提供了数据验证和序列化能力。每个状态都有一个泛型参数 SigT 表示信号的接收类型，next() 方法接收一个可选的信号参数并返回下一个状态。

资料来源：cyclic_agent/state.py:1-10

3.2 CyclicExecutor 执行器

CyclicExecutor 负责执行状态机的运行循环，支持暂停、恢复和终止操作。

class CyclicExecutor:
    def __init__(self, default_time_interval: float):
        self.running = False
        self.lock = threading.Lock()
        self.default_time_interval = default_time_interval
        self.killed = False
        self.thread = None

执行器使用独立线程运行主循环，在每次状态转换后根据 default_time_interval 参数进行休眠，实现周期性的状态检查和转换。

资料来源：cyclic_agent/executor.py:1-33

3.3 执行器控制方法

资料来源：cyclic_agent/executor.py:11-24

3.4 预置状态组件

框架提供了两个预置的状态组件用于常见场景：

Search 组件允许定义搜索查询和回调函数来根据搜索结果决定后续状态：

class Search(State[None]):
    query: str
    exit_: Callable[[[Annotated[str, "search result"]]], State]

    def next(self, signal: None = None) -> State:
        search_result = self.search(self.query)
        return self.exit_(search_result)

资料来源：cyclic_agent/search.py:1-13

CoT 组件实现了思维链模式，通过在提示词后添加"Let's think step by step."来增强 LLM 的推理能力：

class CoT(State[None]):
    exit_: Callable[[str], State]
    llm: State
    prompt: str

    def next(self, signal: None = None) -> State:
        prompt = self.prompt + "Let's think step by step."
        # ... 回调处理

资料来源：cyclic_agent/cot.py:1-17

4 工作流程

4.1 状态转换机制

CyclicAgent 的工作流程遵循以下循环：

graph LR
    A[State 实例] -->|调用| B[next 方法]
    B --> C[执行业务逻辑]
    C --> D[确定下一状态]
    D --> E[返回新状态对象]
    E --> A

1. 用户创建初始状态实例
2. 将初始状态传递给 CyclicExecutor 的 start() 方法
3. 执行器在新线程中运行主循环
4. 每次循环调用当前状态的 next() 方法
5. next() 方法执行业务逻辑并返回下一个状态
6. 循环回到步骤 4

资料来源：cyclic_agent/executor.py:31-35

4.2 简单示例工作原理

以下是一个最简单的两个状态互相转换的示例：

class AskQuestion(State[None]):
    def next(self, signal: None = None) -> AnswerQuestion:
        response = co.chat(message="Ask a question")
        return AnswerQuestion(question=response.text)

class AnswerQuestion(State[None]):
    question: str
    def next(self, signal: None = None) -> AskQuestion:
        answer = co.chat(message=self.question)
        return AskQuestion()

# 启动执行器
executor = CyclicExecutor(5)  # 5秒间隔
executor.start(AskQuestion())

该示例创建了两个状态：AskQuestion 和 AnswerQuestion，它们互相转换形成无限循环。

资料来源：examples/hello_world/hello_world.py:1-34

5 内存管理

5.1 FIFO 队列

框架通过 FIFO（先进先出）队列实现内存管理，用于记录 Agent 的历史行为和决策过程。

class Fifo:
    def __init__(self):
        self.capacity = 100  # 最大容量
        self.queue = []
        self.log_file = "fifo_log.txt"

    def add(self, item):
        if len(self.queue) >= self.capacity:
            self.queue.pop(0)  # 移除最旧的记录
        self.queue.append((item, timestamp))

FIFO 队列支持：

• 容量限制：默认最大存储 100 条记录
• 持久化日志：自动将记录写入文件
• 时间戳记录：每条记录都附带时间戳
• 有序检索：按时间顺序显示历史记录

资料来源：examples/bilibili_surfer/fifo.py:1-31

5.2 状态继承结构

在实际应用中，状态通常需要继承 BilibiliStateBase 等基类来获取内存和 LLM 客户端等共享资源：

class BilibiliStateBase(State[None], ABC):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    initial_prompt: str
    memory: Fifo
    co: Client
    credential: Credential

资料来源：examples/bilibili_surfer/bilibili_surfer.py:26-33

6 包结构

6.1 导出接口

框架通过 cyclic_agent/__init__.py 对外导出核心接口：

from cyclic_agent.state import State
from cyclic_agent.executor import CyclicExecutor

__all__ = ["State", "CyclicExecutor"]

用户只需导入 State 基类和 CyclicExecutor 执行器即可开始使用框架。

资料来源：cyclic_agent/__init__.py:1-6

6.2 项目目录结构

cyclic-agent/
├── cyclic_agent/           # 核心框架
│   ├── __init__.py        # 包初始化，导出公共接口
│   ├── state.py           # State 基类定义
│   ├── executor.py        # CyclicExecutor 执行器
│   ├── search.py          # Search 预置状态
│   └── cot.py             # CoT 预置状态
├── examples/              # 示例代码
│   ├── hello_world/       # 简单示例
│   └── bilibili_surfer/   # B站冲浪者完整示例
└── README.md              # 项目文档

7 安装与使用

7.1 安装方式

通过 pip 安装：

pip install cyclic-agent

资料来源：README.md:11-13

7.2 依赖要求

框架使用 Pydantic 作为数据验证层，示例代码中使用了 Cohere 作为 LLM 提供商。项目依赖包括：

资料来源：examples/hello_world/hello_world.py:1-8

8 高级特性

8.1 状态推理助手

框架提供了 _infer_state_helper() 方法，用于辅助 LLM 在多个可能的状态之间进行选择：

def _infer_state_helper(self, *args: str) -> str:
    prompt = I(
        f"""
        {self.initial_prompt}
        Here are your past actions {self.memory.prompt()}.
        Here are the next states you can go to: {", ".join(args)}
        Give the state that you want to go to. 
        1. Give one word and nothing else.
        2. Be creative and try different routes.
        """
    )
    text = self.co.chat(temperature=1, message=prompt).text
    return text

该方法通过精心设计的提示词模板，引导 LLM 选择下一个合适的状态，同时记录历史行为以保持上下文连贯性。

资料来源：examples/bilibili_surfer/bilibili_surfer.py:35-50

8.2 类型别名定义

框架大量使用 Python 的类型别名（Type Alias）来定义可达状态集合：

type BrowsingVideoReachable = Union[BrowsingVideo, ReadingComments]
type ReadingCommentsReachable = Union[BrowsingVideo, ReadingComments, PostComment]

这种设计使得状态转换的类型提示更加清晰，也便于静态分析和代码维护。

资料来源：examples/bilibili_surfer/bilibili_surfer.py:51-52

9 应用场景

9.1 B站冲浪者示例

一个完整的示例展示了如何使用框架创建一个自动浏览 B 站视频、阅读评论和发表评论的 Agent：

graph TD
    A[BrowsingVideo] -->|搜索视频| B[选择视频]
    B --> C[ReadingComments]
    C -->|查看评论| D[选择评论]
    D --> E[PostComment]
    E -->|发表评论| A
    D -->|继续浏览| A
    C -->|继续浏览| A

该示例实现了四个状态：

• BrowsingVideo：搜索和选择要观看的视频
• ReadingComments：阅读视频的热门评论
• PostComment：回复选定的评论
• 各状态之间通过 _infer_state_helper() 方法进行智能转换

资料来源：examples/bilibili_surfer/bilibili_surfer.py:55-155

10 设计优势

CyclicAgent 通过有限状态机模式为 LLM Agent 的构建提供了一种结构清晰、易于维护的解决方案，使开发者能够专注于业务逻辑的实现，而无需担心底层的状态管理和执行调度。

概述

CyclicAgent 是一个基于有限状态机（FSM）设计的 LLM 驱动自主代理框架。该框架采用状态设计模式，将每个代理抽象为一个状态机，通过状态之间的转换实现循环执行。资料来源：README.md:1-5

安装与依赖模块定义了项目的基础运行环境、核心依赖包以及可选依赖项，是用户成功部署和使用该框架的前提条件。

环境要求

Python 版本

核心依赖

CyclicAgent 的核心功能依赖于以下 Python 包：

核心包仅包含最基础的依赖，确保框架轻量化运行。资料来源：cyclic_agent/state.py:1-6

安装方式

使用 pip 安装

CyclicAgent 可通过 Python 包管理器 pip 直接安装：

pip install cyclic-agent

该命令会自动安装所有核心依赖。资料来源：README.md:8-10

从源码安装

对于开发者或需要修改源码的用户，可通过 GitHub 仓库克隆并安装：

git clone https://github.com/xingjianll/cyclic-agent.git
cd cyclic-agent
pip install -e .

核心模块依赖

模块架构

graph TD
    A[cyclic_agent] --> B[state.py]
    A --> C[executor.py]
    A --> D[__init__.py]
    
    B --> E[pydantic.BaseModel]
    C --> F[threading]
    C --> G[time]
    
    style A fill:#e1f5fe
    style B fill:#f3e5f5
    style C fill:#fff3e0

状态基类依赖

State 是框架的核心抽象类，继承自 pydantic.BaseModel，提供了状态机的数据验证和序列化能力。资料来源：cyclic_agent/state.py:5

from pydantic import BaseModel

class State[SigT](BaseModel):
    @abstractmethod
    def next(self, signal: SigT | None = None) -> State[SigT]:
        """转换到下一个状态"""
        raise NotImplementedError

执行器依赖

CyclicExecutor 负责管理状态机的生命周期，使用标准库的线程机制实现后台运行。资料来源：cyclic_agent/executor.py:1-7

可选依赖

根据具体应用场景，用户可能需要安装以下可选依赖：

LLM 提供商

安装可选 LLM 依赖：

pip install cohere

环境变量管理

第三方集成

复杂示例（如 Bilibili 冲浪者）可能需要额外依赖：

环境变量配置

必需的环境变量

在使用 LLM 提供商时，需要配置相应的 API 密钥：

import os
from dotenv import load_dotenv

load_dotenv()  # 从 .env 文件加载环境变量

co = cohere.Client(os.environ.get("COHERE_API_KEY"))

.env 文件示例

COHERE_API_KEY=your_api_key_here

资料来源：examples/hello_world/hello_world.py:1-11

包导出结构

框架通过 __init__.py 向外暴露核心接口：资料来源：cyclic_agent/__init__.py:1-5

graph LR
    A[cyclic_agent] --> B[State]
    A --> C[CyclicExecutor]
    
    B --> D[状态基类]
    C --> E[执行器]

安装验证

安装完成后，可通过以下方式验证：

from cyclic_agent import State, CyclicExecutor

# 验证导入成功
print("CyclicAgent 安装成功")

依赖冲突处理

类型注解兼容

框架使用 Python 3.11+ 的新语法 class State[SigT]，如果使用旧版本 Python 可能需要以下导入：

from __future__ import annotations

此导入使类型注解使用字符串延迟求值，确保与旧版本兼容。资料来源：examples/hello_world/hello_world.py:1

Pydantic 版本

框架使用 Pydantic v2 的配置方式：

from pydantic import ConfigDict

class BilibiliStateBase(State[None]):
    model_config = ConfigDict(arbitrary_types_allowed=True)

如使用 Pydantic v1，需要调整配置方式。

快速开始依赖清单

完整的快速开始示例所需依赖：

常见问题

Q: 是否需要安装可选依赖？

A: 核心功能仅需安装 cyclic-agent 即可运行。可选依赖根据具体使用场景决定是否安装。

Q: 如何确认安装成功？

A: 执行 python -c "from cyclic_agent import State, CyclicExecutor" 无报错即表示安装成功。

Q: 框架支持哪些 Python 版本？

A: 推荐使用 Python 3.11+，以获得完整的类型注解支持。

概述

CyclicAgent 是一个基于有限状态机（FSM）设计的 LLM 驱动自主 AI Agent 框架。该框架采用状态设计模式，将每个 Agent 抽象为有限状态机，通过状态之间的转换实现 Agent 的循环执行能力。资料来源：README.md:1-5

核心概念

状态（State）

状态是 CyclicAgent 框架的基础单元。每个状态都继承自 State 基类，并实现 next() 方法来决定状态转换逻辑。资料来源：cyclic_agent/state.py:7-10

class State[SigT](BaseModel):
    @abstractmethod
    def next(self, signal: SigT | None = None) -> State[SigT]:
        """Transition to the next state."""
        raise NotImplementedError

循环执行器（CyclicExecutor）

CyclicExecutor 负责管理状态机的执行循环，通过独立的线程持续调用状态的 next() 方法来实现状态转换。资料来源：cyclic_agent/executor.py:8-11

class CyclicExecutor:
    def __init__(self, default_time_interval: float):
        self.running = False
        self.lock = threading.Lock()
        self.default_time_interval = default_time_interval

工作原理

CyclicAgent 的核心工作原理是将 Agent 抽象为有限状态机。在每个状态下，Agent 根据内部状态属性（如 memory、meta prompts）以及外部信号推断下一个状态，并与外部环境进行交互。所有状态都实现状态转换函数，返回另一个状态，从而实现状态的无限链接，使 Agent 具备"循环"特性。资料来源：README.md:10-15

graph TD
    A[初始状态] --> B{State.next}
    B -->|返回新状态| C[下一状态]
    C --> B
    B -->|接收外部信号| D[信号处理]
    D --> C

安装与配置

环境要求

安装方式

pip install cyclic-agent

环境变量配置

创建 .env 文件并配置必要的 API 密钥：

COHERE_API_KEY=your_cohere_api_key

快速开始

环境准备

1. 安装 cyclic-agent 包
2. 配置环境变量或使用 load_dotenv() 加载 .env 文件
3. 导入必要的模块：

from __future__ import annotations
from cyclic_agent import State, CyclicExecutor

基础示例：问答循环

以下示例创建了一个简单的问答 Agent，包含两个相互转换的状态。资料来源：examples/hello_world/hello_world.py:1-40

from __future__ import annotations
import os
import time

import cohere
from dotenv import load_dotenv

from cyclic_agent import State, CyclicExecutor

load_dotenv()
co = cohere.Client(os.environ.get("COHERE_API_KEY"))


class AskQuestion(State[None]):
    def next(self, signal: None = None) -> AnswerQuestion:
        response = co.chat(message="Ask a question", temperature=1)
        print(response.text)
        return AnswerQuestion(question=response.text)


class AnswerQuestion(State[None]):
    question: str

    def next(self, signal: None = None) -> AskQuestion:
        answer = co.chat(message=self.question)
        print(answer)
        return AskQuestion()


if __name__ == "__main__":
    initial_state = AskQuestion()
    executor = CyclicExecutor(5)
    executor.start(initial_state)
    time.sleep(20)

状态转换流程

graph LR
    A[AskQuestion] -->|next()| B[AnswerQuestion]
    B -->|next()| A
    style A fill:#e1f5fe
    style B fill:#fff3e0

进阶示例：B站冲浪 Agent

以下示例展示了更复杂的状态机实现，包含浏览视频、阅读评论、发表评论等多个状态。资料来源：examples/bilibili_surfer/bilibili_surfer.py:1-180

状态结构

状态基类设计

所有 Bilibili 相关状态继承自 BilibiliStateBase，该基类封装了通用逻辑。资料来源：examples/bilibili_surfer/bilibili_surfer.py:27-52

class BilibiliStateBase(State[None], ABC):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    initial_prompt: str
    memory: Fifo
    co: Client
    credential: Credential

    def _infer_state_helper(self, *args: str) -> str:
        prompt = I(
            f"""
            {self.initial_prompt}
            Here are your past actions {self.memory.prompt()}.
            Here are the next states you can go to: {", ".join(args)}
            Give the state that you want to go to. 
            1. Give one word and nothing else.
            2. Be creative and try different routes.
            """
        )
        text = self.co.chat(temperature=1, message=prompt).text
        return text

内存管理

使用 FIFO 队列记录 Agent 的历史行为，辅助 LLM 做决策。资料来源：examples/bilibili_surfer/fifo.py:1-30

class Fifo:
    def __init__(self):
        self.capacity = 100
        self.queue = []
        self.log_file = "fifo_log.txt"

    def add(self, item):
        if len(self.queue) >= self.capacity:
            self.queue.pop(0)
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        self.queue.append((item, timestamp))
        self.log_to_file(item, timestamp)

    def prompt(self):
        return "\n".join([f"{time} - {text}" for text, time in reversed(self.queue)])

完整状态转换图

graph TD
    A[BrowsingVideo] -->|搜索视频| B{状态选择}
    B -->|BrowsingVideo| A
    B -->|ReadingComments| C[ReadingComments]
    C -->|读取评论| D{状态选择}
    D -->|BrowsingVideo| A
    D -->|ReadingComments| C
    D -->|PostComment| E[PostComment]
    E -->|发表评论| A
    style A fill:#e1f5fe
    style C fill:#e8f5e9
    style E fill:#fff3e0

状态机类型定义

框架使用 Python 类型别名定义状态转换规则。资料来源：examples/bilibili_surfer/bilibili_surfer.py:54-55

type BrowsingVideoReachable = Union[BrowsingVideo, ReadingComments]
type ReadingCommentsReachable = Union[BrowsingVideo, ReadingComments, PostComment]
type PostCommentReachable = Union[BrowsingVideo]

执行器控制

CyclicExecutor 提供以下控制方法：

资料来源：cyclic_agent/executor.py:13-31

内置组件

思维链（CoT）

CoT 组件支持思维链推理模式。资料来源：cyclic_agent/cot.py:1-20

class CoT(State[None]):
    exit_: Callable[[str], State]
    llm: State
    prompt: str

    def next(self, signal: None = None) -> State:
        prompt = self.prompt + "Let's think step by step."
        def callback(answer: str) -> State:
            return self.exit_(answer)
        return self.llm(prompt=prompt, callback=callback)

搜索组件（Search）

Search 组件提供搜索功能抽象。资料来源：cyclic_agent/search.py:1-18

class Search(State[None]):
    query: str
    exit_: Callable[[[Annotated[str, "search result"]]], State]

    def next(self, signal: None = None) -> State:
        search_result = self.search(self.query)
        return self.exit_(search_result)

    def search(self, query: str) -> str:
        raise NotImplementedError

最佳实践

前向引用注意事项

由于状态类之间存在循环引用，必须使用 from __future__ import annotations 启用前向引用支持。资料来源：examples/hello_world/hello_world.py:3

from __future__ import annotations

模型_dump() 的使用

在状态转换时使用 Pydantic 的 model_dump() 方法传递状态数据。资料来源：examples/bilibili_surfer/bilibili_surfer.py:71-73

return BrowsingVideo(**self.model_dump(exclude={'video_bvid', 'video_title', 'video_description'}))

异步操作处理

使用 asyncio.run() 在同步状态方法中处理异步 API 调用。资料来源：examples/bilibili_surfer/bilibili_surfer.py:87-90

response = asyncio.run(comment.send_comment(
    text=f"{response} {footnote}",
    oid=video.Video(bvid=self.video_bvid).get_aid(),
    type_=CommentResourceType.VIDEO,
    credential=self.credential
))

LLM 决策引导

通过结构化的 prompt 引导 LLM 做出符合预期的状态选择决策。资料来源：examples/bilibili_surfer/bilibili_surfer.py:35-48

prompt = I(
    f"""
    {self.initial_prompt}
    Here are your past actions {self.memory.prompt()}.
    Here are the next states you can go to: {", ".join(args)}
    Give the state that you want to go to. 
    1. Give one word and nothing else.
    2. Be creative and try different routes.
    """
)

下一步

• 深入阅读 核心组件文档 了解更多内置组件
• 查看 示例项目 获取更多实践参考
• 探索 高级用法 学习自定义状态转换逻辑

概述

CyclicAgent框架的核心设计基于有限状态机（FSM）概念，通过状态设计模式将AI Agent抽象为可循环执行的状态机。State基类是整个框架的基础抽象，定义了状态转换的标准接口。

状态机的工作原理是：每个状态实现一个next()方法，该方法根据内部状态属性（如memory、meta prompts等）和外部信号推断下一个状态，并返回另一个状态对象。这种设计使得状态转换操作可以无限链式执行，从而实现Agent的"循环"特性。

核心架构

状态转换流程

graph TD
    A[State开始] --> B{执行next方法}
    B --> C[根据业务逻辑确定下一状态]
    C --> D[返回新状态对象]
    D --> E{CyclicExecutor是否继续?}
    E -->|是| B
    E -->|否| F[结束]
    
    G[外部信号] -.-> B
    H[内部状态属性] -.-> B

状态与执行器的关系

graph LR
    A[CyclicExecutor] -->|持有当前状态| B[State实例]
    B -->|调用next返回| C[下一State]
    C -->|循环| B
    D[Thread] -->|运行主循环| A

State基类规范

基类定义

class State[SigT](BaseModel):
    @abstractmethod
    def next(self, signal: SigT | None = None) -> State[SigT]:
        """Transition to the next state."""
        raise NotImplementedError

资料来源：cyclic_agent/state.py:6-10

关键特性

类型参数说明

执行器设计

CyclicExecutor类

CyclicExecutor负责在独立线程中运行状态机的主循环，通过time.sleep()控制状态转换的时间间隔。

class CyclicExecutor:
    def __init__(self, default_time_interval: float):
        self.running = False
        self.lock = threading.Lock()
        self.default_time_interval = default_time_interval
        self.killed = False
        self.thread = None

资料来源：cyclic_agent/executor.py:1-7

执行器控制方法

主循环实现

def _main_loop(self, state: State) -> None:
    while True:
        if self.killed:
            return
        if self.running:
            state = state.next()
            time.sleep(self.default_time_interval)

资料来源：cyclic_agent/executor.py:30-36

状态实现模式

简单状态示例

最简单的状态实现只需继承State[None]，当状态不需要传递数据时：

class AskQuestion(State[None]):
    def next(self, signal: None = None) -> AnswerQuestion:
        response = co.chat(message="Ask a question", temperature=1)
        return AnswerQuestion(question=response.text)

资料来源：examples/hello_world/hello_world.py:14-18

带状态属性的状态

通过Pydantic的字段定义，状态可以持有任意业务数据：

class AnswerQuestion(State[None]):
    question: str  # 状态属性：保存问题内容
    
    def next(self, signal: None = None) -> AskQuestion:
        answer = co.chat(message=self.question)
        return AskQuestion()

资料来源：examples/hello_world/hello_world.py:20-25

复杂状态基类

实际应用中，状态通常需要共享资源。框架通过继承模式实现：

class BilibiliStateBase(State[None], ABC):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    initial_prompt: str
    memory: Fifo
    co: Client
    credential: Credential

资料来源：examples/bilibili_surfer/bilibili_surfer.py:27-32

类型别名定义可到达状态

type BrowsingVideoReachable = Union[BrowsingVideo, ReadingComments]

class BrowsingVideo(BilibiliStateBase):
    @overrides
    def next(self, signal: None = None) -> BrowsingVideoReachable:
        ...
        match self._infer_state_helper('BrowsingVideo', 'ReadingComments'):
            case 'BrowsingVideo':
                return self
            case 'ReadingComments':
                return ReadingComments(...)

资料来源：examples/bilibili_surfer/bilibili_surfer.py:72-89

状态转换模式

一对一转换

最基础的转换模式：每个状态明确返回下一个状态类型：

graph LR
    A[AskQuestion] -->|返回AnswerQuestion| B[AnswerQuestion]
    B -->|返回AskQuestion| A

多分支转换

通过类型联合和模式匹配实现条件转换：

type ReadingCommentsReachable = Union[BrowsingVideo, ReadingComments, PostComment]

def next(self, signal: None = None) -> ReadingCommentsReachable:
    ...
    match self._infer_state_helper('BrowsingVideo', 'ReadingComments', 'PostComment'):
        case 'BrowsingVideo':
            return BrowsingVideo(...)
        case 'ReadingComments':
            return self
        case 'PostComment':
            return PostComment(...)

资料来源：examples/bilibili_surfer/bilibili_surfer.py:116-129

自引用转换

状态可以返回自身保持不变：

case 'BrowsingVideo':
    return self  # 保持当前状态

资料来源：examples/bilibili_surfer/bilibili_surfer.py:118

内置状态组件

Search状态

框架提供了可扩展的Search状态基类：

class Search(State[None]):
    query: str
    exit_: Callable[[[Annotated[str, "search result"]]], State]
    
    def next(self, signal: None = None) -> State:
        search_result = self.search(self.query)
        return self.exit_(search_result)
    
    def search(self, query: str) -> str:
        raise NotImplementedError

资料来源：cyclic_agent/search.py:1-12

Chain of Thought状态

class CoT(State[None]):
    exit_: Callable[[str], State]
    llm: State
    prompt: str
    
    def next(self, signal: None = None) -> State:
        prompt = self.prompt + "Let's think step by step."
        def callback(answer: str) -> State:
            return self.exit_(answer)
        return self.llm(prompt=prompt, callback=callback)

资料来源：cyclic_agent/cot.py:1-16

包导出结构

from cyclic_agent import State, CyclicExecutor

__all__ = ["State", "CyclicExecutor"]

资料来源：cyclic_agent/__init__.py:1-6

使用流程

graph TD
    A[定义State子类] --> B[实现next方法]
    B --> C[创建初始状态实例]
    C --> D[创建CyclicExecutor]
    D --> E[调用start方法]
    E --> F[线程运行主循环]
    F --> G[状态无限转换]
    G --> H[外部调用kill终止]

设计优势

最佳实践

1. 前向引用声明

使用状态类型时需要前向引用：

from __future__ import annotations

class AskQuestion(State[None]):
    def next(self, signal: None = None) -> AnswerQuestion:
        ...

资料来源：examples/hello_world/hello_world.py:1-2

2. 使用`@overrides`装饰器

明确标注方法重写：

from overrides import overrides

class BrowsingVideo(BilibiliStateBase):
    @overrides
    def next(self, signal: None = None) -> BrowsingVideoReachable:
        ...

资料来源：examples/bilibili_surfer/bilibili_surfer.py:69-72

3. 状态继承层次

复杂应用建议使用多层继承：

graph TD
    A[State基类] --> B[业务状态基类]
    B --> C[BrowsingVideo]
    B --> D[ReadingComments]
    B --> E[PostComment]

4. 内存管理

使用model_dump()传递状态属性：

return BrowsingVideo(**self.model_dump(exclude={'video_bvid', 'video_title', 'video_description'}))

资料来源：examples/bilibili_surfer/bilibili_surfer.py:83-85

概述

CyclicExecutor是CyclicAgent框架的核心执行引擎，负责驱动基于有限状态机（FSM）的自主代理运行。该执行器通过多线程机制和状态转换循环，使AI代理能够在各个状态之间持续流转，从而实现完全自主的AI行为。

CyclicExecutor的设计遵循状态设计模式，将每个代理抽象为一个有限状态机，其中每个状态都包含一个next()方法，该方法根据内部状态属性（如内存、元提示等）以及外部信号推断下一个状态，并返回另一个状态对象。这种设计使得状态转换操作可以无限链接，从而形成"循环"的代理行为。

资料来源：cyclic_agent/executor.py:1-44

核心架构

组件关系

CyclicExecutor与State基类之间的关系构成了框架的核心架构。State类作为所有状态节点的抽象基类，定义了状态转换的接口规范；CyclicExecutor则负责按照设定的节奏驱动状态机持续运行。

graph TD
    A[CyclicExecutor] -->|管理| B[State 状态节点]
    B -->|返回| B
    A -->|启动线程| C[_main_loop 主循环]
    C -->|循环调用| D[state.next]
    D -->|状态转换| B

类结构

资料来源：cyclic_agent/executor.py:1-44，cyclic_agent/state.py:1-20

CyclicExecutor类详解

类定义与初始化

class CyclicExecutor:
    def __init__(self, default_time_interval: float):
        self.running = False
        self.lock = threading.Lock()
        self.default_time_interval = default_time_interval
        self.killed = False
        self.thread = None

CyclicExecutor在初始化时接受一个default_time_interval参数，该参数指定状态转换之间的默认时间间隔（秒）。初始化后，执行器处于停止状态，需要通过start()方法启动。

资料来源：cyclic_agent/executor.py:8-14

生命周期控制方法

CyclicExecutor提供了完整的生命周期控制接口，支持启动、暂停、恢复和终止操作：

def start(self, initial_state: State) -> None:
    if not self.running:
        self.running = True
        self.thread = threading.Thread(target=self._main_loop, args=(initial_state,))
        self.thread.start()

start()方法检查执行器是否已在运行，若否则创建新的后台线程并启动主循环。初始状态作为参数传入，该状态将成为状态机的第一个活跃节点。

def kill(self):
    with self.lock:
        self.killed = True
        self.running = False
    if self.thread:
        self.thread.join()

kill()方法通过设置killed标志并调用线程的join()方法确保线程完全终止，实现执行器的优雅关闭。

资料来源：cyclic_agent/executor.py:16-39

主循环机制

主循环是CyclicExecutor的核心，它在独立线程中持续运行，不断调用当前状态的next()方法并接收返回的下一个状态：

def _main_loop(self, state: State) -> None:
    while True:
        if self.killed:
            return

        if self.running:
            state = state.next()
            time.sleep(self.default_time_interval)

主循环的工作流程如下：首先检查killed标志，若为真则直接返回退出循环；然后检查running标志，在执行器处于运行状态时，调用当前状态的next()方法获取下一个状态对象，并用返回值更新当前状态引用；最后按照default_time_interval设定的时间间隔休眠。

graph TD
    A[开始循环] --> B{killed == True?}
    B -->|是| Z[退出循环]
    B -->|否| C{running == True?}
    C -->|否| A
    C -->|是| D[调用 state.next]
    D --> E[获取下一状态]
    E --> F[更新 state 引用]
    F --> G[休眠 default_time_interval]
    G --> A

资料来源：cyclic_agent/executor.py:41-48

State基类

State是所有状态节点的抽象基类，基于Pydantic的BaseModel实现，支持类型提示和自动验证：

class State[SigT](BaseModel):
    @abstractmethod
    def next(self, signal: SigT | None = None) -> State[SigT]:
        """Transition to the next state."""
        raise NotImplementedError

State类使用Python 3.12引入的泛型参数SigT来指定信号的泛型类型。next()方法是抽象方法，要求所有子类必须实现，用于定义状态转换逻辑并返回下一个状态对象。

状态转换的核心约定是：每个状态的next()方法返回另一个State对象，这个返回的状态将成为下一个循环迭代的当前状态。这种设计使得状态转换形成了一个可以无限持续的循环。

资料来源：cyclic_agent/state.py:1-20

使用模式与示例

基础问答模式

最简单的应用场景是两个状态之间来回切换，实现自动问答功能：

class AskQuestion(State[None]):
    def next(self, signal: None = None) -> AnswerQuestion:
        response = co.chat(message="Ask a question", temperature=1)
        print(response.text)
        return AnswerQuestion(question=response.text)

class AnswerQuestion(State[None]):
    question: str

    def next(self, signal: None = None) -> AskQuestion:
        answer = co.chat(message=self.question)
        print(answer)
        return AskQuestion()

if __name__ == "__main__":
    initial_state = AskQuestion()
    executor = CyclicExecutor(5)
    executor.start(initial_state)

该示例创建了两个状态：AskQuestion向大模型提问，AnswerQuestion接收问题并回答。两个状态相互返回对方，形成无限循环的执行流。执行器设置5秒间隔，控制状态转换的频率。

资料来源：examples/hello_world/hello_world.py:1-38

复杂多状态模式

Bilibili冲浪者示例展示了更复杂的应用，包含浏览视频、阅读评论和发表评论三个主要状态：

class BrowsingVideo(BilibiliStateBase):
    @overrides
    def next(self, signal: None = None) -> BrowsingVideoReachable:
        # 搜索视频、选择视频
        # 根据LLM决策返回下一状态
        match self._infer_state_helper('BrowsingVideo', 'ReadingComments'):
            case 'BrowsingVideo':
                return self
            case 'ReadingComments':
                return ReadingComments(**self.model_dump(), ...)

class ReadingComments(BilibiliStateBase):
    @overrides
    def next(self, signal: None = None) -> ReadingCommentsReachable:
        # 获取评论、选择评论
        match self._infer_state_helper('BrowsingVideo', 'ReadingComments', 'PostComment'):
            case 'BrowsingVideo':
                return BrowsingVideo(...)
            case 'ReadingComments':
                return self
            case 'PostComment':
                return PostComment(...)

class PostComment(BilibiliStateBase):
    @overrides
    def next(self, signal: None = None) -> PostCommentReachable:
        # 发表评论、发布动态
        return BrowsingVideo(...)

状态之间通过类型别名（Type Alias）定义可达关系，增强了类型安全性和代码可读性：

type BrowsingVideoReachable = Union[BrowsingVideo, ReadingComments]
type ReadingCommentsReachable = Union[BrowsingVideo, ReadingComments, PostComment]
type PostCommentReachable = Union[BrowsingVideo]

资料来源：examples/bilibili_surfer/bilibili_surfer.py:1-180

状态转换流程图

graph LR
    A[BrowsingVideo<br/>浏览视频] -->|阅读评论| B[ReadingComments<br/>阅读评论]
    B -->|继续浏览| A
    B -->|发表评论| C[PostComment<br/>发表评论]
    C -->|完成| A
    A -->|继续浏览| A

内存与记忆机制

复杂代理通常需要维护内存以追踪历史行为。示例中的Fifo类实现了先入先出队列：

class Fifo:
    def __init__(self):
        self.capacity = 100
        self.queue = []
        self.log_file = "fifo_log.txt"

    def add(self, item):
        if len(self.queue) >= self.capacity:
            self.queue.pop(0)
        self.queue.append((item, datetime.now()))

    def prompt(self):
        return "\n".join([f"{time} - {text}" for text, time in reversed(self.queue)])

状态在转换时将行为记录到内存中，LLM在决策时会参考这些历史记录来生成更智能的下一状态选择。

资料来源：examples/bilibili_surfer/fifo.py:1-30

线程安全机制

CyclicExecutor使用threading.Lock保护共享状态，确保多线程环境下的安全访问：

class CyclicExecutor:
    def __init__(self, default_time_interval: float):
        self.running = False
        self.lock = threading.Lock()
        # ...

    def pause(self):
        with self.lock:
            self.running = False

    def resume(self):
        with self.lock:
            self.running = True

with self.lock语句确保在修改running标志时获得锁，防止竞态条件。由于Python的GIL特性，这种粗粒度的锁策略在大多数场景下已经足够安全。

资料来源：cyclic_agent/executor.py:8-34

模块导出

cyclic_agent包的公共接口通过__init__.py导出：

from cyclic_agent.state import State
from cyclic_agent.executor import CyclicExecutor

__all__ = ["State", "CyclicExecutor"]

使用者只需导入这两个核心组件即可开始构建自主代理应用。

资料来源：cyclic_agent/__init__.py:1-6

最佳实践

初始化配置

状态设计原则

1. 单一职责：每个状态应专注于特定功能领域
2. 幂等性：状态的next()方法应能安全重复调用
3. 状态保持：使用Pydantic的model_dump()传递上下文数据
4. 类型约束：通过Type Alias明确状态的可达转换

资源管理

• 长时间运行的应用应设置外部超时机制
• 使用kill()方法确保程序能正常退出
• 配合外部信号（如SIGTERM）实现优雅关闭

概述

CyclicAgent 框架的核心设计理念是将智能代理（Agent）抽象为有限状态机（Finite State Machine, FSM）。这一设计采用了状态模式（State Design Pattern），使代理能够根据内部状态属性（如记忆、元提示等）以及外部信号，在不同状态之间进行推断和转换。

在 CyclicAgent 中，所有状态都实现了状态转换函数，该函数返回另一个状态。这种设计允许状态转换操作无限进行，从而使代理形成"循环"执行模式。资料来源：README.md

核心组件

State 基类

State 是整个框架的抽象基类，定义在 cyclic_agent/state.py 中。它使用了 Python 的泛型语法来表示状态可接收的信号类型。

class State[SigT](BaseModel):
    @abstractmethod
    def next(self, signal: SigT | None = None) -> State[SigT]:
        """Transition to the next state."""
        raise NotImplementedError

State 类继承自 Pydantic 的 BaseModel，这意味着所有具体状态都自动具备以下能力：

• 自动数据验证
• JSON 序列化/反序列化
• 类型提示支持

资料来源：cyclic_agent/state.py:8-13

CyclicExecutor 执行器

CyclicExecutor 负责运行状态机，通过独立的线程持续执行状态转换循环。

class CyclicExecutor:
    def __init__(self, default_time_interval: float):
        self.running = False
        self.lock = threading.Lock()
        self.default_time_interval = default_time_interval
        self.killed = False
        self.thread = None

执行器使用线程安全的方式管理状态转换：

def _main_loop(self, state: State) -> None:
    while True:
        if self.killed:
            return
        if self.running:
            state = state.next()
            time.sleep(self.default_time_interval)

资料来源：cyclic_agent/executor.py:1-36

架构图

graph TD
    A[初始状态 StartState] -->|next()| B[状态 B]
    B -->|next()| C[状态 C]
    C -->|next()| D[...]
    D -->|next()| E[状态 N]
    E -->|next()| A
    
    F[CyclicExecutor] -->|管理| A
    F -->|管理| B
    F -->|管理| C
    
    G[Signal] -.->|可选输入| A
    G -.->|可选输入| B
    G -.->|可选输入| C
    
    H[Memory Context] -.->|状态上下文| A
    H -.->|状态上下文| B

状态转换机制

基础转换模式

最简单的状态转换是两个状态之间的循环切换，如 hello_world 示例所示：

class AskQuestion(State[None]):
    def next(self, signal: None = None) -> AnswerQuestion:
        response = co.chat(message="Ask a question", temperature=1)
        return AnswerQuestion(question=response.text)


class AnswerQuestion(State[None]):
    question: str

    def next(self, signal: None = None) -> AskQuestion:
        answer = co.chat(message=self.question)
        return AskQuestion()

每个状态通过 next() 方法返回下一个状态，形成无限循环。资料来源：examples/hello_world/hello_world.py:10-27

类型联合定义可达状态

CyclicAgent 使用 Python 的类型别名（Type Alias）和 type 声明来明确每个状态可以转换到哪些其他状态：

type BrowsingVideoReachable = Union[BrowsingVideo, ReadingComments]

class BrowsingVideo(BilibiliStateBase):
    @overrides
    def next(self, signal: None = None) -> BrowsingVideoReachable:
        # ... 逻辑处理
        match self._infer_state_helper('BrowsingVideo', 'ReadingComments'):
            case 'BrowsingVideo':
                return self
            case 'ReadingComments':
                return ReadingComments(**self.model_dump(), ...)

这种设计提供了：

• 类型安全：静态检查状态转换的合法性
• IDE 支持：自动补全和导航
• 文档化：明确每个状态的转换路径

资料来源：examples/bilibili_surfer/bilibili_surfer.py:44-59

使用 match-case 进行状态推断

在实际应用中，状态转换往往需要根据 LLM 的响应来决定。CyclicAgent 使用 match-case 语法匹配 LLM 返回的状态名称：

match self._infer_state_helper('BrowsingVideo', 'ReadingComments', 'PostComment'):
    case 'BrowsingVideo':
        return BrowsingVideo(**self.model_dump(exclude={...}))
    case 'ReadingComments':
        return self
    case 'PostComment':
        return PostComment(**self.model_dump(), reply_to=cmt['content']['message'], ...)

资料来源：examples/bilibili_surfer/bilibili_surfer.py:93-100

状态上下文与记忆

BilibiliStateBase 基础状态类

示例项目中的 BilibiliStateBase 展示了如何为状态添加共享上下文：

class BilibiliStateBase(State[None], ABC):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    initial_prompt: str
    memory: Fifo
    co: Client
    credential: Credential

资料来源：examples/bilibili_surfer/bilibili_surfer.py:28-35

FIFO 记忆组件

Fifo 类实现了有限容量的先进先出队列，用于存储代理的历史行为：

class Fifo:
    def __init__(self):
        self.capacity = 100
        self.queue = []
        self.log_file = "fifo_log.txt"

    def add(self, item):
        if len(self.queue) >= self.capacity:
            self.queue.pop(0)  # 移除最旧的记录
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        self.queue.append((item, timestamp))
        self.log_to_file(item, timestamp)

    def prompt(self):
        return "\n".join([f"{time} - {text}" for text, time in reversed(self.queue)])

资料来源：examples/bilibili_surfer/fifo.py:1-34

完整工作流程图

sequenceDiagram
    participant User as 用户
    participant Executor as CyclicExecutor
    participant State as 当前状态
    participant LLM as LLM API
    participant Memory as 记忆组件
    participant Env as 外部环境

    User->>Executor: start(initial_state)
    loop 持续执行
        Executor->>State: next(signal?)
        State->>Memory: prompt() 获取历史
        State->>State: 构建决策提示
        State->>LLM: chat(prompt)
        LLM-->>State: 返回决策结果
        State->>State: match-case 匹配结果
        State->>State: model_dump() 传递上下文
        State->>Env: 执行动作（如发帖、评论）
        State-->>Executor: 返回下一状态
        Executor->>Executor: sleep(interval)
    end

导出接口

CyclicAgent 包仅导出两个核心类供用户使用：

from cyclic_agent import State, CyclicExecutor

资料来源：cyclic_agent/__init__.py:1-6

使用示例

最小示例：问答循环

from cyclic_agent import State, CyclicExecutor

class AskQuestion(State[None]):
    def next(self, signal=None) -> AnswerQuestion:
        response = co.chat(message="Ask a question")
        return AnswerQuestion(question=response.text)

class AnswerQuestion(State[None]):
    question: str
    def next(self, signal=None) -> AskQuestion:
        answer = co.chat(message=self.question)
        return AskQuestion()

# 启动执行器
executor = CyclicExecutor(5)  # 每次转换间隔 5 秒
executor.start(AskQuestion())

Bilibili 冲浪代理

完整的 Bilibili 冲浪代理展示了复杂的实际应用场景，包含四个状态：

资料来源：examples/bilibili_surfer/bilibili_surfer.py:44-140

设计优势

注意事项

1. 泛型类型：from __future__ import annotations 是必需的，用于支持前向引用语法
2. 线程安全：CyclicExecutor 内部使用锁保护共享状态
3. 状态不可变性：建议状态对象设计为不可变或使用 Pydantic 的深拷贝机制
4. 信号机制：当前实现中 signal 参数默认为 None，可根据需要扩展

概述

Search状态模块是CyclicAgent框架中用于封装搜索功能的核心组件。该模块通过继承State基类实现了一个专门处理搜索查询的状态类型，允许代理（Agent）在状态机中执行搜索操作并根据搜索结果进行状态转换。

Search模块采用了策略模式设计，search方法被声明为抽象方法（NotImplementedError），由子类实现具体的搜索逻辑。这种设计使得Search状态模块具有高度的灵活性和可扩展性，可以适配各种不同的搜索数据源和搜索API。 资料来源：cyclic_agent/search.py:1-15

架构设计

类结构

classDiagram
    class State~SigT~ {
        <<abstract>>
        +next(signal: SigT | None) State
    }
    
    class Search {
        +query: str
        +exit_: Callable
        +next(signal: None) State
        +search(query: str) str
    }
    
    State <|-- Search : 继承

Search类继承自State[None]基类，这意味着该状态的信号类型为None。状态在执行next()方法时不需要外部信号输入，而是直接执行内置的搜索逻辑。 资料来源：cyclic_agent/state.py:1-18cyclic_agent/search.py:5

类型定义

回调函数的Annotated注解用于标记搜索结果的语义含义，便于类型检查和代码文档化。 资料来源：cyclic_agent/search.py:6-7

核心方法

next方法

next方法是Search状态的状态转换实现。当执行器调用此方法时，会触发搜索操作并通过回调函数返回下一个状态。

graph TD
    A[执行器调用 next] --> B[执行 searchquery]
    B --> C[获取搜索结果]
    C --> D[调用 exit_ 回调]
    D --> E[返回下一状态]

方法签名如下：

def next(self, signal: None = None) -> State:
    search_result = self.search(self.query)
    return self.exit_(search_result)

该方法接收一个可选的signal参数（类型为None），这是由于继承自泛型State[None]基类的接口要求。在实际执行中，signal参数不会被使用，搜索所需的查询内容已经存储在实例的query属性中。 资料来源：cyclic_agent/search.py:9-12

search方法

search方法是抽象搜索逻辑的占位符实现，设计者意图由子类重写此方法以提供具体的搜索功能。

def search(self, query: str) -> str:
    raise NotImplementedError

该方法接收一个query字符串参数，返回搜索结果的字符串表示。在基类中直接抛出NotImplementedError异常，要求子类必须实现具体的搜索逻辑。开发者可以在子类中接入各种搜索服务，如全文搜索引擎、API调用、数据库查询等。 资料来源：cyclic_agent/search.py:14-15

使用模式

基本集成流程

在CyclicAgent框架中，Search状态通常与其他状态配合使用，形成完整的工作流。以下是在状态机中集成Search状态的典型模式：

1. 定义Search子类：继承Search类并实现search方法
2. 配置退出回调：在初始化时传入exit_回调函数，用于处理搜索结果
3. 与执行器配合：将Search状态作为初始状态或中间状态交给CyclicExecutor执行

graph LR
    A[其他状态] -->|转换| B[Search状态]
    B -->|执行搜索| C[获取结果]
    C -->|回调| D[下一状态]
    D -->|循环| A

与Fifo内存组件配合

在真实应用场景中（如B站冲浪示例），Search操作通常与Fifo内存组件配合使用。Fifo类实现了一个容量为100的先进先出队列，用于记录代理的历史行为：

class Fifo:
    def __init__(self):
        self.capacity = 100
        self.queue = []
        self.log_file = "fifo_log.txt"

每次执行搜索后，搜索关键词会被记录到内存中，帮助代理在后续决策时考虑历史上下文。 资料来源：examples/bilibili_surfer/fifo.py:1-17

实际应用示例

在B站冲浪代理中，虽然没有直接继承Search类，但展示了类似的搜索模式实现：

async def search_by_type(response,
                         search_type=search.SearchObjectType.VIDEO,
                         order_type=search.OrderUser.FANS,
                         order_sort=0):
    # 实现B站视频搜索

代理通过生成搜索关键词、调用搜索API、解析结果、选择目标视频等步骤，形成完整的搜索-决策工作流。这种模式与Search状态模块的设计理念高度一致。 资料来源：examples/bilibili_surfer/bilibili_surfer.py:1-120

与框架其他组件的关系

与State基类的关系

Search是State基类的具体实现，遵循状态模式的设计原则。每个状态必须实现next方法，该方法返回下一个要转换到的状态。Search类在next方法中嵌入了搜索执行逻辑，使得状态转换与业务逻辑紧密结合。 资料来源：cyclic_agent/state.py:5-9

与CyclicExecutor的关系

CyclicExecutor负责管理和执行状态机的循环：

class CyclicExecutor:
    def __init__(self, default_time_interval: float):
        self.running = False
        self.lock = threading.Lock()
        self.default_time_interval = default_time_interval

执行器在每个时间间隔调用当前状态的next方法获取下一个状态，从而推动状态机向前运行。当Search状态被设置为当前状态时，执行器会触发其next方法执行搜索操作。 资料来源：cyclic_agent/executor.py:1-40

模块导出

Search类作为cyclic_agent包的一部分被导出，与核心组件State和CyclicExecutor一起构成框架的基础API：

from cyclic_agent import State, CyclicExecutor

Search类本身未包含在__all__列表中，这可能意味着它被视为内部实现或需要直接从子模块导入：

from cyclic_agent.search import Search

扩展指南

实现自定义Search子类

开发者可以通过继承Search类来实现特定数据源的搜索功能：

from cyclic_agent.search import Search
from cyclic_agent import State

class WebSearch(Search):
    def search(self, query: str) -> str:
        # 实现网络搜索逻辑
        results = some_search_api.search(query)
        return str(results)
    
    def handle_results(self, results: str) -> State:
        # 处理搜索结果，决定下一状态
        return SomeDecisionState(results=results)

# 使用示例
initial_state = WebSearch(
    query="Python最佳实践",
    exit_=lambda results: WebSearch(query=f"关于{results}的详细信息")
)

集成外部搜索API

Search模块的灵活性使其可以与各种外部搜索服务集成：

设计理念

Search状态模块体现了CyclicAgent框架的几个核心设计原则：

1. 状态即行为：每个状态封装了特定的行为逻辑，搜索操作被建模为状态而非普通方法
2. 回调驱动的状态转换：通过exit_回调函数实现搜索结果与状态转换的解耦
3. 抽象与实现分离：search方法的抽象声明允许子类提供不同实现，保持框架核心稳定
4. 有限状态机语义：状态只能转换到exit_回调指定的下一状态，维持状态机的可预测性

这种设计使得代理系统既能执行复杂的搜索任务，又能保持清晰的状态流转逻辑，便于调试和扩展。

概述

CoT（Chain of Thought，思维链）状态模块是CyclicAgent框架中的一个核心组件，专门用于实现链式思维推理能力。该模块继承自基础State类，通过在提示词后附加"Let's think step by step."指令，引导大语言模型（LLM）进行逐步推理，并将推理结果通过回调机制传递给下一个状态。

CoT状态模块的设计遵循有限状态机（FSM）模式，允许在复杂的代理（Agent）工作流中无缝集成思维链推理能力。资料来源：cyclic_agent/cot.py:1-18

核心架构

类定义与继承关系

State[SigT] (抽象基类)
    └── CoT (思维链状态)

CoT类继承自State[None]，这意味着它是一个无信号类型的状态类。资料来源：cyclic_agent/cot.py:5

组件构成

资料来源：cyclic_agent/cot.py:6-8

工作原理

思维链增强机制

当CoT状态的next()方法被调用时，会自动对原始提示词进行思维链增强处理。具体流程如下：

def next(self, signal: None = None) -> State:
    prompt = self.prompt + "Let's think step by step."
    
    def callback(answer: str) -> State:
        return self.exit_(answer)
    
    return self.llm(prompt=prompt, callback=callback)

1. 提示词增强：将"Let's think step by step."追加到原始prompt末尾
2. 回调闭包：创建一个内部callback函数，捕获exit_回调
3. 状态传递：将增强后的提示词和回调传递给底层LLM状态

资料来源：cyclic_agent/cot.py:10-17

状态转换流程

graph TD
    A[CoT.next 调用] --> B[拼接思维链提示词]
    B --> C[调用 LLM 处理]
    C --> D[获取模型输出]
    D --> E[触发 callback]
    E --> F[执行 exit_ 回调]
    F --> G[返回下一个 State]

与State基类的关系

抽象基类定义

State是一个使用Pydantic的BaseModel实现的泛型抽象类，定义了状态转换的核心接口：

class State[SigT](BaseModel):
    @abstractmethod
    def next(self, signal: SigT | None = None) -> State[SigT]:
        """Transition to the next state."""
        raise NotImplementedError

资料来源：cyclic_agent/state.py:6-11

接口契约

CoT类完整实现了这一接口，确保与CyclicExecutor执行器的兼容性。

执行器集成

CyclicExecutor负责管理和调度所有状态，包括CoT状态。执行器采用独立线程运行主循环，周期性地调用状态的next()方法：

def _main_loop(self, state: State) -> None:
    while True:
        if self.killed:
            return

        if self.running:
            state = state.next()
            time.sleep(self.default_time_interval)

资料来源：cyclic_agent/executor.py:33-39

执行流程图

graph TD
    A[CyclicExecutor.start] --> B[创建工作线程]
    B --> C[_main_loop 运行]
    C --> D{检查 running?}
    D -->|否| E[等待]
    E --> D
    D -->|是| F[调用 state.next]
    F --> G[获取下一个状态]
    G --> H[休眠 default_time_interval]
    H --> C

使用模式

基础使用结构

在实际应用中，CoT状态通常与其他自定义状态结合使用。以下是基于Hello World示例的典型模式：

from cyclic_agent import State, CyclicExecutor
from cyclic_agent.cot import CoT

class MyLLM(State[None]):
    prompt: str
    callback: Callable[[str], State]
    
    def next(self, signal=None) -> State:
        # 调用LLM API获取回答
        response = llm.chat(message=self.prompt)
        return self.callback(response.text)

class MyExitState(State[None]):
    result: str
    
    def next(self, signal=None) -> State:
        print(f"最终结果: {self.result}")
        return self  # 或返回其他状态

回调机制详解

exit_回调是连接CoT状态与下游处理的关键机制。它允许：

• 动态状态路由：根据LLM输出决定下一步状态
• 上下文传递：将推理结果携带到新状态
• 条件分支：实现复杂的条件逻辑

模块导出

CoT类通过cyclic_agent.cot模块导出，可通过以下方式导入：

from cyclic_agent.cot import CoT

资料来源：cyclic_agent/cot.py:1

设计优势

与其他状态的关系

在CyclicAgent框架中，CoT状态与其他状态类型形成互补：

graph LR
    A[CoT] -->|思维链推理| B[普通State]
    C[Search] -->|搜索增强| B
    D[自定义状态] -->|领域逻辑| B
    B --> E[CyclicExecutor]

• CoT：提供链式推理能力
• Search：提供搜索增强能力
• 自定义State：处理特定业务逻辑

总结

CoT状态模块是CyclicAgent框架实现自主代理智能推理的核心组件。通过在提示词中嵌入思维链指令，该模块能够引导大语言模型进行更深入、更结构化的思考。结合状态机模式和回调机制，CoT状态为构建复杂、可扩展的AI代理系统提供了坚实的技术基础。

概述

Hello World示例是CyclicAgent框架的入门级演示项目，展示了如何使用状态机模式构建自主运行的AI代理。该示例创建了两个相互转换的状态，形成一个无限循环的问答系统。资料来源：examples/hello_world/hello_world.py:1-27

核心概念

状态设计模式

CyclicAgent将代理抽象为有限状态机（FSM），每个状态实现状态转换函数，返回另一个状态。这种设计允许状态转换操作无限进行，从而实现代理的“循环”特性。资料来源：README.md:1-20

状态基类

所有状态必须继承State抽象基类，并实现next()方法。State类使用Pydantic的BaseModel，支持类型注解和自动验证。资料来源：cyclic_agent/state.py:1-15

系统架构

组件关系图

graph TD
    A[AskQuestion 状态] -->|next()| B[AnswerQuestion 状态]
    B -->|next()| A
    C[CyclicExecutor] -->|管理执行| A
    C -->|管理执行| B

类结构

状态实现详解

AskQuestion状态

class AskQuestion(State[None]):
    def next(self, signal: None = None) -> AnswerQuestion:
        response = co.chat(message="Ask a question", temperature=1)
        print(response.text)
        return AnswerQuestion(question=response.text)

该状态负责生成问题，使用Cohere API进行对话生成，温度参数设为1以获得更具创造性的输出。状态转换到AnswerQuestion，携带生成的问题内容。资料来源：examples/hello_world/hello_world.py:17-22

AnswerQuestion状态

class AnswerQuestion(State[None]):
    question: str

    def next(self, signal: None = None) -> AskQuestion:
        answer = co.chat(message=self.question)
        print(answer)
        return AskQuestion()

该状态持有问题内容字段question，使用Cohere API获取回答后，转换回AskQuestion状态，形成循环。资料来源：examples/hello_world/hello_world.py:24-30

CyclicExecutor执行器

核心功能

CyclicExecutor类负责在独立线程中运行状态机，支持运行控制操作。资料来源：cyclic_agent/executor.py:1-42

执行流程

sequenceDiagram
    participant User
    participant Executor
    participant State
    
    User->>Executor: start(initial_state)
    loop 状态循环
        Executor->>State: state.next()
        State-->>Executor: next_state
        Executor->>Executor: sleep(default_time_interval)
    end
    User->>Executor: kill()

线程管理实现

执行器使用threading.Thread在后台线程中运行主循环，通过锁机制确保线程安全的状态控制。资料来源：cyclic_agent/executor.py:16-23

配置与环境

依赖项

环境变量

from dotenv import load_dotenv
load_dotenv()
co = cohere.Client(os.environ.get("COHERE_API_KEY"))

资料来源：examples/hello_world/hello_world.py:7-10

运行示例

代码

if __name__ == "__main__":
    initial_state = AskQuestion()
    executor = CyclicExecutor(5)
    executor.start(initial_state)
    time.sleep(20)

参数说明

进阶示例参考

除了Hello World示例外，仓库还提供了更复杂的示例：

CoT（思维链）示例

CoT类展示了如何实现思维链模式，通过在提示词后添加"Let's think step by step."引导模型分步骤推理。资料来源：cyclic_agent/cot.py:1-16

总结

Hello World示例演示了CyclicAgent框架的核心用法：

1. 状态定义：继承State基类，实现next()方法
2. 状态转换：通过next()方法返回下一个状态
3. 执行管理：使用CyclicExecutor控制状态机运行
4. 无限循环：状态间相互转换形成闭环

该框架的核心理念是将AI代理建模为有限状态机，每个状态根据内部属性和外部信号推理下一状态，实现自主、循环的代理行为。

概述

Bilibili Surfer是cyclic-agent框架的一个完整示例应用，演示了如何构建一个自主运行的AI代理，使其能够像真实用户一样浏览Bilibili视频网站、阅读评论并自动回复。该示例展示了有限状态机（FSM）设计模式在实际应用中的强大能力，将AI代理的行为抽象为多个可相互转换的状态，每个状态根据内部属性和外部信号推断下一个状态。

这个示例的核心价值在于它演示了如何将复杂的AI行为与外部API集成。通过结合Cohere大语言模型和Bilibili API，系统能够理解自然语言指令，做出合理的决策，并在视频网站上执行实际操作。整个系统运行在CyclicExecutor的调度下，按照预设的时间间隔不断循环执行状态转换，直到达到预设的迭代次数或被手动终止。

系统架构

整体架构图

graph TD
    A[CyclicExecutor] -->|启动| B[BrowsingVideo状态]
    B -->|搜索视频| C{LLM决策}
    C -->|BrowsingVideo| B
    C -->|ReadingComments| D[ReadingComments状态]
    D -->|选择评论| C
    C -->|PostComment| E[PostComment状态]
    E -->|完成回复| B
    B -->|记录| F[Fifo内存]
    D -->|记录| F
    E -->|记录| F

核心组件层次

系统采用分层架构设计，从底层到顶层依次为：

1. 核心框架层（cyclic_agent包）

• State基类：定义所有状态的抽象接口
• CyclicExecutor：状态机执行器，负责驱动状态循环
• Search、CoT等辅助组件

1. 基础服务层

• Fifo：内存管理组件，持久化记录代理行为历史
• Bilibili API集成：通过bilibili_api库访问B站服务

1. 业务逻辑层

• BilibiliStateBase：所有业务状态的基类
• BrowsingVideo、ReadingComments、PostComment：具体业务状态

1. AI决策层

• Cohere Client：大语言模型集成
• 提示词模板：指导AI做出合理的决策

状态机设计

状态定义

系统定义了三种主要状态，每种状态都继承自BilibiliStateBase，而BilibiliStateBase又继承自State[None]基类。

状态转换关系

stateDiagram-v2
    [*] --> BrowsingVideo
    BrowsingVideo --> BrowsingVideo: 返回继续浏览
    BrowsingVideo --> ReadingComments: 选择评论
    ReadingComments --> BrowsingVideo: 返回视频浏览
    ReadingComments --> ReadingComments: 继续读评论
    ReadingComments --> PostComment: 选择回复
    PostComment --> BrowsingVideo: 完成回复

类型别名定义

为了确保状态转换的类型安全，系统使用Python的类型别名（type语句）定义每个状态可到达的下一个状态：

type BrowsingVideoReachable = Union[BrowsingVideo, ReadingComments]
type ReadingCommentsReachable = Union[BrowsingVideo, ReadingComments, PostComment]
type PostCommentReachable = Union[BrowsingVideo]

这种设计确保了编译时就能发现无效的状态转换，防止运行时错误。

核心类详解

BilibiliStateBase

BilibiliStateBase是所有Bilibili相关状态的抽象基类，它封装了状态机运行所需的所有共享资源和通用逻辑。

class BilibiliStateBase(State[None], ABC):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    initial_prompt: str
    memory: Fifo
    co: Client
    credential: Credential

属性说明：

核心方法：_infer_state_helper方法使用LLM来决定下一个状态。当调用此方法时，会向LLM传递当前状态描述、历史行为记录以及可选的下一个状态列表，LLM返回的文本与预定义的选项进行模式匹配，从而决定状态转换的方向。

资料来源：examples/bilibili_surfer/bilibili_surfer.py:43-58

BrowsingVideo状态

BrowsingVideo是系统的入口状态，负责视频搜索和浏览功能。

class BrowsingVideo(BilibiliStateBase):
    @overrides
    def next(self, signal: None = None) -> BrowsingVideoReachable:
        # 1. 生成搜索关键词
        prompt = I(f"""{self.initial_prompt}...Generate a keyword phrase...""")
        response = self.co.chat(temperature=1, message=prompt).text
        
        # 2. 调用B站搜索API
        res = asyncio.run(search.search_by_type(...))
        self.memory.add(f"searched for {response}")
        
        # 3. 获取Top10视频供选择
        top_10 = [...]
        
        # 4. 让LLM选择视频
        video = res['result'][int(response)]
        self.memory.add(f"finds {video['title']}")
        
        # 5. 决定下一步
        return self._infer_state_helper('BrowsingVideo', 'ReadingComments')

该状态的工作流程包括：使用LLM生成创造性的搜索关键词避免重复；调用B站API按粉丝数排序搜索视频；将Top10视频列表展示给LLM供选择；根据_infer_state_helper的决策结果决定是继续浏览还是转向评论阅读。

ReadingComments状态

ReadingComments状态专注于视频评论的阅读和选择。当系统处于此状态时，会获取指定视频的热门评论并展示给LLM进行选择。

class ReadingComments(BilibiliStateBase):
    video_bvid: str
    video_title: str
    video_description: str

    @overrides
    def next(self, signal: None = None) -> ReadingCommentsReachable:
        c = asyncio.run(comment.get_comments(...))
        top_10 = []
        for i in range(min(5, c['page']['count'])):
            cmt = c['replies'][i]
            top_10.append(f"{i + 1} {cmt['member']['uname']}: {cmt['content']['message']}")
        
        # 选择评论后决定下一步
        match self._infer_state_helper('BrowsingVideo', 'ReadingComments', 'PostComment'):
            case 'BrowsingVideo':
                return BrowsingVideo(...)
            case 'ReadingComments':
                return self
            case 'PostComment':
                return PostComment(...)

PostComment状态

PostComment是系统的执行状态，负责生成并发送评论回复。

class PostComment(BilibiliStateBase):
    video_bvid: str
    video_title: str
    video_description: str
    reply_to: str | None
    reply_to_oid: int | None

    @overrides
    def next(self, signal: None = None) -> PostCommentReachable:
        if self.reply_to:
            prompt = I(f"""You are replying to a comment: {self.reply_to}...""")
            response = self.co.chat(temperature=1, message=prompt).text
            footnote = f"\nI am a bot, contact {os.getenv('name')}..."
            
            # 发送评论
            asyncio.run(comment.send_comment(...))
            self.memory.add(f"commented {response} to {self.video_title}")
            
            # 发布动态
            asyncio.run(dynamic.send_dynamic(...))
        
        return BrowsingVideo(...)

该状态的亮点在于：生成回复后会自动附加免责声明；评论发送成功后会在个人动态中同步发布；无论是否需要回复，都会返回BrowsingVideo状态继续新的浏览循环。

内存管理系统

Fifo类设计

Fifo（First-In-First-Out）类是一个轻量级的内存管理组件，用于记录AI代理的所有历史行为。

class Fifo:
    def __init__(self):
        self.capacity = 100
        self.queue = []
        self.log_file = "fifo_log.txt"
        if not os.path.exists(self.log_file):
            with open(self.log_file, 'w') as file:
                file.write("")

主要特性：

核心方法：

def add(self, item):
    if not isinstance(item, str):
        raise ValueError("Only strings can be added to the FIFO")
    if len(self.queue) >= self.capacity:
        self.queue.pop(0)  # 移除最旧的记录
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    self.queue.append((item, timestamp))
    self.log_to_file(item, timestamp)

def prompt(self):
    return "\n".join([f"{time} - {text}" for text, time in reversed(self.queue)])

内存数据流

graph LR
    A[状态执行] -->|add| B[Fifo队列]
    B -->|prompt| C[LLM提示词]
    C -->|决策| D[下一状态]
    B -->|log_to_file| E[fifo_log.txt]

内存系统在状态转换中扮演着至关重要的角色。每个状态的next方法都会调用memory.add()记录当前行为，而_infer_state_helper方法会调用memory.prompt()获取历史记录并包含在LLM提示词中，使AI能够基于过去的决策历史做出更加连贯和智能的选择。

执行器机制

CyclicExecutor工作原理

CyclicExecutor是驱动整个状态机运转的核心执行器，采用多线程架构实现状态循环。

class CyclicExecutor:
    def __init__(self, default_time_interval: float):
        self.running = False
        self.lock = threading.Lock()
        self.default_time_interval = default_time_interval
        self.killed = False
        self.thread = None

线程安全设计：执行器使用threading.Lock来保护running和killed状态标志，确保在多线程环境下状态切换的安全性。pause()和resume()方法允许在运行时暂停和恢复执行，而kill()方法则可以干净地终止整个循环。

主循环逻辑

def _main_loop(self, state: State) -> None:
    while True:
        if self.killed:
            return
        
        if self.running:
            state = state.next()
            time.sleep(self.default_time_interval)

主循环采用经典的"检查-执行-休眠"模式：每次迭代检查是否被终止或暂停；如果正常运行则调用当前状态的next()方法获取下一个状态；然后根据预设的时间间隔进行休眠，控制状态转换的频率。

示例启动流程

if __name__ == "__main__":
    initial_prompt = "You are a dude browsing Bilibili..."
    
    initial_state = BrowsingVideo(
        memory=Fifo(),
        initial_prompt=initial_prompt,
        co=Client(os.environ.get("COHERE_API_KEY")),
        credential=Credential(
            sessdata=SESSDATA,
            bili_jct=BILI_JCT,
            buvid3=BUVID3
        )
    )
    executor = CyclicExecutor(5)
    executor.start(initial_state)
    time.sleep(1000)

启动时需要准备：FIFO内存实例供状态共享；定义AI角色的初始提示词；配置Bilibili API凭证；创建执行器并传入初始状态。执行器会在后台线程中运行，主线程可以继续执行其他操作或使用sleep控制运行时间。

环境配置

必需的环境变量

系统运行需要配置以下环境变量，通常通过.env文件管理：

依赖库

from bilibili_api import search, Credential, comment, video, dynamic
from bilibili_api.comment import OrderType, CommentResourceType
from bilibili_api.dynamic import BuildDynamic
from cohere import Client
from overrides import overrides
from pydantic import ConfigDict

主要依赖包括：bilibili_api提供B站API封装；cohere用于LLM交互；pydantic处理数据模型验证；overrides确保方法正确重写父类接口。

设计模式应用

状态设计模式

整个系统建立在GoF的状态设计模式之上。每个具体状态类（如BrowsingVideo）都封装了特定状态下的行为和状态转换逻辑，状态之间通过next()方法返回下一个状态对象来实现无缝切换。这种设计避免了大型条件语句，使新增状态类型变得简单——只需定义新的状态类并实现next()方法即可。

工厂模式暗示

虽然代码中没有显式的工厂类，但next()方法实际上充当了状态工厂的角色。每个状态根据当前上下文"生产"出下一个状态对象，决定系统的后续走向。这种隐式工厂模式与状态模式紧密结合，形成了灵活的状态转换机制。

策略模式暗示

_infer_state_helper方法可以被视为策略模式的应用。不同的状态可以根据需要传入不同的可选状态列表，LLM的决策策略可以被替换为其他实现（如基于规则的决策），而不影响状态类的整体结构。

扩展与定制

添加新状态

要添加新的业务状态，需要：

1. 创建新的状态类继承BilibiliStateBase
2. 实现next()方法定义状态行为
3. 定义可达状态类型别名
4. 在相关状态中添加对该新状态的转换选项

替换AI模型

当前使用Cohere作为AI后端，如需替换为其他LLM（如OpenAI GPT），只需：

1. 修改导入语句
2. 替换Client实例化方式
3. 调整API调用参数（确保参数语义兼容）

调整运行参数

总结

Bilibili Surfer示例完整展示了cyclic-agent框架在实际应用中的使用方法。通过有限状态机将复杂的AI行为分解为清晰的状态单元，结合大语言模型实现智能决策，再配合外部API执行实际操作，整个系统能够自主、连续地在Bilibili网站上执行浏览、评论和回复等任务。这个示例不仅是一个可运行的演示程序，更是一个展示AI Agent设计最佳实践的教学案例，为开发者构建自己的AI代理应用提供了完整的参考模板。

Pitfall Log / 踩坑日志

项目：xingjianll/cyclic-agent

摘要：发现 6 个潜在踩坑项，其中 0 个为 high/blocking；最高优先级：能力坑 - 能力判断依赖假设。

1. 能力坑 · 能力判断依赖假设

• 严重度：medium
• 证据强度：source_linked
• 发现：README/documentation is current enough for a first validation pass.
• 对用户的影响：假设不成立时，用户拿不到承诺的能力。
• 建议检查：将假设转成下游验证清单。
• 防护动作：假设必须转成验证项；没有验证结果前不能写成事实。
• 证据：capability.assumptions | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | README/documentation is current enough for a first validation pass.

2. 维护坑 · 维护活跃度未知

• 严重度：medium
• 证据强度：source_linked
• 发现：未记录 last_activity_observed。
• 对用户的影响：新项目、停更项目和活跃项目会被混在一起，推荐信任度下降。
• 建议检查：补 GitHub 最近 commit、release、issue/PR 响应信号。
• 防护动作：维护活跃度未知时，推荐强度不能标为高信任。
• 证据：evidence.maintainer_signals | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | last_activity_observed missing

3. 安全/权限坑 · 下游验证发现风险项

• 严重度：medium
• 证据强度：source_linked
• 发现：no_demo
• 对用户的影响：下游已经要求复核，不能在页面中弱化。
• 建议检查：进入安全/权限治理复核队列。
• 防护动作：下游风险存在时必须保持 review/recommendation 降级。
• 证据：downstream_validation.risk_items | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | no_demo; severity=medium

4. 安全/权限坑 · 存在评分风险

• 严重度：medium
• 证据强度：source_linked
• 发现：no_demo
• 对用户的影响：风险会影响是否适合普通用户安装。
• 建议检查：把风险写入边界卡，并确认是否需要人工复核。
• 防护动作：评分风险必须进入边界卡，不能只作为内部分数。
• 证据：risks.scoring_risks | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | no_demo; severity=medium

5. 维护坑 · issue/PR 响应质量未知

• 严重度：low
• 证据强度：source_linked
• 发现：issue_or_pr_quality=unknown。
• 对用户的影响：用户无法判断遇到问题后是否有人维护。
• 建议检查：抽样最近 issue/PR，判断是否长期无人处理。
• 防护动作：issue/PR 响应未知时，必须提示维护风险。
• 证据：evidence.maintainer_signals | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | issue_or_pr_quality=unknown

6. 维护坑 · 发布节奏不明确

• 严重度：low
• 证据强度：source_linked
• 发现：release_recency=unknown。
• 对用户的影响：安装命令和文档可能落后于代码，用户踩坑概率升高。
• 建议检查：确认最近 release/tag 和 README 安装命令是否一致。
• 防护动作：发布节奏未知或过期时，安装说明必须标注可能漂移。
• 证据：evidence.maintainer_signals | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | release_recency=unknown