# https://github.com/xingjianll/cyclic-agent 项目说明书 生成时间：2026-05-20 05:59:53 UTC ## 目录 - [项目介绍](#page-1) - [安装与依赖](#page-2) - [快速入门指南](#page-3) - [State基类设计](#page-4) - [CyclicExecutor执行器](#page-5) - [有限状态机模式](#page-6) - [Search状态模块](#page-7) - [CoT状态模块](#page-8) - [Hello World示例](#page-9) - [Bilibili Surfer示例](#page-10) ## 项目介绍 ### 相关页面 相关主题：[安装与依赖](#page-2), [State基类设计](#page-4)
相关源码文件 以下源码文件用于生成本页说明： - [README.md](https://github.com/xingjianll/cyclic-agent/blob/main/README.md) - [cyclic_agent/__init__.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/__init__.py) - [cyclic_agent/state.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/state.py) - [cyclic_agent/executor.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py) - [cyclic_agent/search.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/search.py) - [cyclic_agent/cot.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/cot.py) - [examples/hello_world/hello_world.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/hello_world/hello_world.py) - [examples/bilibili_surfer/bilibili_surfer.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py) - [examples/bilibili_surfer/fifo.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/fifo.py)
# 项目介绍 ## 1 项目概述 CyclicAgent 是一个专为创建 LLM 驱动的完全自主 AI Agent 而设计的框架。该框架的核心创新在于将 Agent 抽象为有限状态机（FSM），采用状态设计模式实现。在每个状态中，Agent 根据内部状态属性（如内存、元提示等）以及外部信号来推断下一个状态，并与外部环境进行交互。 资料来源：[README.md:1-10]() ## 2 核心设计理念 ### 2.1 有限状态机架构 CyclicAgent 的核心理念基于有限状态机（FSM）设计。框架中的所有状态都实现了一个状态转换函数，该函数返回另一个状态对象。这种设计允许状态转换操作无限链接，从而使 Agent 变得“循环”。 ```mermaid 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()` 方法。 ```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 作为基类，提供了数据验证和序列化能力。每个状态都有一个泛型参数 `SigT` 表示信号的接收类型，`next()` 方法接收一个可选的信号参数并返回下一个状态。 资料来源：[cyclic_agent/state.py:1-10]() ### 3.2 CyclicExecutor 执行器 CyclicExecutor 负责执行状态机的运行循环，支持暂停、恢复和终止操作。 ```python 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 执行器控制方法 | 方法 | 功能描述 | 线程安全 | |------|---------|----------| | `start(initial_state)` | 启动执行器，传入初始状态 | 是 | | `pause()` | 暂停执行循环 | 是 | | `resume()` | 恢复执行循环 | 是 | | `kill()` | 终止执行器 | 是 | 资料来源：[cyclic_agent/executor.py:11-24]() ### 3.4 预置状态组件 框架提供了两个预置的状态组件用于常见场景： | 组件 | 文件位置 | 用途 | |------|----------|------| | Search | `cyclic_agent/search.py` | 执行搜索操作并根据结果转换状态 | | CoT | `cyclic_agent/cot.py` | 实现思维链（Chain of Thought）模式 | Search 组件允许定义搜索查询和回调函数来根据搜索结果决定后续状态： ```python 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 的推理能力： ```python 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 的工作流程遵循以下循环： ```mermaid 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 简单示例工作原理 以下是一个最简单的两个状态互相转换的示例： ```python 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 的历史行为和决策过程。 ```python 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 客户端等共享资源： ```python 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` 对外导出核心接口： ```python 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 安装： ```shell pip install cyclic-agent ``` 资料来源：[README.md:11-13]() ### 7.2 依赖要求 框架使用 Pydantic 作为数据验证层，示例代码中使用了 Cohere 作为 LLM 提供商。项目依赖包括： | 依赖包 | 用途 | |--------|------| | pydantic | 数据验证和序列化 | | cohere | LLM 接口调用 | | dotenv | 环境变量管理 | 资料来源：[examples/hello_world/hello_world.py:1-8]() ## 8 高级特性 ### 8.1 状态推理助手 框架提供了 `_infer_state_helper()` 方法，用于辅助 LLM 在多个可能的状态之间进行选择： ```python 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）来定义可达状态集合： ```python 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： ```mermaid 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 设计优势 | 优势 | 说明 | |------|------| | 简洁直观 | 状态即普通的 Python 类，易于理解和实现 | | 类型安全 | 基于 Pydantic 的数据验证和泛型支持 | | 可扩展性 | 可自由定义状态数量和转换逻辑 | | 线程安全 | 执行器使用锁保护共享状态 | | 循环执行 | 支持无限循环的任务执行 | | 记忆能力 | 内置 FIFO 队列支持历史记录管理 | CyclicAgent 通过有限状态机模式为 LLM Agent 的构建提供了一种结构清晰、易于维护的解决方案，使开发者能够专注于业务逻辑的实现，而无需担心底层的状态管理和执行调度。 --- ## 安装与依赖 ### 相关页面 相关主题：[项目介绍](#page-1), [快速入门指南](#page-3)
相关源码文件 以下源码文件用于生成本页说明： - [pyproject.toml](https://github.com/xingjianll/cyclic-agent/blob/main/pyproject.toml) - [README.md](https://github.com/xingjianll/cyclic-agent/blob/main/README.md) - [cyclic_agent/__init__.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/__init__.py) - [cyclic_agent/state.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/state.py) - [cyclic_agent/executor.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py)
# 安装与依赖 ## 概述 CyclicAgent 是一个基于有限状态机（FSM）设计的 LLM 驱动自主代理框架。该框架采用状态设计模式，将每个代理抽象为一个状态机，通过状态之间的转换实现循环执行。资料来源：[README.md:1-5]() 安装与依赖模块定义了项目的基础运行环境、核心依赖包以及可选依赖项，是用户成功部署和使用该框架的前提条件。 ## 环境要求 ### Python 版本 | 要求项 | 最低版本 | 说明 | |--------|----------|------| | Python | 3.11+ | 需要支持泛型类型参数语法 `class State[SigT]` | ### 核心依赖 CyclicAgent 的核心功能依赖于以下 Python 包： | 依赖包 | 版本要求 | 用途 | |--------|----------|------| | pydantic | ≥2.0 | 状态基类的数据模型实现 | | typing_extensions | 最新版 | 支持 Python 3.11 之前的类型注解 | 核心包仅包含最基础的依赖，确保框架轻量化运行。资料来源：[cyclic_agent/state.py:1-6]() ## 安装方式 ### 使用 pip 安装 CyclicAgent 可通过 Python 包管理器 pip 直接安装： ```shell pip install cyclic-agent ``` 该命令会自动安装所有核心依赖。资料来源：[README.md:8-10]() ### 从源码安装 对于开发者或需要修改源码的用户，可通过 GitHub 仓库克隆并安装： ```shell git clone https://github.com/xingjianll/cyclic-agent.git cd cyclic-agent pip install -e . ``` ## 核心模块依赖 ### 模块架构 ```mermaid 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]() ```python 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]() | 模块 | 来源 | 用途 | |------|------|------| | threading | 标准库 | 多线程状态循环执行 | | time | 标准库 | 执行间隔控制 | ## 可选依赖 根据具体应用场景，用户可能需要安装以下可选依赖： ### LLM 提供商 | 包名 | 说明 | 用途示例 | |------|------|----------| | cohere | Cohere API 客户端 | 实现与 Cohere LLM 的交互 | 安装可选 LLM 依赖： ```shell pip install cohere ``` ### 环境变量管理 | 包名 | 说明 | |------|------| | python-dotenv | 从 `.env` 文件加载环境变量 | ### 第三方集成 复杂示例（如 Bilibili 冲浪者）可能需要额外依赖： | 包名 | 用途 | |------|------| | bilibili-api-python | Bilibili API 封装 | | overrides | 方法重写装饰器 | ## 环境变量配置 ### 必需的环境变量 在使用 LLM 提供商时，需要配置相应的 API 密钥： ```python 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]() ```mermaid graph LR A[cyclic_agent] --> B[State] A --> C[CyclicExecutor] B --> D[状态基类] C --> E[执行器] ``` | 导出项 | 模块路径 | 说明 | |--------|----------|------| | `State` | cyclic_agent.state | 状态机抽象基类 | | `CyclicExecutor` | cyclic_agent.executor | 状态循环执行器 | ## 安装验证 安装完成后，可通过以下方式验证： ```python from cyclic_agent import State, CyclicExecutor # 验证导入成功 print("CyclicAgent 安装成功") ``` ## 依赖冲突处理 ### 类型注解兼容 框架使用 Python 3.11+ 的新语法 `class State[SigT]`，如果使用旧版本 Python 可能需要以下导入： ```python from __future__ import annotations ``` 此导入使类型注解使用字符串延迟求值，确保与旧版本兼容。资料来源：[examples/hello_world/hello_world.py:1]() ### Pydantic 版本 框架使用 Pydantic v2 的配置方式： ```python from pydantic import ConfigDict class BilibiliStateBase(State[None]): model_config = ConfigDict(arbitrary_types_allowed=True) ``` 如使用 Pydantic v1，需要调整配置方式。 ## 快速开始依赖清单 完整的快速开始示例所需依赖： | 依赖包 | 安装命令 | 用途 | |--------|----------|------| | cyclic-agent | `pip install cyclic-agent` | 核心框架 | | cohere | `pip install cohere` | LLM 调用 | | python-dotenv | `pip install python-dotenv` | 环境变量 | ## 常见问题 ### Q: 是否需要安装可选依赖？ A: 核心功能仅需安装 `cyclic-agent` 即可运行。可选依赖根据具体使用场景决定是否安装。 ### Q: 如何确认安装成功？ A: 执行 `python -c "from cyclic_agent import State, CyclicExecutor"` 无报错即表示安装成功。 ### Q: 框架支持哪些 Python 版本？ A: 推荐使用 Python 3.11+，以获得完整的类型注解支持。 --- ## 快速入门指南 ### 相关页面 相关主题：[安装与依赖](#page-2), [State基类设计](#page-4), [Bilibili Surfer示例](#page-10)
相关源码文件 以下源码文件用于生成本页说明： - [examples/hello_world/hello_world.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/hello_world/hello_world.py) - [cyclic_agent/executor.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py) - [cyclic_agent/state.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/state.py) - [cyclic_agent/cot.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/cot.py) - [cyclic_agent/search.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/search.py) - [examples/bilibili_surfer/bilibili_surfer.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py) - [examples/bilibili_surfer/fifo.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/fifo.py)
# 快速入门指南 ## 概述 CyclicAgent 是一个基于有限状态机（FSM）设计的 LLM 驱动自主 AI Agent 框架。该框架采用状态设计模式，将每个 Agent 抽象为有限状态机，通过状态之间的转换实现 Agent 的循环执行能力。资料来源：[README.md:1-5]() ## 核心概念 ### 状态（State） 状态是 CyclicAgent 框架的基础单元。每个状态都继承自 `State` 基类，并实现 `next()` 方法来决定状态转换逻辑。资料来源：[cyclic_agent/state.py:7-10]() ```python 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]() ```python 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]() ```mermaid graph TD A[初始状态] --> B{State.next} B -->|返回新状态| C[下一状态] C --> B B -->|接收外部信号| D[信号处理] D --> C ``` ## 安装与配置 ### 环境要求 | 要求项 | 说明 | |--------|------| | Python 版本 | 3.8+ | | 核心依赖 | Pydantic | | LLM 提供商 | Cohere（示例使用） | | 环境管理 | dotenv | ### 安装方式 ```shell pip install cyclic-agent ``` ### 环境变量配置 创建 `.env` 文件并配置必要的 API 密钥： ```bash COHERE_API_KEY=your_cohere_api_key ``` ## 快速开始 ### 环境准备 1. 安装 cyclic-agent 包 2. 配置环境变量或使用 `load_dotenv()` 加载 `.env` 文件 3. 导入必要的模块： ```python from __future__ import annotations from cyclic_agent import State, CyclicExecutor ``` ### 基础示例：问答循环 以下示例创建了一个简单的问答 Agent，包含两个相互转换的状态。资料来源：[examples/hello_world/hello_world.py:1-40]() ```python 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) ``` ### 状态转换流程 ```mermaid 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]() ### 状态结构 | 状态类 | 功能 | 可转换状态 | |--------|------|------------| | `BrowsingVideo` | 搜索和浏览视频 | `BrowsingVideo`, `ReadingComments` | | `ReadingComments` | 读取视频评论 | `BrowsingVideo`, `ReadingComments`, `PostComment` | | `PostComment` | 发表/回复评论 | `BrowsingVideo` | ### 状态基类设计 所有 Bilibili 相关状态继承自 `BilibiliStateBase`，该基类封装了通用逻辑。资料来源：[examples/bilibili_surfer/bilibili_surfer.py:27-52]() ```python 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]() ```python 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)]) ``` ### 完整状态转换图 ```mermaid 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]() ```python type BrowsingVideoReachable = Union[BrowsingVideo, ReadingComments] type ReadingCommentsReachable = Union[BrowsingVideo, ReadingComments, PostComment] type PostCommentReachable = Union[BrowsingVideo] ``` ## 执行器控制 `CyclicExecutor` 提供以下控制方法： | 方法 | 功能 | 线程安全 | |------|------|----------| | `start(initial_state)` | 启动状态机 | 是 | | `pause()` | 暂停执行 | 是 | | `resume()` | 恢复执行 | 是 | | `kill()` | 停止并清理 | 是 | 资料来源：[cyclic_agent/executor.py:13-31]() ## 内置组件 ### 思维链（CoT） `CoT` 组件支持思维链推理模式。资料来源：[cyclic_agent/cot.py:1-20]() ```python 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]() ```python 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]() ```python from __future__ import annotations ``` ### 模型_dump() 的使用 在状态转换时使用 Pydantic 的 `model_dump()` 方法传递状态数据。资料来源：[examples/bilibili_surfer/bilibili_surfer.py:71-73]() ```python return BrowsingVideo(**self.model_dump(exclude={'video_bvid', 'video_title', 'video_description'})) ``` ### 异步操作处理 使用 `asyncio.run()` 在同步状态方法中处理异步 API 调用。资料来源：[examples/bilibili_surfer/bilibili_surfer.py:87-90]() ```python 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]() ```python 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. """ ) ``` ## 下一步 - 深入阅读 [核心组件文档](./core-components.md) 了解更多内置组件 - 查看 [示例项目](./examples.md) 获取更多实践参考 - 探索 [高级用法](./advanced-usage.md) 学习自定义状态转换逻辑 --- ## State基类设计 ### 相关页面 相关主题：[CyclicExecutor执行器](#page-5), [有限状态机模式](#page-6)
相关源码文件 以下源码文件用于生成本页说明： - [cyclic_agent/state.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/state.py) - [cyclic_agent/executor.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py) - [cyclic_agent/__init__.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/__init__.py) - [cyclic_agent/search.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/search.py) - [cyclic_agent/cot.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/cot.py) - [examples/hello_world/hello_world.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/hello_world/hello_world.py) - [examples/bilibili_surfer/bilibili_surfer.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py)
# State基类设计 ## 概述 CyclicAgent框架的核心设计基于**有限状态机（FSM）**概念，通过状态设计模式将AI Agent抽象为可循环执行的状态机。`State`基类是整个框架的基础抽象，定义了状态转换的标准接口。 状态机的工作原理是：每个状态实现一个`next()`方法，该方法根据内部状态属性（如memory、meta prompts等）和外部信号推断下一个状态，并返回另一个状态对象。这种设计使得状态转换操作可以无限链式执行，从而实现Agent的"循环"特性。 ## 核心架构 ### 状态转换流程 ```mermaid graph TD A[State开始] --> B{执行next方法} B --> C[根据业务逻辑确定下一状态] C --> D[返回新状态对象] D --> E{CyclicExecutor是否继续?} E -->|是| B E -->|否| F[结束] G[外部信号] -.-> B H[内部状态属性] -.-> B ``` ### 状态与执行器的关系 ```mermaid graph LR A[CyclicExecutor] -->|持有当前状态| B[State实例] B -->|调用next返回| C[下一State] C -->|循环| B D[Thread] -->|运行主循环| A ``` ## State基类规范 ### 基类定义 ```python 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]() ### 关键特性 | 特性 | 说明 | |------|------| | 泛型设计 | 使用Python泛型`[SigT]`定义信号类型参数 | | Pydantic集成 | 继承`BaseModel`，支持数据验证和序列化 | | 抽象方法 | `next()`必须由子类实现 | | 信号机制 | 支持可选的外部信号参数 | ### 类型参数说明 | 参数 | 含义 | 用途 | |------|------|------| | `SigT` | Signal Type | 定义状态间传递的信号数据类型 | | `State[None]` | 无信号状态 | 状态间无需传递数据，如`AskQuestion` | | `State[str]` | 字符串信号 | 可携带文本信息进行状态转换 | ## 执行器设计 ### CyclicExecutor类 `CyclicExecutor`负责在独立线程中运行状态机的主循环，通过`time.sleep()`控制状态转换的时间间隔。 ```python 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]() ### 执行器控制方法 | 方法 | 功能 | 线程安全性 | |------|------|------------| | `start(initial_state)` | 启动执行线程 | 线程安全 | | `pause()` | 暂停状态机 | 线程安全 | | `resume()` | 恢复执行 | 线程安全 | | `kill()` | 终止执行 | 线程安全 | ### 主循环实现 ```python 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]`，当状态不需要传递数据时： ```python 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的字段定义，状态可以持有任意业务数据： ```python 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]() ### 复杂状态基类 实际应用中，状态通常需要共享资源。框架通过继承模式实现： ```python 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]() ### 类型别名定义可到达状态 ```python 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]() ## 状态转换模式 ### 一对一转换 最基础的转换模式：每个状态明确返回下一个状态类型： ```mermaid graph LR A[AskQuestion] -->|返回AnswerQuestion| B[AnswerQuestion] B -->|返回AskQuestion| A ``` ### 多分支转换 通过类型联合和模式匹配实现条件转换： ```python 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]() ### 自引用转换 状态可以返回自身保持不变： ```python case 'BrowsingVideo': return self # 保持当前状态 ``` 资料来源：[examples/bilibili_surfer/bilibili_surfer.py:118]() ## 内置状态组件 ### Search状态 框架提供了可扩展的`Search`状态基类： ```python 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状态 ```python 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]() ## 包导出结构 ```python from cyclic_agent import State, CyclicExecutor __all__ = ["State", "CyclicExecutor"] ``` 资料来源：[cyclic_agent/__init__.py:1-6]() ## 使用流程 ```mermaid graph TD A[定义State子类] --> B[实现next方法] B --> C[创建初始状态实例] C --> D[创建CyclicExecutor] D --> E[调用start方法] E --> F[线程运行主循环] F --> G[状态无限转换] G --> H[外部调用kill终止] ``` ## 设计优势 | 优势 | 说明 | |------|------| | 简洁性 | 单一抽象方法`next()`定义状态行为 | | 灵活性 | Pydantic提供强大的数据验证 | | 可扩展性 | 继承基类即可创建新状态类型 | | 类型安全 | 泛型设计支持类型检查 | | 可测试性 | 每个状态独立，易于单元测试 | ## 最佳实践 ### 1. 前向引用声明 使用状态类型时需要前向引用： ```python 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`装饰器 明确标注方法重写： ```python from overrides import overrides class BrowsingVideo(BilibiliStateBase): @overrides def next(self, signal: None = None) -> BrowsingVideoReachable: ... ``` 资料来源：[examples/bilibili_surfer/bilibili_surfer.py:69-72]() ### 3. 状态继承层次 复杂应用建议使用多层继承： ```mermaid graph TD A[State基类] --> B[业务状态基类] B --> C[BrowsingVideo] B --> D[ReadingComments] B --> E[PostComment] ``` ### 4. 内存管理 使用`model_dump()`传递状态属性： ```python return BrowsingVideo(**self.model_dump(exclude={'video_bvid', 'video_title', 'video_description'})) ``` 资料来源：[examples/bilibili_surfer/bilibili_surfer.py:83-85]() --- ## CyclicExecutor执行器 ### 相关页面 相关主题：[State基类设计](#page-4), [快速入门指南](#page-3)
相关源码文件 以下源码文件用于生成本页说明： - [cyclic_agent/executor.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py) - [cyclic_agent/state.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/state.py) - [cyclic_agent/__init__.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/__init__.py) - [examples/hello_world/hello_world.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/hello_world/hello_world.py) - [examples/bilibili_surfer/bilibili_surfer.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py) - [examples/bilibili_surfer/fifo.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/fifo.py)
# CyclicExecutor执行器 ## 概述 CyclicExecutor是CyclicAgent框架的核心执行引擎，负责驱动基于有限状态机（FSM）的自主代理运行。该执行器通过多线程机制和状态转换循环，使AI代理能够在各个状态之间持续流转，从而实现完全自主的AI行为。 CyclicExecutor的设计遵循状态设计模式，将每个代理抽象为一个有限状态机，其中每个状态都包含一个`next()`方法，该方法根据内部状态属性（如内存、元提示等）以及外部信号推断下一个状态，并返回另一个状态对象。这种设计使得状态转换操作可以无限链接，从而形成"循环"的代理行为。 资料来源：[cyclic_agent/executor.py:1-44](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py) ## 核心架构 ### 组件关系 CyclicExecutor与State基类之间的关系构成了框架的核心架构。State类作为所有状态节点的抽象基类，定义了状态转换的接口规范；CyclicExecutor则负责按照设定的节奏驱动状态机持续运行。 ```mermaid graph TD A[CyclicExecutor] -->|管理| B[State 状态节点] B -->|返回| B A -->|启动线程| C[_main_loop 主循环] C -->|循环调用| D[state.next] D -->|状态转换| B ``` ### 类结构 | 类名 | 文件位置 | 职责 | |------|----------|------| | CyclicExecutor | cyclic_agent/executor.py | 状态机执行引擎 | | State | cyclic_agent/state.py | 状态抽象基类 | | Timer | cyclic_agent/executor.py | 占位类（未实现） | 资料来源：[cyclic_agent/executor.py:1-44](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py)，[cyclic_agent/state.py:1-20](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/state.py) ## CyclicExecutor类详解 ### 类定义与初始化 ```python 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](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py) ### 生命周期控制方法 CyclicExecutor提供了完整的生命周期控制接口，支持启动、暂停、恢复和终止操作： | 方法 | 功能 | 线程安全 | |------|------|----------| | start(initial_state) | 启动执行器，创建新线程运行主循环 | 否（仅首次调用有效） | | pause() | 暂停状态机执行 | 是（使用Lock） | | resume() | 恢复状态机执行 | 是（使用Lock） | | kill() | 彻底终止执行器 | 是（使用Lock） | ```python 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()`方法检查执行器是否已在运行，若否则创建新的后台线程并启动主循环。初始状态作为参数传入，该状态将成为状态机的第一个活跃节点。 ```python 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](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py) ### 主循环机制 主循环是CyclicExecutor的核心，它在独立线程中持续运行，不断调用当前状态的`next()`方法并接收返回的下一个状态： ```python 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`设定的时间间隔休眠。 ```mermaid 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](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py) ## State基类 State是所有状态节点的抽象基类，基于Pydantic的BaseModel实现，支持类型提示和自动验证： ```python 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](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/state.py) ## 使用模式与示例 ### 基础问答模式 最简单的应用场景是两个状态之间来回切换，实现自动问答功能： ```python 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](https://github.com/xingjianll/cyclic-agent/blob/main/examples/hello_world/hello_world.py) ### 复杂多状态模式 Bilibili冲浪者示例展示了更复杂的应用，包含浏览视频、阅读评论和发表评论三个主要状态： ```python 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）定义可达关系，增强了类型安全性和代码可读性： ```python type BrowsingVideoReachable = Union[BrowsingVideo, ReadingComments] type ReadingCommentsReachable = Union[BrowsingVideo, ReadingComments, PostComment] type PostCommentReachable = Union[BrowsingVideo] ``` 资料来源：[examples/bilibili_surfer/bilibili_surfer.py:1-180](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py) ### 状态转换流程图 ```mermaid graph LR A[BrowsingVideo
浏览视频] -->|阅读评论| B[ReadingComments
阅读评论] B -->|继续浏览| A B -->|发表评论| C[PostComment
发表评论] C -->|完成| A A -->|继续浏览| A ``` ### 内存与记忆机制 复杂代理通常需要维护内存以追踪历史行为。示例中的Fifo类实现了先入先出队列： ```python 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](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/fifo.py) ## 线程安全机制 CyclicExecutor使用`threading.Lock`保护共享状态，确保多线程环境下的安全访问： ```python 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](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py) ## 模块导出 cyclic_agent包的公共接口通过`__init__.py`导出： ```python from cyclic_agent.state import State from cyclic_agent.executor import CyclicExecutor __all__ = ["State", "CyclicExecutor"] ``` 使用者只需导入这两个核心组件即可开始构建自主代理应用。 资料来源：[cyclic_agent/__init__.py:1-6](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/__init__.py) ## 最佳实践 ### 初始化配置 | 配置项 | 说明 | 推荐值 | |--------|------|--------| | default_time_interval | 状态转换间隔（秒） | 3-10秒 | | 初始状态选择 | 应选择有明确转换路径的状态 | 确保能进入循环 | ### 状态设计原则 1. **单一职责**：每个状态应专注于特定功能领域 2. **幂等性**：状态的`next()`方法应能安全重复调用 3. **状态保持**：使用Pydantic的`model_dump()`传递上下文数据 4. **类型约束**：通过Type Alias明确状态的可达转换 ### 资源管理 - 长时间运行的应用应设置外部超时机制 - 使用`kill()`方法确保程序能正常退出 - 配合外部信号（如SIGTERM）实现优雅关闭 --- ## 有限状态机模式 ### 相关页面 相关主题：[State基类设计](#page-4), [CyclicExecutor执行器](#page-5)
相关源码文件 以下源码文件用于生成本页说明： - [cyclic_agent/state.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/state.py) - [cyclic_agent/executor.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py) - [examples/bilibili_surfer/bilibili_surfer.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py) - [examples/hello_world/hello_world.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/hello_world/hello_world.py) - [cyclic_agent/__init__.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/__init__.py) - [examples/bilibili_surfer/fifo.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/fifo.py)
# 有限状态机模式 ## 概述 CyclicAgent 框架的核心设计理念是将智能代理（Agent）抽象为**有限状态机（Finite State Machine, FSM）**。这一设计采用了状态模式（State Design Pattern），使代理能够根据内部状态属性（如记忆、元提示等）以及外部信号，在不同状态之间进行推断和转换。 在 CyclicAgent 中，所有状态都实现了状态转换函数，该函数返回另一个状态。这种设计允许状态转换操作无限进行，从而使代理形成"循环"执行模式。资料来源：[README.md]() ## 核心组件 ### State 基类 `State` 是整个框架的抽象基类，定义在 `cyclic_agent/state.py` 中。它使用了 Python 的泛型语法来表示状态可接收的信号类型。 ```python class State[SigT](BaseModel): @abstractmethod def next(self, signal: SigT | None = None) -> State[SigT]: """Transition to the next state.""" raise NotImplementedError ``` | 属性/方法 | 类型 | 说明 | |-----------|------|------| | `SigT` | 泛型类型参数 | 表示状态可接收的信号类型 | | `next(signal)` | 抽象方法 | 执行状态转换，返回下一个状态 | | `model_dump()` | 继承自 BaseModel | 序列化状态对象为字典 | State 类继承自 Pydantic 的 `BaseModel`，这意味着所有具体状态都自动具备以下能力： - 自动数据验证 - JSON 序列化/反序列化 - 类型提示支持 资料来源：[cyclic_agent/state.py:8-13]() ### CyclicExecutor 执行器 `CyclicExecutor` 负责运行状态机，通过独立的线程持续执行状态转换循环。 ```python 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 ``` | 方法 | 说明 | |------|------| | `start(initial_state)` | 启动执行器，传入初始状态 | | `pause()` | 暂停状态机执行 | | `resume()` | 恢复状态机执行 | | `kill()` | 终止执行器 | | `_main_loop()` | 内部主循环，持续调用 `state.next()` | 执行器使用线程安全的方式管理状态转换： ```python 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]() ## 架构图 ```mermaid 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 示例所示： ```python 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` 声明来明确每个状态可以转换到哪些其他状态： ```python 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 返回的状态名称： ```python 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` 展示了如何为状态添加共享上下文： ```python class BilibiliStateBase(State[None], ABC): model_config = ConfigDict(arbitrary_types_allowed=True) initial_prompt: str memory: Fifo co: Client credential: Credential ``` | 上下文属性 | 类型 | 说明 | |-----------|------|------| | `initial_prompt` | str | 初始化 LLM 交互的提示词 | | `memory` | Fifo | FIFO 队列，记录历史行为 | | `co` | Client | Cohere LLM 客户端 | | `credential` | Credential | Bilibili API 认证凭证 | 资料来源：[examples/bilibili_surfer/bilibili_surfer.py:28-35]() ### FIFO 记忆组件 `Fifo` 类实现了有限容量的先进先出队列，用于存储代理的历史行为： ```python 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)]) ``` | 方法 | 说明 | |------|------| | `add(item)` | 添加新记录，自动管理容量上限 | | `prompt()` | 生成提示字符串，最新记录在前 | | `log_to_file()` | 将记录持久化到日志文件 | 资料来源：[examples/bilibili_surfer/fifo.py:1-34]() ## 完整工作流程图 ```mermaid 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 包仅导出两个核心类供用户使用： ```python from cyclic_agent import State, CyclicExecutor ``` | 导出项 | 来源文件 | 说明 | |-------|----------|------| | `State` | cyclic_agent/state.py | 状态机抽象基类 | | `CyclicExecutor` | cyclic_agent/executor.py | 状态机执行器 | 资料来源：[cyclic_agent/__init__.py:1-6]() ## 使用示例 ### 最小示例：问答循环 ```python 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 冲浪代理展示了复杂的实际应用场景，包含四个状态： | 状态 | 可达状态 | 功能 | |------|----------|------| | `BrowsingVideo` | `BrowsingVideo`, `ReadingComments` | 搜索和浏览视频 | | `ReadingComments` | `BrowsingVideo`, `ReadingComments`, `PostComment` | 阅读视频评论 | | `PostComment` | `BrowsingVideo` | 回复评论 | | 初始状态 | `BrowsingVideo` | 从视频浏览开始 | 资料来源：[examples/bilibili_surfer/bilibili_surfer.py:44-140]() ## 设计优势 | 优势 | 说明 | |------|------| | **状态内聚** | 每个状态包含自身数据和转换逻辑 | | **易于扩展** | 添加新状态只需继承 `State` 并实现 `next()` | | **类型安全** | 静态类型检查确保状态转换的合法性 | | **测试友好** | 每个状态可独立单元测试 | | **无限循环** | CyclicExecutor 支持状态机持续运行 | | **上下文传递** | 通过 Pydantic 的 `model_dump()` 传递状态属性 | ## 注意事项 1. **泛型类型**：`from __future__ import annotations` 是必需的，用于支持前向引用语法 2. **线程安全**：`CyclicExecutor` 内部使用锁保护共享状态 3. **状态不可变性**：建议状态对象设计为不可变或使用 Pydantic 的深拷贝机制 4. **信号机制**：当前实现中 `signal` 参数默认为 `None`，可根据需要扩展 --- ## Search状态模块 ### 相关页面 相关主题：[CoT状态模块](#page-8), [有限状态机模式](#page-6)
相关源码文件 以下源码文件用于生成本页说明： - [cyclic_agent/search.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/search.py) - [cyclic_agent/state.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/state.py) - [cyclic_agent/__init__.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/__init__.py) - [cyclic_agent/executor.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py) - [examples/bilibili_surfer/bilibili_surfer.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py) - [examples/bilibili_surfer/fifo.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/fifo.py)
# Search状态模块 ## 概述 Search状态模块是CyclicAgent框架中用于封装搜索功能的核心组件。该模块通过继承`State`基类实现了一个专门处理搜索查询的状态类型，允许代理（Agent）在状态机中执行搜索操作并根据搜索结果进行状态转换。 Search模块采用了策略模式设计，`search`方法被声明为抽象方法（`NotImplementedError`），由子类实现具体的搜索逻辑。这种设计使得Search状态模块具有高度的灵活性和可扩展性，可以适配各种不同的搜索数据源和搜索API。 资料来源：[cyclic_agent/search.py:1-15]() ## 架构设计 ### 类结构 ```mermaid classDiagram class State~SigT~ { <> +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-18]()[cyclic_agent/search.py:5]() ### 类型定义 | 属性名 | 类型 | 说明 | |--------|------|------| | `query` | `str` | 搜索查询字符串，包含用户或代理要搜索的关键词 | | `exit_` | `Callable[[Annotated[str, "search result"]], State]` | 退出回调函数，接收搜索结果字符串作为参数，返回下一个状态 | 回调函数的`Annotated`注解用于标记搜索结果的语义含义，便于类型检查和代码文档化。 资料来源：[cyclic_agent/search.py:6-7]() ## 核心方法 ### next方法 `next`方法是Search状态的状态转换实现。当执行器调用此方法时，会触发搜索操作并通过回调函数返回下一个状态。 ```mermaid graph TD A[执行器调用 next] --> B[执行 searchquery] B --> C[获取搜索结果] C --> D[调用 exit_ 回调] D --> E[返回下一状态] ``` 方法签名如下： ```python 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`方法是抽象搜索逻辑的占位符实现，设计者意图由子类重写此方法以提供具体的搜索功能。 ```python 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`执行 ```mermaid graph LR A[其他状态] -->|转换| B[Search状态] B -->|执行搜索| C[获取结果] C -->|回调| D[下一状态] D -->|循环| A ``` ### 与Fifo内存组件配合 在真实应用场景中（如B站冲浪示例），Search操作通常与`Fifo`内存组件配合使用。`Fifo`类实现了一个容量为100的先进先出队列，用于记录代理的历史行为： ```python 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`类，但展示了类似的搜索模式实现： ```python 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`负责管理和执行状态机的循环： ```python 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： ```python from cyclic_agent import State, CyclicExecutor ``` Search类本身未包含在`__all__`列表中，这可能意味着它被视为内部实现或需要直接从子模块导入： ```python from cyclic_agent.search import Search ``` ## 扩展指南 ### 实现自定义Search子类 开发者可以通过继承Search类来实现特定数据源的搜索功能： ```python 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模块的灵活性使其可以与各种外部搜索服务集成： | 搜索服务 | 集成方式 | 适用场景 | |----------|----------|----------| | Elasticsearch | 同步HTTP请求 | 全文搜索 | | Algolia | SDK调用 | 应用内搜索 | | Bing/Google API | REST API | 网络搜索 | | 数据库全文索引 | SQL查询 | 结构化数据搜索 | ## 设计理念 Search状态模块体现了CyclicAgent框架的几个核心设计原则： 1. **状态即行为**：每个状态封装了特定的行为逻辑，搜索操作被建模为状态而非普通方法 2. **回调驱动的状态转换**：通过`exit_`回调函数实现搜索结果与状态转换的解耦 3. **抽象与实现分离**：`search`方法的抽象声明允许子类提供不同实现，保持框架核心稳定 4. **有限状态机语义**：状态只能转换到`exit_`回调指定的下一状态，维持状态机的可预测性 这种设计使得代理系统既能执行复杂的搜索任务，又能保持清晰的状态流转逻辑，便于调试和扩展。 --- ## CoT状态模块 ### 相关页面 相关主题：[Search状态模块](#page-7)
相关源码文件 以下源码文件用于生成本页说明： - [cyclic_agent/cot.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/cot.py) - [cyclic_agent/state.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/state.py) - [cyclic_agent/executor.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py) - [cyclic_agent/__init__.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/__init__.py) - [examples/hello_world/hello_world.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/hello_world/hello_world.py)
# CoT状态模块 ## 概述 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]() ### 组件构成 | 组件名称 | 类型 | 说明 | |---------|------|------| | `exit_` | `Callable[[str], State]` | 退出回调函数，接收LLM输出并返回下一个状态 | | `llm` | `State` | 底层LLM状态处理器，负责与语言模型交互 | | `prompt` | `str` | 用户提供的原始提示词 | 资料来源：[cyclic_agent/cot.py:6-8]() ## 工作原理 ### 思维链增强机制 当`CoT`状态的`next()`方法被调用时，会自动对原始提示词进行思维链增强处理。具体流程如下： ```python 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]() ### 状态转换流程 ```mermaid graph TD A[CoT.next 调用] --> B[拼接思维链提示词] B --> C[调用 LLM 处理] C --> D[获取模型输出] D --> E[触发 callback] E --> F[执行 exit_ 回调] F --> G[返回下一个 State] ``` ## 与State基类的关系 ### 抽象基类定义 `State`是一个使用Pydantic的`BaseModel`实现的泛型抽象类，定义了状态转换的核心接口： ```python 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]() ### 接口契约 | 方法 | 参数 | 返回值 | 说明 | |-----|------|--------|------| | `next` | `signal: SigT \| None` | `State[SigT]` | 执行状态转换，返回下一个状态 | `CoT`类完整实现了这一接口，确保与CyclicExecutor执行器的兼容性。 ## 执行器集成 `CyclicExecutor`负责管理和调度所有状态，包括`CoT`状态。执行器采用独立线程运行主循环，周期性地调用状态的`next()`方法： ```python 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]() ### 执行流程图 ```mermaid 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示例的典型模式： ```python 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`模块导出，可通过以下方式导入： ```python from cyclic_agent.cot import CoT ``` 资料来源：[cyclic_agent/cot.py:1]() ## 设计优势 | 优势 | 说明 | |------|------| | **解耦设计** | CoT逻辑与LLM交互分离，便于单元测试 | | **可组合性** | 可与任何实现`State`接口的组件配合使用 | | **回调灵活性** | 允许外部控制状态转换逻辑 | | **框架兼容** | 无缝集成到CyclicExecutor状态机执行体系 | ## 与其他状态的关系 在CyclicAgent框架中，`CoT`状态与其他状态类型形成互补： ```mermaid graph LR A[CoT] -->|思维链推理| B[普通State] C[Search] -->|搜索增强| B D[自定义状态] -->|领域逻辑| B B --> E[CyclicExecutor] ``` - **CoT**：提供链式推理能力 - **Search**：提供搜索增强能力 - **自定义State**：处理特定业务逻辑 ## 总结 CoT状态模块是CyclicAgent框架实现自主代理智能推理的核心组件。通过在提示词中嵌入思维链指令，该模块能够引导大语言模型进行更深入、更结构化的思考。结合状态机模式和回调机制，CoT状态为构建复杂、可扩展的AI代理系统提供了坚实的技术基础。 --- ## Hello World示例 ### 相关页面 相关主题：[快速入门指南](#page-3), [Bilibili Surfer示例](#page-10)
相关源码文件 以下源码文件用于生成本页说明： - [examples/hello_world/hello_world.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/hello_world/hello_world.py) - [cyclic_agent/state.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/state.py) - [cyclic_agent/executor.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py) - [cyclic_agent/__init__.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/__init__.py) - [cyclic_agent/cot.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/cot.py)
# Hello World示例 ## 概述 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]() ## 系统架构 ### 组件关系图 ```mermaid graph TD A[AskQuestion 状态] -->|next()| B[AnswerQuestion 状态] B -->|next()| A C[CyclicExecutor] -->|管理执行| A C -->|管理执行| B ``` ### 类结构 | 类名 | 文件位置 | 继承关系 | 职责 | |------|---------|---------|------| | `State` | cyclic_agent/state.py | BaseModel | 状态基类，定义`next()`抽象方法 | | `CyclicExecutor` | cyclic_agent/executor.py | - | 执行器，管理状态机的运行、暂停、终止 | | `AskQuestion` | examples/hello_world/hello_world.py | State[None] | 提问状态 | | `AnswerQuestion` | examples/hello_world/hello_world.py | State[None] | 回答状态 | ## 状态实现详解 ### AskQuestion状态 ```python 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状态 ```python 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]() | 方法 | 功能描述 | |------|---------| | `start(initial_state)` | 启动执行器，使用独立线程运行状态机 | | `pause()` | 暂停状态转换 | | `resume()` | 恢复状态转换 | | `kill()` | 终止执行器 | ### 执行流程 ```mermaid 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]() ## 配置与环境 ### 依赖项 | 依赖包 | 用途 | |-------|------| | cohere | LLM对话API客户端 | | python-dotenv | 环境变量管理 | | cyclic_agent | 核心框架 | ### 环境变量 | 变量名 | 说明 | |-------|------| | COHERE_API_KEY | Cohere API密钥 | ```python from dotenv import load_dotenv load_dotenv() co = cohere.Client(os.environ.get("COHERE_API_KEY")) ``` 资料来源：[examples/hello_world/hello_world.py:7-10]() ## 运行示例 ### 代码 ```python if __name__ == "__main__": initial_state = AskQuestion() executor = CyclicExecutor(5) executor.start(initial_state) time.sleep(20) ``` ### 参数说明 | 参数 | 值 | 说明 | |------|-----|------| | default_time_interval | 5 | 每次状态转换间隔5秒 | | sleep时间 | 20 | 程序运行20秒后退出 | ## 进阶示例参考 除了Hello World示例外，仓库还提供了更复杂的示例： | 示例名称 | 文件路径 | 特性 | |---------|---------|------| | Chain of Thought | cyclic_agent/cot.py | 思维链模式 | | Search | cyclic_agent/search.py | 搜索功能封装 | | Bilibili冲浪者 | examples/bilibili_surfer/ | 完整代理实现 | ### 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示例 ### 相关页面 相关主题：[Hello World示例](#page-9)
相关源码文件 以下源码文件用于生成本页说明： - [examples/bilibili_surfer/bilibili_surfer.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py) - [examples/bilibili_surfer/fifo.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/fifo.py) - [cyclic_agent/state.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/state.py) - [cyclic_agent/executor.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/executor.py) - [cyclic_agent/__init__.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/__init__.py)
# Bilibili Surfer示例 ## 概述 Bilibili Surfer是cyclic-agent框架的一个完整示例应用，演示了如何构建一个自主运行的AI代理，使其能够像真实用户一样浏览Bilibili视频网站、阅读评论并自动回复。该示例展示了有限状态机（FSM）设计模式在实际应用中的强大能力，将AI代理的行为抽象为多个可相互转换的状态，每个状态根据内部属性和外部信号推断下一个状态。 这个示例的核心价值在于它演示了如何将复杂的AI行为与外部API集成。通过结合Cohere大语言模型和Bilibili API，系统能够理解自然语言指令，做出合理的决策，并在视频网站上执行实际操作。整个系统运行在CyclicExecutor的调度下，按照预设的时间间隔不断循环执行状态转换，直到达到预设的迭代次数或被手动终止。 ## 系统架构 ### 整体架构图 ```mermaid 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`等辅助组件 2. **基础服务层** - `Fifo`：内存管理组件，持久化记录代理行为历史 - Bilibili API集成：通过bilibili_api库访问B站服务 3. **业务逻辑层** - `BilibiliStateBase`：所有业务状态的基类 - `BrowsingVideo`、`ReadingComments`、`PostComment`：具体业务状态 4. **AI决策层** - Cohere Client：大语言模型集成 - 提示词模板：指导AI做出合理的决策 ## 状态机设计 ### 状态定义 系统定义了三种主要状态，每种状态都继承自`BilibiliStateBase`，而`BilibiliStateBase`又继承自`State[None]`基类。 | 状态类名 | 功能描述 | 主要属性 | |---------|---------|---------| | `BrowsingVideo` | 搜索并浏览视频 | 无特殊属性，继承基类属性 | | `ReadingComments` | 阅读视频下的热门评论 | `video_bvid`、`video_title`、`video_description` | | `PostComment` | 回复选定的评论 | 包含`reply_to`、`reply_to_oid`用于定位回复目标 | ### 状态转换关系 ```mermaid stateDiagram-v2 [*] --> BrowsingVideo BrowsingVideo --> BrowsingVideo: 返回继续浏览 BrowsingVideo --> ReadingComments: 选择评论 ReadingComments --> BrowsingVideo: 返回视频浏览 ReadingComments --> ReadingComments: 继续读评论 ReadingComments --> PostComment: 选择回复 PostComment --> BrowsingVideo: 完成回复 ``` ### 类型别名定义 为了确保状态转换的类型安全，系统使用Python的类型别名（`type`语句）定义每个状态可到达的下一个状态： ```python type BrowsingVideoReachable = Union[BrowsingVideo, ReadingComments] type ReadingCommentsReachable = Union[BrowsingVideo, ReadingComments, PostComment] type PostCommentReachable = Union[BrowsingVideo] ``` 这种设计确保了编译时就能发现无效的状态转换，防止运行时错误。 ## 核心类详解 ### BilibiliStateBase `BilibiliStateBase`是所有Bilibili相关状态的抽象基类，它封装了状态机运行所需的所有共享资源和通用逻辑。 ```python class BilibiliStateBase(State[None], ABC): model_config = ConfigDict(arbitrary_types_allowed=True) initial_prompt: str memory: Fifo co: Client credential: Credential ``` **属性说明**： | 属性名 | 类型 | 用途 | |-------|------|------| | `initial_prompt` | str | 初始提示词，定义AI的角色和行为准则 | | `memory` | Fifo | 先进先出内存，存储历史行为记录 | | `co` | Client | Cohere客户端，用于调用大语言模型 | | `credential` | Credential | Bilibili API凭证，包含SESSDATA、BILI_JCT、BUVID3 | **核心方法**：`_infer_state_helper`方法使用LLM来决定下一个状态。当调用此方法时，会向LLM传递当前状态描述、历史行为记录以及可选的下一个状态列表，LLM返回的文本与预定义的选项进行模式匹配，从而决定状态转换的方向。 资料来源：[examples/bilibili_surfer/bilibili_surfer.py:43-58]() ### BrowsingVideo状态 `BrowsingVideo`是系统的入口状态，负责视频搜索和浏览功能。 ```python 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进行选择。 ```python 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`是系统的执行状态，负责生成并发送评论回复。 ```python 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代理的所有历史行为。 ```python 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("") ``` **主要特性**： | 特性 | 说明 | |-----|------| | 容量限制 | 最多存储100条记录，超出后自动移除最旧的记录 | | 时间戳 | 每条记录都带有精确到秒的时间戳 | | 持久化 | 所有操作都会追加写入`fifo_log.txt`日志文件 | | 提示生成 | `prompt()`方法返回格式化的历史记录供LLM参考 | **核心方法**： ```python 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)]) ``` ### 内存数据流 ```mermaid 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`是驱动整个状态机运转的核心执行器，采用多线程架构实现状态循环。 ```python 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()`方法则可以干净地终止整个循环。 ### 主循环逻辑 ```python 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()`方法获取下一个状态；然后根据预设的时间间隔进行休眠，控制状态转换的频率。 ### 示例启动流程 ```python 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`文件管理： | 变量名 | 来源 | 用途 | |-------|------|------| | `COHERE_API_KEY` | Cohere官网 | 调用Cohere大语言模型 | | `SESSDATA` | Bilibili网站 | 用户会话认证 | | `BILI_JCT` | Bilibili网站 | CSRF令牌 | | `BUVID3` | Bilibili网站 | 设备标识 | | `name` | 自定义 | 机器人联系信息（用于自动回复的免责声明） | ### 依赖库 ```python 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调用参数（确保参数语义兼容） ### 调整运行参数 | 参数 | 位置 | 说明 | |-----|------|------| | 状态转换间隔 | `CyclicExecutor(5)` | 5秒，可调整执行频率 | | 内存容量 | `Fifo.capacity` | 默认100条记录 | | 搜索结果数量 | `range(10)` | 获取Top10视频 | | 评论展示数量 | `min(5, ...)` | 显示5条热门评论 | ## 总结 Bilibili Surfer示例完整展示了cyclic-agent框架在实际应用中的使用方法。通过有限状态机将复杂的AI行为分解为清晰的状态单元，结合大语言模型实现智能决策，再配合外部API执行实际操作，整个系统能够自主、连续地在Bilibili网站上执行浏览、评论和回复等任务。这个示例不仅是一个可运行的演示程序，更是一个展示AI Agent设计最佳实践的教学案例，为开发者构建自己的AI代理应用提供了完整的参考模板。 --- --- ## Doramagic 踩坑日志 项目：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