# https://github.com/xingjianll/cyclic-agent 项目说明书
生成时间：2026-05-20 06:16:55 UTC
## 目录
- [Installation and Setup](#page-1)
- [Project Overview](#page-2)
- [State Base Class](#page-3)
- [CyclicExecutor](#page-4)
- [System Architecture](#page-5)
- [Creating Custom States](#page-6)
- [Memory Management with Fifo](#page-7)
- [Hello World Example](#page-8)
- [Bilibili Surfer Example](#page-9)
- [Helper States](#page-10)
## Installation and Setup
### 相关页面
相关主题：[State Base Class](#page-3), [Hello World Example](#page-8)
相关源码文件
以下源码文件用于生成本页说明：
- [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)
- [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)
# Installation and Setup
## Overview
CyclicAgent is a framework for creating LLM-powered, fully-autonomous AI agents using a finite state machine (FSM) architecture. The framework abstracts an agent as a collection of states that implement transition functions, allowing for cyclic execution patterns.
资料来源：[README.md:1-8]()
## Prerequisites
Before installing CyclicAgent, ensure your environment meets the following requirements:
| Requirement | Minimum Version | Notes |
|-------------|-----------------|-------|
| Python | 3.10+ | Required for Pydantic v2 features |
| pip | Latest recommended | For package installation |
| LLM API Key | Provider-specific | Cohere, OpenAI, or similar |
### Required Dependencies
The core dependencies include:
- **Pydantic** - Base model and type validation
- **Cohere SDK** or **OpenAI SDK** - For LLM integration
- **python-dotenv** - For environment variable management
资料来源：[cyclic_agent/state.py:1-4]()
## Installation Methods
### Standard Installation (Recommended)
Install the package directly from PyPI:
```bash
pip install cyclic-agent
```
资料来源：[README.md:7-9]()
### Development Installation
For contributing or modifying the source code:
```bash
git clone https://github.com/xingjianll/cyclic-agent.git
cd cyclic-agent
pip install -e .
```
### Dependencies in pyproject.toml
The package configuration specifies core dependencies for state management and LLM integration.
资料来源：[pyproject.toml](https://github.com/xingjianll/cyclic-agent/blob/main/pyproject.toml)
## Project Configuration
### Environment Variables Setup
Create a `.env` file in your project root to store sensitive credentials:
```bash
COHERE_API_KEY=your_api_key_here
```
#### Example .env Configuration
```bash
# LLM Configuration
COHERE_API_KEY=your_cohere_api_key
# Platform-specific (for Bilibili example)
SESSDATA=your_bilibili_sessdata
BILI_JCT=your_bilibili_jct
BUVID3=your_bilibili_buvid3
NAME=your_name_for_bot_footers
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:1-20]()
## Core Package Imports
After installation, import the essential components:
```python
from cyclic_agent import State, CyclicExecutor
```
资料来源：[cyclic_agent/__init__.py:1-6]()
### Available Exports
| Symbol | Type | Description |
|--------|------|-------------|
| `State` | Class | Abstract base class for all agent states |
| `CyclicExecutor` | Class | Execution engine for running the state machine |
## State Machine Architecture
The framework uses a state design pattern where each state implements a `next()` method returning the subsequent state.
```mermaid
graph TD
A[Initial State] -->|next()| B[State 1]
B -->|next()| C[State 2]
C -->|next()| D[State N]
D -->|next()| A
```
资料来源：[README.md:10-16]()
## Creating Your First Agent
### Basic Setup Pattern
```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"))
```
资料来源：[examples/hello_world/hello_world.py:1-16]()
### Implementing States
Each state must inherit from `State` and implement the `next()` method:
```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)
```
资料来源：[examples/hello_world/hello_world.py:18-24]()
### State Base Class Requirements
The `State` class is a Pydantic `BaseModel` with an abstract method:
```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-12]()
## Running the Executor
### Basic Execution
```python
if __name__ == "__main__":
initial_state = AskQuestion()
executor = CyclicExecutor(5) # 5-second interval
executor.start(initial_state)
time.sleep(20)
```
资料来源：[examples/hello_world/hello_world.py:34-40]()
### Executor Configuration
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `default_time_interval` | `float` | Required | Seconds between state transitions |
| `running` | `bool` | `False` | Execution status |
| `killed` | `bool` | `False` | Termination flag |
资料来源：[cyclic_agent/executor.py:4-10]()
### Executor Control Methods
```mermaid
graph LR
A[start] --> B[running=True]
B --> C[_main_loop]
C --> D{state.next}
D --> E[time.sleep]
E --> C
F[pause] --> G[running=False]
H[resume] --> B
I[kill] --> J[killed=True]
J --> K[thread.join]
```
The `CyclicExecutor` class provides thread-safe control:
```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
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()
```
资料来源：[cyclic_agent/executor.py:4-25]()
## Advanced Setup: Memory Integration
For agents requiring persistent memory, extend the base state class:
```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:28-35]()
### Memory Class Implementation
```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()))
```
资料来源：[examples/bilibili_surfer/fifo.py:1-14]()
## Complete Project Structure
```
cyclic-agent/
├── cyclic_agent/
│ ├── __init__.py # Package exports: State, CyclicExecutor
│ ├── state.py # State base class definition
│ ├── executor.py # CyclicExecutor implementation
│ ├── cot.py # Chain-of-thought state
│ └── search.py # Search state interface
├── examples/
│ ├── hello_world/ # Basic agent example
│ └── bilibili_surfer/ # Complex multi-state example
└── pyproject.toml # Package configuration
```
## Verification Checklist
- [ ] Python 3.10+ installed
- [ ] `pip install cyclic-agent` completed successfully
- [ ] `.env` file created with required API keys
- [ ] `from cyclic_agent import State, CyclicExecutor` imports without errors
- [ ] Basic state class compiles correctly
- [ ] Executor starts without exceptions
## Troubleshooting Common Setup Issues
| Issue | Solution |
|-------|----------|
| ImportError for State | Ensure `from __future__ import annotations` is present |
| Type validation errors | Use `model_config = ConfigDict(arbitrary_types_allowed=True)` |
| Threading issues | Executor methods are thread-safe via locks |
| Forward reference errors | Use `from __future__ import annotations` at file top |
资料来源：[cyclic_agent/state.py:1-2](), [examples/bilibili_surfer/bilibili_surfer.py:30]()
---
## Project Overview
### 相关页面
相关主题：[State Base Class](#page-3), [CyclicExecutor](#page-4)
相关源码文件
以下源码文件用于生成本页说明：
- [README.md](https://github.com/xingjianll/cyclic-agent/blob/main/README.md)
- [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/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)
- [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)
- [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)
# Project Overview
## Introduction
CyclicAgent is a Python framework designed for creating **LLM-powered, fully-autonomous AI agents**. It provides a clean abstraction for building AI systems that can continuously operate by modeling agent behavior as a finite state machine (FSM). 资料来源：[README.md:1]()
The framework enables developers to define AI behavior through a state design pattern, where each state represents a distinct behavior or action the agent can perform. States transition to one another based on internal attributes (such as memory and meta prompts) as well as external signals, creating a cyclic execution model that can run indefinitely. 资料来源：[README.md:6-9]()
## Core Design Philosophy
CyclicAgent abstracts an agent as a **finite state machine (FSM)** using the state design pattern. This approach offers several advantages:
| Benefit | Description |
|---------|-------------|
| **Modularity** | Each behavior is encapsulated in its own state class |
| **Scalability** | New states can be added without modifying existing ones |
| **Predictability** | State transitions follow defined rules |
| **Testability** | Individual states can be tested in isolation |
| **Extensibility** | The framework supports custom state types and transitions |
资料来源：[README.md:6-9]()
### State Transition Model
The fundamental principle of CyclicAgent is that **all states implement a state transition function**, which returns another state. This allows for chaining transition operations indefinitely, making the agent "Cyclic". 资料来源：[README.md:10-11]()
```mermaid
graph TD
A[Initial State] -->|next()| B[State 1]
B -->|next()| C[State 2]
C -->|next()| D[State 3]
D -->|next()| B
E[External Signal] -.->|signal| C
```
## Architecture
### System Components
CyclicAgent consists of the following core components:
| Component | File | Purpose |
|-----------|------|---------|
| `State` | `cyclic_agent/state.py` | Abstract base class for all agent states |
| `CyclicExecutor` | `cyclic_agent/executor.py` | Manages the execution loop of states |
| `Search` | `cyclic_agent/search.py` | Specialized state for search operations |
| `CoT` | `cyclic_agent/cot.py` | Chain-of-thought reasoning state |
资料来源：[cyclic_agent/__init__.py:1-6]()
### Class Hierarchy
```mermaid
classDiagram
class State~SigT~ {
<>
+next(signal) State
}
class CyclicExecutor {
-running: bool
-default_time_interval: float
+start(initial_state)
+pause()
+resume()
+kill()
}
class Search {
+query: str
+exit_: Callable
+search(query) str
}
class CoT {
+exit_: Callable
+llm: State
+prompt: str
}
State <|-- Search
State <|-- CoT
```
## Core Components
### State Base Class
The `State` class is the foundation of the framework. It is a **generic abstract base class** built on Pydantic's `BaseModel`, providing type safety and data validation. 资料来源：[cyclic_agent/state.py:1-13]()
```python
class State[SigT](BaseModel):
@abstractmethod
def next(self, signal: SigT | None = None) -> State[SigT]:
"""Transition to the next state."""
raise NotImplementedError
```
**Key Features:**
- **Generic Type Parameter**: `SigT` represents the signal type for state transitions
- **Pydantic Integration**: Enables automatic data validation and serialization
- **Abstract Method**: `next()` must be implemented by all concrete states
资料来源：[cyclic_agent/state.py:1-13]()
### CyclicExecutor
The `CyclicExecutor` is responsible for running the state machine in a separate thread. It manages the execution lifecycle including start, pause, resume, and kill operations. 资料来源：[cyclic_agent/executor.py:1-45]()
```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
```
| Parameter | Type | Description |
|-----------|------|-------------|
| `default_time_interval` | `float` | Sleep duration (seconds) between state transitions |
| `running` | `bool` | Flag indicating if executor is actively running |
| `killed` | `bool` | Flag indicating if executor has been terminated |
**Execution Flow:**
```mermaid
graph TD
A[Start] --> B{running?}
B -->|Yes| C[state = state.next()]
B -->|No| D[Wait]
C --> E[Sleep for interval]
E --> B
D --> B
F[Kill Signal] --> G[killed = True]
G --> H[Exit Loop]
```
资料来源：[cyclic_agent/executor.py:1-45]()
## State Implementation Patterns
### Simple State Pattern
States are implemented by extending the `State` class and implementing the `next()` method. Each state returns the next state to transition to. 资料来源：[examples/hello_world/hello_world.py:1-38]()
```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()
```
**Transition Diagram:**
```mermaid
graph LR
A[AskQuestion] -->|next()| B[AnswerQuestion]
B -->|next()| A
```
### State with Memory Pattern
Advanced states can maintain memory to track past actions and decisions. The Bilibili Surfer example demonstrates this pattern using a FIFO (First-In-First-Out) queue. 资料来源：[examples/bilibili_surfer/fifo.py:1-27]()
```python
class BilibiliStateBase(State[None], ABC):
model_config = ConfigDict(arbitrary_types_allowed=True)
initial_prompt: str
memory: Fifo
co: Client
credential: Credential
```
| Attribute | Type | Purpose |
|-----------|------|---------|
| `initial_prompt` | `str` | Base prompt for LLM interactions |
| `memory` | `Fifo` | Queue storing past agent actions |
| `co` | `Client` | Cohere LLM client instance |
| `credential` | `Credential` | Authentication credentials |
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:1-55]()
### State Inference Pattern
States can use LLM-driven inference to determine the next state dynamically:
```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
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:28-43]()
## Memory System
### FIFO Queue Implementation
The framework includes a `Fifo` class for maintaining conversation history and action logs:
```python
class Fifo:
def __init__(self):
self.capacity = 100
self.queue = []
self.log_file = "fifo_log.txt"
```
| Feature | Description |
|---------|-------------|
| `capacity` | Maximum number of items (default: 100) |
| `queue` | List storing items with timestamps |
| `log_file` | Persistent log file for audit trail |
**Behavior:**
- When capacity is reached, the oldest item is automatically removed
- Each item is timestamped and logged to file
- The `prompt()` method returns items in reverse chronological order
资料来源：[examples/bilibili_surfer/fifo.py:1-27]()
## Specialized States
### Search State
A generic search state that executes queries and transitions based on results:
```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-14]()
### Chain-of-Thought State
Enables step-by-step reasoning by appending "Let's think step by step." to prompts:
```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."
return self.llm(prompt=prompt, callback=self.exit_)
```
资料来源：[cyclic_agent/cot.py:1-15]()
## Complete Workflow Example
The Bilibili Surfer demonstrates a complete autonomous agent workflow:
```mermaid
graph TD
A[BrowsingVideo] -->|search videos| B[Display Top 10 Videos]
B --> C{Select Video}
C -->|browse| D[ReadingComments]
D -->|view comments| E{Choose Action}
E -->|reply| F[PostComment]
E -->|continue| D
F -->|done| A
D -->|back| A
```
**State Transitions:**
| Current State | Next State(s) | Trigger |
|---------------|---------------|---------|
| `BrowsingVideo` | `BrowsingVideo`, `ReadingComments` | LLM inference |
| `ReadingComments` | `BrowsingVideo`, `ReadingComments`, `PostComment` | LLM inference |
| `PostComment` | `BrowsingVideo` | Comment posted |
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:1-180]()
## Package Exports
The framework exposes a minimal public API:
```python
from cyclic_agent import State, CyclicExecutor
```
All core functionality is accessible through these two classes, keeping the API simple and intuitive for developers. 资料来源：[cyclic_agent/__init__.py:1-6]()
## Installation
Install the framework via pip:
```shell
pip install cyclic-agent
```
The package is available on PyPI and can be integrated into any Python project. 资料来源：[README.md:4-6]()
## Requirements
- Python 3.10+ (for generic syntax support)
- Pydantic for data validation
- `from __future__ import annotations` may be needed for forward references in some Python versions 资料来源：[README.md:28]()
## Summary
CyclicAgent provides a clean, Pythonic way to build autonomous AI agents using the finite state machine pattern. Its key strengths include:
1. **Simple Core API**: Only two main classes (`State` and `CyclicExecutor`) needed for basic usage
2. **Flexible State Definition**: States are Pydantic models with type safety
3. **Memory Integration**: Built-in support for FIFO queues and action logging
4. **Thread-Safe Execution**: Concurrent execution with pause/resume/kill controls
5. **Extensible Design**: Custom state types can be created by extending the base class
The framework is well-suited for building autonomous agents that need to make decisions, maintain context, and perform actions in a continuous loop.
---
## State Base Class
### 相关页面
相关主题：[CyclicExecutor](#page-4), [Creating Custom States](#page-6)
Relevant Source Files
以下源码文件用于生成本页说明：
- [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/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 Base Class
## Overview
The `State` class is the foundational abstraction in CyclicAgent, implementing the **State Design Pattern** to model agents as finite state machines (FSM). Every agent behavior is encapsulated as a state that can transition to another state through the `next()` method.
**Core Design Principle:** States implement a transition function that returns another state, enabling indefinite chaining and creating the "cyclic" nature of the agent framework.
资料来源：[cyclic_agent/state.py:6-9]()
---
## Architecture
### Class Definition
```python
class State[SigT](BaseModel):
@abstractmethod
def next(self, signal: SigT | None = None) -> State[SigT]:
"""Transition to the next state."""
raise NotImplementedError
```
| Property | Description |
|----------|-------------|
| Generic Parameter `SigT` | Signal type for external communication between states |
| Base Class | Pydantic `BaseModel` for data validation and serialization |
| Abstract Method `next()` | Must be implemented by subclasses to define state transitions |
资料来源：[cyclic_agent/state.py:6-9]()
### Generic Type Parameter `SigT`
The `SigT` generic parameter defines the type of signal an external entity can send to influence state transitions:
| Signal Type | Use Case |
|-------------|----------|
| `State[None]` | Autonomous states with no external input |
| `State[str]` | States that receive string-based signals |
| `State[Any]` | States accepting any signal type |
资料来源：[cyclic_agent/state.py:6]()
### Relationship with CyclicExecutor
```mermaid
graph TD
A[Initial State] -->|next()| B[State B]
B -->|next()| C[State C]
C -->|next()| A
D[CyclicExecutor] -->|controls| E[Running Loop]
E -->|calls next()| A
E -->|calls next()| B
E -->|calls next()| C
```
The `CyclicExecutor` runs a continuous loop that repeatedly calls `next()` on the current state, enabling cyclic transitions.
资料来源：[cyclic_agent/executor.py:28-33]()
---
## Implementation Patterns
### Pattern 1: Simple Two-State Cyclic Agent
```python
from cyclic_agent import State, CyclicExecutor
import cohere
class AskQuestion(State[None]):
def next(self, signal: None = None) -> AnswerQuestion:
response = cohere.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 = cohere.chat(message=self.question)
return AskQuestion()
```
资料来源：[examples/hello_world/hello_world.py:10-24]()
### Pattern 2: State with Context Attributes
States inherit from `BaseModel`, allowing Pydantic field definitions for context:
```python
class AnswerQuestion(State[None]):
question: str # Context attribute persisted across transitions
def next(self, signal: None = None) -> AskQuestion:
answer = cohere.chat(message=self.question)
return AskQuestion()
```
资料来源：[examples/hello_world/hello_world.py:19-24]()
### Pattern 3: Complex State with Type Unions
Define reachable states using Python type unions for type-safe transitions:
```python
type BrowsingVideoReachable = Union[BrowsingVideo, ReadingComments]
class BrowsingVideo(BilibiliStateBase):
@overrides
def next(self, signal: None = None) -> BrowsingVideoReachable:
# ... logic ...
match self._infer_state_helper('BrowsingVideo', 'ReadingComments'):
case 'BrowsingVideo':
return self
case 'ReadingComments':
return ReadingComments(**self.model_dump(), ...)
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:1-85]()
---
## State Machine Design
### Transition Flow
```mermaid
graph LR
S1[State 1] -->|next() returns State 2| S2[State 2]
S2 -->|next() returns State 3| S3[State 3]
S3 -->|next() returns State 1| S1
```
Each state transition:
1. Receives optional `signal` parameter
2. Executes state-specific logic
3. Returns the next state instance
4. Context (fields) can be preserved or replaced
### Base Class Requirements
| Requirement | Implementation |
|-------------|----------------|
| Inherit from `State[SigT]` | Provides FSM contract |
| Implement `next()` | Defines transition logic |
| Use `from __future__ import annotations` | Required for forward references |
资料来源：[examples/hello_world/hello_world.py:1-3]()
---
## Advanced Usage: Abstract Base State
For complex applications, create an abstract base state with shared functionality:
```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:
"""Helper method for LLM-driven state inference."""
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)}
"""
)
return self.co.chat(temperature=1, message=prompt).text
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:25-40]()
---
## Integration with CyclicExecutor
| Executor Method | State Interaction |
|-----------------|-------------------|
| `start(initial_state)` | Begins execution with starting state |
| `_main_loop(state)` | Calls `state.next()` in loop |
| `pause()` / `resume()` | Controls execution flow |
| `kill()` | Stops execution permanently |
```python
class CyclicExecutor:
def _main_loop(self, state: State) -> None:
while True:
if self.killed:
return
if self.running:
state = state.next() # State transition
time.sleep(self.default_time_interval)
```
资料来源：[cyclic_agent/executor.py:28-33]()
---
## Export and Import
The `State` class is exported from the package root:
```python
from cyclic_agent import State, CyclicExecutor
```
资料来源：[cyclic_agent/__init__.py:1-4]()
---
## Summary
| Aspect | Description |
|--------|-------------|
| **Purpose** | Base class for all agent states in FSM architecture |
| **Base** | Pydantic `BaseModel` |
| **Generic** | `State[SigT]` for signal type parameterization |
| **Key Method** | `next(signal)` returns next state |
| **Usage** | Inherit and implement `next()` for state behavior |
---
## CyclicExecutor
### 相关页面
相关主题：[State Base Class](#page-3), [Creating Custom States](#page-6), [Helper States](#page-10)
相关源码文件
以下源码文件用于生成本页说明：
- [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)
- [README.md](https://github.com/xingjianll/cyclic-agent/blob/main/README.md)
# CyclicExecutor
## Overview
CyclicExecutor is the core execution engine in the CyclicAgent framework. It manages the continuous execution loop that drives the finite state machine (FSM) pattern implemented by agents. The executor handles state transitions, threading, and timing control for autonomous agent behavior.
**Purpose:** The CyclicExecutor provides a runtime environment for executing state machines where each state can transition to another state indefinitely, enabling fully autonomous AI agents.
**Scope:** It abstracts away the complexity of thread management, timing, and execution flow, allowing developers to focus purely on defining state behavior.
资料来源：[cyclic_agent/executor.py:1-34]()
## Architecture
### Design Pattern
CyclicExecutor implements the **State Design Pattern** where agents are represented as finite state machines. Each state implements a `next()` method that returns the subsequent state, creating an unbounded chain of transitions.
```mermaid
graph TD
A[Initial State] -->|next()| B[State 1]
B -->|next()| C[State 2]
C -->|next()| D[State N]
D -->|next()| A
D -->|next()| B
```
### Class Structure
| Component | Type | Description |
|-----------|------|-------------|
| `running` | `bool` | Flag indicating if executor is active |
| `killed` | `bool` | Flag indicating permanent shutdown |
| `lock` | `threading.Lock` | Thread synchronization primitive |
| `thread` | `threading.Thread` | Background execution thread |
| `default_time_interval` | `float` | Sleep duration between state transitions |
资料来源：[cyclic_agent/executor.py:7-14]()
## API Reference
### Constructor
```python
def __init__(self, default_time_interval: float)
```
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `default_time_interval` | `float` | required | Seconds to sleep between each state transition |
### Methods
#### `start(initial_state: State) -> None`
Initiates the executor in a new background thread. The executor begins calling `state.next()` repeatedly, transitioning through states at intervals defined by `default_time_interval`.
| Parameter | Type | Description |
|-----------|------|-------------|
| `initial_state` | `State[SigT]` | The starting state of the FSM |
```python
executor = CyclicExecutor(5)
executor.start(initial_state)
```
#### `pause() -> None`
Temporarily halts execution without terminating the thread. The thread remains alive but the main loop skips state transitions.
#### `resume() -> None`
Resumes execution after a pause. State transitions continue from the current state.
#### `kill() -> None`
Permanently stops the executor and joins the thread. Sets `killed` flag to `True` ensuring the main loop terminates cleanly.
```python
executor.kill() # Clean shutdown
```
资料来源：[cyclic_agent/executor.py:16-31]()
## Execution Flow
### Main Loop Logic
```mermaid
flowchart TD
A[_main_loop started] --> B{killed?}
B -->|Yes| Z[Return]
B -->|No| C{running?}
C -->|No| D[Sleep default_time_interval]
D --> B
C -->|Yes| E[state = state.next()]
E --> F[Sleep default_time_interval]
F --> B
```
The `_main_loop` method runs in a dedicated thread and performs the following operations:
1. Check `killed` flag; exit if `True`
2. Check `running` flag; skip transition if `False`
3. Call `state.next()` to get the next state
4. Sleep for `default_time_interval` seconds
5. Repeat
资料来源：[cyclic_agent/executor.py:33-35]()
### Thread Safety
The executor uses a `threading.Lock` to protect shared state modifications:
```python
self.lock = threading.Lock()
def pause(self):
with self.lock:
self.running = False
```
All state modifications (`start`, `pause`, `resume`, `kill`) acquire the lock before modifying `running` or `killed` flags.
资料来源：[cyclic_agent/executor.py:10]()
## State Integration
### State Base Class
The executor expects states to implement the `State[SigT]` protocol:
```python
class State[SigT](BaseModel):
@abstractmethod
def next(self, signal: SigT | None = None) -> State[SigT]:
"""Transition to the next state."""
raise NotImplementedError
```
States are Pydantic models, enabling automatic serialization and validation of state data.
资料来源：[cyclic_agent/state.py:6-9]()
### Type Parameter Usage
The generic type `SigT` allows states to receive external signals of any type:
| Type Parameter | Signal Type | Usage |
|----------------|--------------|-------|
| `State[None]` | No signal | Simple autonomous agents |
| `State[str]` | String messages | Interactive agents |
| `State[dict]` | Structured data | Data-driven agents |
资料来源：[cyclic_agent/state.py:6]()
## Configuration Options
| Option | Value Type | Default | Effect |
|--------|------------|---------|--------|
| `default_time_interval` | `float` | Required | Controls execution speed; lower = faster transitions |
| Thread spawning | Automatic | - | Creates daemon thread on `start()` |
| Signal handling | Passive | - | No signal handling implemented |
### Interval Selection Guidelines
| Interval Range | Use Case |
|----------------|----------|
| `0.1 - 1.0` | Fast response scenarios, API polling |
| `1.0 - 5.0` | Balanced interaction, LLM response times |
| `5.0 - 60.0` | Slow operations, external API rate limits |
资料来源：[cyclic_agent/executor.py:12]()
## Usage Examples
### Basic Hello World
```python
from cyclic_agent import State, CyclicExecutor
class AskQuestion(State[None]):
def next(self, signal=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) -> AskQuestion:
answer = co.chat(message=self.question)
return AskQuestion()
# Initialize and run
executor = CyclicExecutor(5) # 5-second intervals
executor.start(AskQuestion())
time.sleep(20)
executor.kill()
```
资料来源：[examples/hello_world/hello_world.py:1-26]()
### Complex Agent: Bilibili Surfer
```python
from cyclic_agent import State, CyclicExecutor
class BrowsingVideo(BilibiliStateBase):
@overrides
def next(self, signal=None) -> BrowsingVideoReachable:
# Search and select video
response = self.co.chat(temperature=1, message=prompt).text
video = res['result'][int(response)]
# Return next state based on decision
match self._infer_state_helper('BrowsingVideo', 'ReadingComments'):
case 'BrowsingVideo':
return self
case 'ReadingComments':
return ReadingComments(**self.model_dump(), ...)
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:1-95]()
### Typical Execution Pattern
```mermaid
sequenceDiagram
participant Main as Main Thread
participant Executor as CyclicExecutor
participant State as Current State
Main->>Executor: start(initial_state)
Executor->>Executor: Create Thread
loop Every default_time_interval seconds
Executor->>State: state.next()
State-->>Executor: Next State
Executor->>State: Update state reference
end
Main->>Executor: kill()
Executor->>Executor: Set killed=True
Executor->>Main: Thread joins
```
## Exports
CyclicExecutor is exported from the main package:
```python
from cyclic_agent import State, CyclicExecutor
```
```python
__all__ = ["State", "CyclicExecutor"]
```
资料来源：[cyclic_agent/__init__.py:1-6]()
## Limitations
- **Daemon Thread:** The execution thread is a daemon thread, meaning it won't prevent the Python process from exiting
- **No Signal Handling:** External signals cannot interrupt the execution loop
- **No State Persistence:** States are not persisted between sessions
- **Single Thread:** Only one thread executes states; no parallel state machines
资料来源：[cyclic_agent/executor.py:18]()
## Related Components
| Component | File | Relationship |
|-----------|------|--------------|
| `State` | `cyclic_agent/state.py` | Base class for states |
| `Search` | `cyclic_agent/search.py` | Generic search state |
| `CoT` | `cyclic_agent/cot.py` | Chain-of-thought state |
| `Fifo` | `examples/bilibili_surfer/fifo.py` | Memory for states |
资料来源：[cyclic_agent/search.py:1-14](), [cyclic_agent/cot.py:1-18](), [examples/bilibili_surfer/fifo.py:1-30]()
---
## System Architecture
### 相关页面
相关主题：[State Base Class](#page-3), [CyclicExecutor](#page-4), [Memory Management with Fifo](#page-7)
相关源码文件
以下源码文件用于生成本页说明：
- [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)
# System Architecture
## Overview
CyclicAgent is a framework for building autonomous AI agents based on the **Finite State Machine (FSM)** design pattern. The architecture abstracts an agent as a collection of states, where each state implements a transition function that returns another state, enabling cyclic execution flows. 资料来源：[README.md:9]()
The system is designed to handle autonomous workflows where agents can:
- Maintain internal state (memory, prompts)
- Respond to external signals
- Transition between states infinitely
- Interact with external environments (APIs, platforms)
## Core Design Principles
### State Pattern Implementation
The architecture uses Python's generic type syntax (PEP 695) combined with Pydantic for state modeling. Each state is a Pydantic model that encapsulates both data and behavior. 资料来源：[cyclic_agent/state.py:8]()
### Key Characteristics
| Characteristic | Description |
|----------------|-------------|
| **Type Safety** | Generic type parameters enable compile-time checking |
| **Serialization** | Pydantic BaseModel provides automatic serialization |
| **Extensibility** | Abstract base allows custom state implementations |
| **Thread Safety** | Executor uses locks for safe concurrent operation |
## Architecture Components
### State Base Class
```python
class State[SigT](BaseModel):
@abstractmethod
def next(self, signal: SigT | None = None) -> State[SigT]:
"""Transition to the next state."""
raise NotImplementedError
```
The `State` class serves as the foundation for all agent states:
| Parameter | Type | Description |
|-----------|------|-------------|
| `SigT` | Generic | Signal type for state transitions |
| `next()` | Method | Abstract transition function returning next state |
| `signal` | Parameter | Optional external signal influencing transition |
资料来源：[cyclic_agent/state.py:7-11]()
### CyclicExecutor
The `CyclicExecutor` manages the runtime execution loop for state machines:
```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
```
| Method | Description |
|--------|-------------|
| `start(initial_state)` | Initiates the execution loop in a separate thread |
| `pause()` | Halts execution without terminating |
| `resume()` | Continues execution from paused state |
| `kill()` | Stops the executor and joins the thread |
资料来源：[cyclic_agent/executor.py:1-29]()
## Execution Flow
```mermaid
graph TD
A[Start: CyclicExecutor.start] --> B{self.running?}
B -->|Yes| C[state = state.next]
B -->|No| D[Wait]
C --> E[Sleep: default_time_interval]
E --> B
D --> B
F[Kill Signal] --> G[self.killed = True]
G --> B
B -->|self.killed| H[Exit Loop]
```
The main loop continuously calls `state.next()` and sleeps for the configured interval. Thread safety is maintained through a lock that guards the `running` and `killed` flags. 资料来源：[cyclic_agent/executor.py:27-33]()
## State Machine Structure
### Typical State Implementation
```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()
```
Each state class:
1. Inherits from `State[SignalType]`
2. Stores relevant data as Pydantic fields
3. Implements `next()` returning the next state instance
资料来源：[examples/hello_world/hello_world.py:13-26]()
### State Transitions
```mermaid
graph LR
A[AskQuestion] -->|next()| B[AnswerQuestion]
B -->|next()| A
```
The cyclic nature allows indefinite chaining of states, enabling continuous autonomous operation. 资料来源：[README.md:27-42]()
## Supporting Components
### Memory Management (Fifo)
The `Fifo` class provides persistent memory for states:
```python
class Fifo:
def __init__(self):
self.capacity = 100
self.queue = []
self.log_file = "fifo_log.txt"
```
| Feature | Description |
|---------|-------------|
| `capacity` | Maximum stored items (default: 100) |
| `add(item)` | Adds item with timestamp, auto-evicts oldest |
| `prompt()` | Returns formatted history string |
| Logging | All additions logged to file with timestamps |
资料来源：[examples/bilibili_surfer/fifo.py:1-27]()
### Base State Pattern
Complex states inherit from a base class providing common functionality:
```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:
# LLM-based state inference logic
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:25-44]()
## Advanced State Patterns
### Chain of Thought (CoT)
The framework includes a `CoT` state for step-by-step reasoning:
```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]()
### Search State
The `Search` state provides a template for search operations:
```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-16]()
## Complex State Machine Example
The Bilibili Surfer example demonstrates a multi-state architecture:
```mermaid
graph TD
A[BrowsingVideo] -->|next| B[ReadingComments]
A -->|next| A
B -->|next| C[PostComment]
B -->|next| A
C -->|next| A
```
| State | Purpose | Reachable States |
|-------|---------|------------------|
| `BrowsingVideo` | Search and select videos | `BrowsingVideo`, `ReadingComments` |
| `ReadingComments` | View and select comments | `BrowsingVideo`, `ReadingComments`, `PostComment` |
| `PostComment` | Reply to selected comment | `BrowsingVideo` |
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:60-130]()
## Configuration and Threading
### Executor Threading Model
```mermaid
graph TD
subgraph MainThread
A[CyclicExecutor.start]
end
subgraph WorkerThread
B[_main_loop]
C[state.next]
D[time.sleep]
end
A -->|spawns| B
B --> C
C --> D
D --> B
```
The executor spawns a dedicated thread for the main loop, ensuring non-blocking operation. The lock ensures safe modification of `running` and `killed` flags from external threads. 资料来源：[cyclic_agent/executor.py:16-18]()
### Time Interval Configuration
```python
executor = CyclicExecutor(5) # 5-second interval between state transitions
executor.start(initial_state)
time.sleep(20) # Main thread sleeps while executor runs
```
The `default_time_interval` parameter controls the delay between state transitions, allowing rate limiting and API compliance. 资料来源：[examples/hello_world/hello_world.py:28-32]()
## Summary
The CyclicAgent architecture provides:
1. **State Pattern Foundation** - Generic, type-safe state classes with Pydantic
2. **Cyclic Execution** - Infinite state transitions enabling autonomous agents
3. **Thread-Safe Execution** - Lock-protected executor with pause/resume/kill controls
4. **Memory Persistence** - FIFO-based history with file logging
5. **Extensibility** - Abstract base classes for custom state implementations
The framework's simplicity—combining a base `State` class with an `Executor`—enables complex autonomous behaviors through composition of simple state transitions.
---
## Creating Custom States
### 相关页面
相关主题：[State Base Class](#page-3), [Hello World Example](#page-8), [Bilibili Surfer Example](#page-9)
相关源码文件
以下源码文件用于生成本页说明：
- [cyclic_agent/state.py](https://github.com/xingjianll/cyclic-agent/blob/main/cyclic_agent/state.py)
- [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/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)
# Creating Custom States
## Overview
In CyclicAgent, **states** are the fundamental building blocks of autonomous agents. Each state represents a distinct behavior or step in an agent's workflow, and states transition to one another through a well-defined interface. This design pattern enables the creation of complex, cyclic agent behaviors where execution can loop indefinitely through different states.
The framework abstracts an agent as a **finite state machine (FSM)** using the state design pattern. Each state implements a `next()` method that returns another state, creating an infinite transition chain that makes the agent "cyclic." 资料来源：[README.md:12-14]()
## Architecture
### State Pattern Design
The `State` class serves as the abstract base for all agent states. It leverages Python's generic type system to provide type safety for signals passed between states.
```mermaid
graph TD
A[State SigT] -->|extends| B[AskQuestion]
A -->|extends| C[AnswerQuestion]
A -->|extends| D[CustomState]
B -->|next() returns| C
C -->|next() returns| B
D -->|next() returns| E[NextState]
```
### Key Components
| Component | File | Purpose |
|-----------|------|---------|
| `State[SigT]` | `cyclic_agent/state.py` | Abstract base class for all states |
| `CyclicExecutor` | `cyclic_agent/executor.py` | Executes the state machine loop |
| `CyclicExecutor.start()` | `cyclic_agent/executor.py` | Initiates the state machine |
## Creating Basic States
### Step 1: Define Your State Class
Extend the `State` class and implement the `next()` method:
```python
from __future__ import annotations
from cyclic_agent import State, CyclicExecutor
class MyState(State[None]):
def next(self, signal: None = None) -> AnotherState:
# Perform state actions
return AnotherState()
```
### Step 2: Implement State Logic
The `next()` method should contain the logic for the current state and return the next state to transition to:
```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)
```
资料来源：[examples/hello_world/hello_world.py:14-18]()
### Step 3: Create the State Transition Pair
```python
class AnswerQuestion(State[None]):
question: str
def next(self, signal: None = None) -> AskQuestion:
answer = co.chat(message=self.question)
print(answer)
return AskQuestion()
```
资料来源：[examples/hello_world/hello_world.py:21-26]()
### Step 4: Execute the State Machine
```python
if __name__ == "__main__":
initial_state = AskQuestion()
executor = CyclicExecutor(5) # 5-second interval between transitions
executor.start(initial_state)
```
资料来源：[examples/hello_world/hello_world.py:29-33]()
## State API Reference
### State Class
```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:1-9]()
| Parameter | Type | Description |
|-----------|------|-------------|
| `signal` | `SigT \| None` | Optional signal passed to the state for external input |
| **Returns** | `State[SigT]` | The next state in the transition chain |
### State Data Model
States inherit from Pydantic's `BaseModel`, providing automatic validation and serialization:
```python
class AnswerQuestion(State[None]):
question: str # State-specific attribute
```
资料来源：[examples/hello_world/hello_world.py:20-21]()
## CyclicExecutor
### Execution Model
The `CyclicExecutor` runs the state machine in a background thread, transitioning between states at configurable intervals.
```mermaid
graph TD
A[Start] --> B{self.running?}
B -->|Yes| C[state = state.next()]
B -->|No| D[Wait...]
C --> E[Sleep interval]
E --> B
D --> B
F[Kill] --> G[Exit loop]
```
### CyclicExecutor API
```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:6-12]()
| Method | Description |
|--------|-------------|
| `start(initial_state)` | Start execution with an initial state |
| `pause()` | Pause the state machine |
| `resume()` | Resume from pause |
| `kill()` | Stop execution completely |
### Thread Safety
The executor uses a threading lock to ensure thread-safe state transitions:
```python
def pause(self):
with self.lock:
self.running = False
```
资料来源：[cyclic_agent/executor.py:27-29]()
## Advanced State Patterns
### States with Memory
Complex agents often need to remember past actions. States can include memory components:
```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:16-22]()
### FIFO Memory Implementation
```python
class Fifo:
def __init__(self):
self.capacity = 100
self.queue = []
def add(self, item):
if len(self.queue) >= self.capacity:
self.queue.pop(0)
self.queue.append(item)
def prompt(self):
return "\n".join([f"{time} - {text}" for text, time in reversed(self.queue)])
```
资料来源：[examples/bilibili_surfer/fifo.py:1-19]()
### State-Independent Transitions
States can transition to different next states based on internal logic using pattern matching:
```python
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:58-63]()
### State Inference Helper
```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.
"""
)
text = self.co.chat(temperature=1, message=prompt).text
return text
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:25-36]()
## Type Aliases for State Transitions
CyclicAgent uses Python type aliases to define valid state transitions, enabling static type checking:
```python
type BrowsingVideoReachable = Union[BrowsingVideo, ReadingComments]
type ReadingCommentsReachable = Union[BrowsingVideo, ReadingComments, PostComment]
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:13-14]()
## Complete Example: Two-State Loop
```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)
```
资料来源：[examples/hello_world/hello_world.py:1-33]()
## Best Practices
### Forward References
Always use `from __future__ import annotations` for forward references in type hints:
```python
from __future__ import annotations
```
### State Immutability
Return new state instances rather than modifying existing ones to ensure thread safety:
```python
def next(self, signal: None = None) -> AskQuestion:
return AskQuestion() # Return new instance
```
### Signal Usage
Leverage the signal parameter for external state control:
```python
class ControlledState(State[str]):
def next(self, signal: str | None = None) -> State:
if signal == "exit":
return ExitState()
return AnotherState()
```
### Pydantic Configuration
Use `ConfigDict(arbitrary_types_allowed=True)` when states contain non-Pydantic types like external clients or credentials:
```python
model_config = ConfigDict(arbitrary_types_allowed=True)
```
## Execution Flow
```mermaid
sequenceDiagram
participant User
participant Executor
participant State1
participant State2
User->>Executor: start(AskQuestion)
Executor->>State1: next()
State1-->>Executor: AnswerQuestion
Executor->>State2: next()
State2-->>Executor: AskQuestion
Executor->>State1: next()
Note over Executor: Loop continues...
```
## Summary
Creating custom states in CyclicAgent involves:
1. **Extending** `State[SigT]` with your state class
2. **Implementing** the `next()` method with state logic and transitions
3. **Using** Pydantic for data model attributes
4. **Executing** through `CyclicExecutor` with configurable intervals
5. **Leveraging** memory and state inference for complex behaviors
The state pattern enables modular, maintainable agent design where each state is independent and transitions are explicit, making the agent behavior predictable and debuggable.
---
## Memory Management with Fifo
### 相关页面
相关主题：[System Architecture](#page-5), [Bilibili Surfer Example](#page-9)
Related Source Files
以下源码文件用于生成本页说明：
- [examples/bilibili_surfer/fifo.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/fifo.py)
- [examples/bilibili_surfer/bilibili_surfer.py](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py)
# Memory Management with Fifo
## Overview
The `Fifo` class implements a First-In-First-Out (FIFO) memory buffer for the CyclicAgent framework. It serves as a persistent memory mechanism that tracks agent actions over time, enabling stateful behavior across multiple execution cycles. The memory stores action records with timestamps and provides a formatted prompt interface for LLM consumption.
**Purpose and Scope:**
- Maintains a rolling history of agent actions with temporal ordering
- Persists memory entries to a log file for debugging and auditability
- Provides context to language models about past agent behavior
- Enforces a configurable capacity limit to prevent unbounded memory growth
资料来源：[examples/bilibili_surfer/fifo.py:1-35](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/fifo.py)
## Architecture
The `Fifo` class is a simple yet effective memory component that integrates with the broader `BilibiliStateBase` abstraction. It operates as a bounded queue with automatic eviction of oldest entries.
```mermaid
graph TD
A[Agent Action] -->|add| B[Fifo Memory]
B --> C{Capacity Check}
C -->|queue full| D[Evict Oldest Entry]
C -->|space available| E[Append to Queue]
D --> E
E --> F[Log to fifo_log.txt]
F --> G[Timestamp Added]
G --> H[Memory Buffer]
H --> I[LLM Context via prompt]
style A fill:#e1f5fe
style I fill:#fff3e0
```
The memory flows into agent decision-making through the `_infer_state_helper` method, which includes past actions in LLM prompts.
资料来源：[examples/bilibili_surfer/fifo.py:1-35](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/fifo.py)
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:38-52](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py)
## Class Reference
### Fifo
The core memory management class.
| Attribute | Type | Default | Description |
|-----------|------|---------|-------------|
| `capacity` | `int` | `100` | Maximum number of entries before oldest eviction |
| `queue` | `list` | `[]` | Internal storage for memory entries |
| `log_file` | `str` | `"fifo_log.txt"` | Path to persistent log file |
#### Methods
##### `__init__()`
Initializes the Fifo instance and ensures the log file exists.
```python
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("")
```
资料来源：[examples/bilibili_surfer/fifo.py:6-14](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/fifo.py)
##### `add(item: str) -> None`
Adds a string item to the memory buffer.
| Parameter | Type | Description |
|-----------|------|-------------|
| `item` | `str` | The action/event string to record |
**Behavior:**
1. Validates that `item` is a string (raises `ValueError` otherwise)
2. If capacity is reached, removes the oldest entry from index 0
3. Attaches a timestamp in `YYYY-MM-DD HH:MM:SS` format
4. Appends the tuple `(item, timestamp)` to the queue
5. Logs the entry to the file system
```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) # Remove the oldest item to maintain capacity
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.queue.append((item, timestamp))
self.log_to_file(item, timestamp)
```
资料来源：[examples/bilibili_surfer/fifo.py:16-23](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/fifo.py)
##### `prompt() -> str`
Generates a formatted string of all memory entries for LLM consumption.
**Returns:** A newline-separated string with entries in reverse chronological order (most recent first), formatted as `{time} - {text}`.
```python
def prompt(self):
return "\n".join([f"{time} - {text}" for text, time in reversed(self.queue)])
```
资料来源：[examples/bilibili_surfer/fifo.py:30-31](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/fifo.py)
##### `log_to_file(item: str, timestamp: str) -> None`
Appends a timestamped entry to the persistent log file.
```python
def log_to_file(self, item, timestamp):
with open(self.log_file, 'a') as file:
file.write(f"{timestamp} - {item}\n")
```
资料来源：[examples/bilibili_surfer/fifo.py:25-28](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/fifo.py)
## Integration with State Machine
The `Fifo` memory is integrated into the agent architecture through the `BilibiliStateBase` abstract base class, which composes the memory buffer with other required components.
```mermaid
graph LR
A[BilibiliStateBase] --> B[Fifo Memory]
A --> C[Cohere Client]
A --> D[Bilibili Credential]
A --> E[Initial Prompt]
F[_infer_state_helper] --> B
G[BrowsingVideo] --> A
H[ReadingComments] --> A
I[PostComment] --> A
```
**Base Class Composition:**
```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:29-35](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py)
## Usage Patterns
### Adding Memory Entries
Agents record actions by calling `memory.add()` with descriptive strings:
```python
# Track search operations
self.memory.add(f"searched for {response} while browsing video")
# Record comment discovery
self.memory.add(f"finds comment: {cmt['content']['message']} while browsing {self.video_title}")
# Log posting activity
self.memory.add(f"commented {response} to {self.video_title}")
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:88](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py)
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:108](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py)
### Including Memory in LLM Prompts
The `_infer_state_helper` method constructs prompts that include memory context:
```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.
"""
)
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:38-49](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py)
### Initialization
Fifo is instantiated empty and passed to the initial state:
```python
initial_state = BrowsingVideo(memory=Fifo(),
initial_prompt=initial_prompt,
co=Client(os.environ.get("COHERE_API_KEY")),
credential=Credential(...))
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:135-142](https://github.com/xingjianll/cyclic-agent/blob/main/examples/bilibili_surfer/bilibili_surfer.py)
## Behavior Specifications
| Aspect | Specification |
|--------|---------------|
| **Eviction Policy** | Oldest entry removed when capacity (100) is exceeded |
| **Ordering** | Queue maintains insertion order; prompt reverses for recency-first display |
| **Persistence** | All additions logged to `fifo_log.txt` in append mode |
| **Type Constraint** | Only string values accepted; raises `ValueError` for non-strings |
| **Timestamp Format** | ISO-style: `YYYY-MM-DD HH:MM:SS` |
## Limitations and Considerations
1. **Fixed Capacity**: The 100-item limit is hardcoded and cannot be configured per-instance without modifying source code.
2. **File I/O on Every Add**: Each memory addition triggers a synchronous file write, which may impact performance in high-frequency scenarios.
3. **No Query Filtering**: The `prompt()` method returns all entries; there is no mechanism to retrieve a subset by time range or pattern.
4. **Shared Log File**: All instances write to `fifo_log.txt` by default; concurrent access from multiple processes may cause race conditions.
---
## Hello World Example
### 相关页面
相关主题：[Installation and Setup](#page-1), [Creating Custom States](#page-6), [Bilibili Surfer Example](#page-9)
Relevant Source Files
以下源码文件用于生成本页说明：
- [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)
# Hello World Example
## Overview
The Hello World Example (`examples/hello_world/hello_world.py`) is the simplest demonstration of the CyclicAgent framework. It showcases the core state machine pattern by creating two mutually transitioning states that simulate a continuous question-answer loop using a Cohere LLM. 资料来源：[examples/hello_world/hello_world.py:1-31]()
## Architecture
CyclicAgent abstracts an agent as a **Finite State Machine (FSM)** using the State design pattern. Each state implements a `next()` method that returns another state, enabling indefinite chaining of transitions. 资料来源：[README.md](https://github.com/xingjianll/cyclic-agent#how-does-cyclicagent-work)
### State Machine Diagram
```mermaid
graph TD
A((Start)) --> B[AskQuestion State]
B -->|next()| C[AnswerQuestion State]
C -->|next()| B
B -->|5 iterations| D((End))
C -->|5 iterations| D
```
## Core Components
### State Base Class
The `State` class is the foundation of the framework, defined as a generic Pydantic model:
```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:5-9]()
| Property | Type | Description |
|----------|------|-------------|
| `SigT` | Generic TypeVar | Signal type for state communication |
| `next()` | Abstract Method | Returns the next state in the FSM |
### CyclicExecutor
The executor manages the state machine lifecycle:
```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-8]()
| Method | Description |
|--------|-------------|
| `start(initial_state)` | Begins executing the FSM in a separate thread |
| `pause()` | Pauses execution (thread-safe) |
| `resume()` | Resumes execution (thread-safe) |
| `kill()` | Stops execution and joins the thread |
## Example Implementation
### AskQuestion State
```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)
```
资料来源：[examples/hello_world/hello_world.py:13-18]()
| Field | Type | Description |
|-------|------|-------------|
| Signal Type | `None` | This state does not accept external signals |
| Return Type | `AnswerQuestion` | Transitions to the answer state with the question |
### AnswerQuestion State
```python
class AnswerQuestion(State[None]):
question: str
def next(self, signal: None = None) -> AskQuestion:
answer = co.chat(message=self.question)
print(answer)
return AskQuestion()
```
资料来源：[examples/hello_world/hello_world.py:20-25]()
| Field | Type | Description |
|-------|------|-------------|
| `question` | `str` | Stored question from previous state |
| Signal Type | `None` | Does not accept external signals |
| Return Type | `AskQuestion` | Transitions back to ask a new question |
## Execution Flow
```mermaid
sequenceDiagram
participant Main
participant Executor
participant AskQuestion
participant AnswerQuestion
participant CohereAPI
Main->>Executor: start(AskQuestion())
Executor->>AskQuestion: next()
AskQuestion->>CohereAPI: chat("Ask a question")
CohereAPI-->>AskQuestion: question_text
AskQuestion->>AnswerQuestion: next(question=...)
loop 5 times
Executor->>AnswerQuestion: next()
AnswerQuestion->>CohereAPI: chat(question)
CohereAPI-->>AnswerQuestion: answer
AnswerQuestion->>AskQuestion: next()
Executor->>AskQuestion: next()
AskQuestion->>CohereAPI: chat("Ask a question")
CohereAPI-->>AskQuestion: new_question
AskQuestion->>AnswerQuestion: next(question=...)
end
```
## Configuration
### Environment Setup
The example requires a `.env` file with the Cohere API key:
```bash
COHERE_API_KEY=your_api_key_here
```
### Executor Configuration
```python
executor = CyclicExecutor(5) # 5-second interval between state transitions
```
资料来源：[examples/hello_world/hello_world.py:28-29]()
| Parameter | Value | Description |
|-----------|-------|-------------|
| `default_time_interval` | `5` (seconds) | Pause between each `next()` call |
| `time.sleep(20)` | `20` (seconds) | Total runtime before natural exit |
## Key Takeaways
1. **State Pattern**: Each state is a Pydantic model that implements `next()`, returning the subsequent state 资料来源：[cyclic_agent/state.py:5-9]()
2. **Thread Safety**: The executor uses `threading.Lock` for safe pause/resume/kill operations 资料来源：[cyclic_agent/executor.py:4]()
3. **Type Safety**: Generic type parameters (`State[SigT]`) ensure compile-time safety for signal types 资料来源：[cyclic_agent/state.py:5]()
4. **Extensibility**: New states can be added by inheriting from `State` and implementing the `next()` method
## Running the Example
```bash
# Install dependencies
pip install cyclic-agent cohere python-dotenv
# Set up environment
export COHERE_API_KEY=your_key
# Run
python examples/hello_world/hello_world.py
```
The example will print alternating questions and answers for approximately 20 seconds before exiting.
---
## Bilibili Surfer Example
### 相关页面
相关主题：[Creating Custom States](#page-6), [Memory Management with Fifo](#page-7), [Hello World Example](#page-8)
相关源码文件
以下源码文件用于生成本页说明：
- [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 Example
The Bilibili Surfer is a demonstration application built on the CyclicAgent framework that creates an autonomous AI agent capable of browsing Bilibili, a Chinese video-sharing platform. The agent autonomously searches for videos, reads comments, and interacts by posting replies, simulating a realistic user experience with automatic disclosure as a bot.
## Architecture Overview
The Bilibili Surfer exemplifies the CyclicAgent's finite state machine (FSM) design pattern. The agent operates as a cyclical state machine where each state implements a `next()` method that returns the subsequent state, enabling indefinite chaining of state transitions.
```mermaid
graph TD
A([BrowsingVideo]) -->|select video| B([ReadingComments])
A -->|search again| A
B -->|select comment| C([PostComment])
B -->|back to browse| A
C -->|complete| A
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:1-150]()
## State Components
### BilibiliStateBase (Abstract Base)
All Bilibili states inherit from `BilibiliStateBase`, which combines the `State` abstract class with common attributes required across all states.
| Attribute | Type | Description |
|-----------|------|-------------|
| `memory` | `Fifo` | Queue storing past agent actions with timestamps |
| `co` | `Client` | Cohere API client for LLM interactions |
| `credential` | `Credential` | Bilibili API authentication credentials |
| `initial_prompt` | `str` | System prompt defining agent persona |
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:35-43]()
The base class provides the `_infer_state_helper()` method that constructs prompts for LLM-based state inference:
```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.
"""
)
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:45-58]()
### BrowsingVideo State
The entry point state where the agent searches for and selects videos to watch.
**Type Definition:**
```python
type BrowsingVideoReachable = Union[BrowsingVideo, ReadingComments]
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:61-62]()
**Workflow:**
1. Generate a search keyword phrase via LLM (max 3 Chinese words)
2. Execute Bilibili video search ordered by fan count
3. Present top 10 results to LLM for selection
4. Log selection to memory
5. Transition to `ReadingComments` or remain in `BrowsingVideo`
```python
def next(self, signal: None = None) -> BrowsingVideoReachable:
print('BrowsingVideo')
prompt = I(
f"""
{self.initial_prompt}
Here are your past actions {self.memory.prompt()} Generate a keyword phrase for videos you want to watch.
Be creative and avoid repeating. Respond with a maximum of three words in Chinese.
"""
)
response = self.co.chat(temperature=1, message=prompt).text
res = asyncio.run(search.search_by_type(response,
search_type=search.SearchObjectType.VIDEO,
order_type=search.OrderUser.FANS,
order_sort=0
)
)
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:64-79]()
### ReadingComments State
Enables the agent to examine and select comments from a specific video for potential interaction.
**Type Definition:**
```python
type ReadingCommentsReachable = Union[BrowsingVideo, ReadingComments, PostComment]
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:118-119]()
**Required Attributes:**
| Attribute | Type | Description |
|-----------|------|-------------|
| `video_bvid` | `str` | Bilibili video BV identifier |
| `video_title` | `str` | Video title for context |
| `video_description` | `str` | Video description for context |
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:121-126]()
**Workflow:**
1. Fetch top comments ordered by likes via Bilibili API
2. Present comments to LLM for selection
3. Based on LLM decision, either:
- Return to `BrowsingVideo`
- Stay in `ReadingComments`
- Transition to `PostComment` with reply context
```python
match self._infer_state_helper('BrowsingVideo', 'ReadingComments', 'PostComment'):
case 'BrowsingVideo':
return BrowsingVideo(**self.model_dump(exclude={'video_bvid', 'video_title', 'video_description'}))
case 'ReadingComments':
return self
case 'PostComment':
return PostComment(**self.model_dump(), reply_to=cmt['content']['message'], reply_to_oid=cmt['oid'])
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:141-148]()
### PostComment State
Handles automatic reply generation and posting to Bilibili.
**Type Definition:**
```python
type PostCommentReachable = Union[BrowsingVideo]
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:93-94]()
**Attributes:**
| Attribute | Type | Description |
|-----------|------|-------------|
| `reply_to` | `str \| None` | Original comment content being replied to |
| `reply_to_oid` | `int \| None` | Original comment OID for threading |
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:96-99]()
**Workflow:**
1. Generate reply content via LLM based on:
- Original comment being replied to
- Video context (title, description)
2. Append automatic disclosure footnote
3. Post comment to video via Bilibili API
4. Create and publish a Bilibili "Dynamic" (post) with:
- Video link
- Reply content
- Disclosure notice
5. Return to `BrowsingVideo` state
```python
response = self.co.chat(temperature=1, message=prompt).text
footnote = (
f"\n I am a bot, and this action was performed automatically. Please contact {os.getenv('name')}"
f" if you have any questions or concerns."
)
asyncio.run(comment.send_comment(text=f"{response} {footnote}",
oid=video.Video(bvid=self.video_bvid).get_aid(),
type_=CommentResourceType.VIDEO,
credential=self.credential
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:106-117]()
## Memory System: Fifo
The `Fifo` class implements a bounded queue for storing agent action history, providing the agent with contextual awareness of past behaviors.
资料来源：[examples/bilibili_surfer/fifo.py:1-25]()
### Configuration
| Parameter | Default | Description |
|-----------|---------|-------------|
| `capacity` | 100 | Maximum stored actions before oldest are evicted |
| `log_file` | `"fifo_log.txt"` | Persistent log file for action history |
### Core Methods
| Method | Description |
|--------|-------------|
| `add(item: str)` | Appends action with timestamp, evicts oldest if at capacity |
| `prompt()` | Returns formatted history string (newest first) |
```python
def prompt(self):
return "\n".join([f"{time} - {text}" for text, time in reversed(self.queue)])
```
资料来源：[examples/bilibili_surfer/fifo.py:24-25]()
## Execution Model
The `CyclicExecutor` manages the state machine lifecycle in a dedicated thread.
资料来源：[cyclic_agent/executor.py:1-47]()
### Executor Configuration
| Parameter | Type | Description |
|-----------|------|-------------|
| `default_time_interval` | `float` | Seconds between state transitions |
| `running` | `bool` | Execution status flag |
| `killed` | `bool` | Permanent stop signal |
### Control Methods
| Method | Description |
|--------|-------------|
| `start(initial_state)` | Launch executor thread with initial state |
| `pause()` | Suspend state transitions |
| `resume()` | Continue suspended execution |
| `kill()` | Stop execution and join thread |
```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:36-44]()
## Data Flow Diagram
```mermaid
sequenceDiagram
participant User
participant Executor
participant LLM as Cohere LLM
participant Bilibili as Bilibili API
participant Memory as Fifo Memory
Executor->>BrowsingVideo: start(initial_state)
BrowsingVideo->>LLM: Generate search query
BrowsingVideo->>Bilibili: search.search_by_type()
Bilibili-->>BrowsingVideo: Top 10 videos
BrowsingVideo->>Memory: add(finds {title})
BrowsingVideo->>ReadingComments: next()
ReadingComments->>Bilibili: comment.get_comments()
Bilibili-->>ReadingComments: Top 5 comments
ReadingComments->>Memory: add(find comment)
ReadingComments->>PostComment: next(reply_to)
PostComment->>LLM: Generate reply
PostComment->>Memory: add(commented)
PostComment->>Bilibili: comment.send_comment()
PostComment->>Bilibili: dynamic.send_dynamic()
PostComment->>BrowsingVideo: next()
```
## Setup and Usage
### Prerequisites
Required environment variables:
| Variable | Description |
|----------|-------------|
| `COHERE_API_KEY` | Cohere API key for LLM access |
| `SESSDATA` | Bilibili session data cookie |
| `BILI_JCT` | Bilibili CSRF token |
| `BUVID3` | Bilibili user identifier cookie |
| `name` | Contact name for bot disclosure |
### Running the Example
```python
if __name__ == "__main__":
initial_prompt = "You are a dude browsing Bilibili, A Chinese video sharing platform."
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) # 5-second intervals
executor.start(initial_state)
time.sleep(1000) # Run for ~16 minutes
```
资料来源：[examples/bilibili_surfer/bilibili_surfer.py:163-180]()
## State Transition Matrix
| Current State | Action | Next State |
|---------------|--------|------------|
| `BrowsingVideo` | Select video | `ReadingComments` |
| `BrowsingVideo` | Search again | `BrowsingVideo` |
| `ReadingComments` | Choose comment | `PostComment` |
| `ReadingComments` | Browse more | `BrowsingVideo` |
| `ReadingComments` | Stay | `ReadingComments` |
| `PostComment` | Complete | `BrowsingVideo` |
## Integration with Core Framework
The Bilibili Surfer demonstrates CyclicAgent's extensibility:
1. **State Pattern**: Each state extends `State[None]`, implementing the `next()` abstract method
- 资料来源：[cyclic_agent/state.py:1-15]()
2. **Type Annotations**: Python 3.12+ type unions define reachable states for type safety
- 资料来源：[cyclic_agent/__init__.py:1-6]()
3. **Pydantic Models**: States use `BaseModel` for automatic serialization and validation
- 资料来源：[cyclic_agent/state.py:5]()
4. **Dependency Injection**: States receive dependencies via constructor, enabling testability
---
## Helper States
### 相关页面
相关主题：[State Base Class](#page-3), [Creating Custom States](#page-6)
相关源码文件
以下源码文件用于生成本页说明：
- [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)
- [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)
# Helper States
## Overview
Helper States are reusable, composable state components in CyclicAgent that encapsulate common patterns for LLM-powered agent workflows. They serve as building blocks that developers can extend, combine, or integrate into custom state machines to accelerate agent development.
Unlike application-specific states (e.g., `AskQuestion` or `BrowsingVideo`), helper states focus on generic operational patterns such as search, chain-of-thought reasoning, and callback-driven transitions. They follow the same finite state machine (FSM) architecture as all states in CyclicAgent, ensuring seamless integration with the `CyclicExecutor`.
资料来源：[cyclic_agent/__init__.py:1-6]()
## Architecture
Helper states inherit from the base `State` class and implement the `next()` method to define transitions. They typically leverage callbacks for flexible exit behavior, allowing parent states to control the flow after the helper completes its operation.
```mermaid
graph TD
A[Parent State] -->|creates| B[Helper State]
B -->|executes operation| C[Result/Action]
C -->|callback| D[Exit Callback]
D -->|returns| A[Parent State or Next State]
```
### Base State Contract
All states, including helpers, must implement:
```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-9]()
## Chain of Thought (CoT) Helper
### Purpose
The `CoT` class provides a reusable wrapper for chain-of-thought reasoning patterns. It automatically appends "Let's think step by step." to prompts, enabling step-by-step reasoning in LLM responses.
资料来源：[cyclic_agent/cot.py:1-5]()
### Class Definition
```python
class CoT(State[None]):
exit_: Callable[[str], State]
llm: State
prompt: str
```
### Parameters
| Parameter | Type | Description |
|-----------|------|-------------|
| `exit_` | `Callable[[str], State]` | Callback invoked with the LLM's response; returns the next state |
| `llm` | `State` | The LLM state used to generate responses |
| `prompt` | `str` | The base prompt to which chain-of-thought suffix is added |
### Implementation
```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)
```
资料来源：[cyclic_agent/cot.py:7-14]()
### Usage Pattern
The `CoT` helper demonstrates a callback-driven transition pattern where:
1. The helper augments the prompt with reasoning instructions
2. The LLM processes the enhanced prompt
3. The result is passed to the exit callback
4. Control returns to the calling state or transitions to a new state
```mermaid
sequenceDiagram
participant P as Parent State
participant C as CoT Helper
participant L as LLM State
participant E as Exit Callback
P->>C: Creates CoT with prompt, llm, exit_
C->>C: Appends "Let's think step by step."
C->>L: Calls llm with enhanced prompt
L-->>C: Returns reasoning answer
C->>E: Invokes callback with answer
E-->>P: Returns next State
```
## Search Helper
### Purpose
The `Search` class provides a generic search operation pattern. It accepts a query, executes a search (to be implemented by subclass), and passes results to an exit callback for further processing.
资料来源：[cyclic_agent/search.py:1-5]()
### Class Definition
```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
```
### Parameters
| Parameter | Type | Description |
|-----------|------|-------------|
| `query` | `str` | The search query string |
| `exit_` | `Callable[[list[str]], State]` | Callback receiving search results; returns the next state |
### Key Design Patterns
1. **Abstract Search Method**: The `search()` method raises `NotImplementedError`, requiring subclasses to provide concrete implementations.
2. **Typed Callback**: The exit callback uses `Annotated` to document that it receives search results as a list of strings.
3. **Separation of Concerns**: Search logic is decoupled from state transition logic.
资料来源：[cyclic_agent/search.py:6-12]()
### Subclass Implementation Example
```python
class VideoSearch(Search):
def search(self, query: str) -> str:
# Concrete implementation
results = video_api.search(query)
return formatted_results
```
## Callback-Driven Transition Pattern
Both helper states employ a consistent callback pattern for state transitions:
```mermaid
graph LR
A[Helper State] --> B{Operation Complete}
B --> C[Exit Callback]
C --> D[Next State]
D --> E[CyclicExecutor]
E --> A
```
### Benefits
| Benefit | Description |
|---------|-------------|
| **Decoupling** | Helper states don't need to know about subsequent states |
| **Composability** | Same helper can be reused with different callbacks |
| **Flexibility** | Callbacks can implement complex routing logic |
| **Testability** | Callbacks can be easily mocked for unit testing |
## Integration with CyclicExecutor
Helper states integrate seamlessly with the executor:
```python
class CyclicExecutor:
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:26-32]()
The executor continuously calls `next()` on any state, including helpers. Each helper's `next()` either:
- Returns another state (transition)
- Returns the same state (loop continuation)
## Creating Custom Helper States
### Template
```python
from cyclic_agent import State
class MyHelper(State[None]):
data: str
exit_: Callable[[ResultType], State]
def next(self, signal: None = None) -> State:
# Perform helper operation
result = self.perform_operation()
# Delegate to callback for next state
return self.exit_(result)
def perform_operation(self) -> ResultType:
raise NotImplementedError
```
### Checklist
- [ ] Inherit from `State[None]`
- [ ] Define exit callback as `Callable` parameter
- [ ] Implement `next()` method returning `State`
- [ ] Call exit callback with operation result
- [ ] Raise `NotImplementedError` for subclass-specific methods
## Comparison: CoT vs Search Helpers
| Aspect | CoT | Search |
|--------|-----|--------|
| **Primary Function** | Reasoning wrapper | Search operation |
| **Exit Callback Parameter** | `Callable[[str], State]` | `Callable[[list[str]], State]` |
| **Abstract Methods** | None | `search(query: str)` |
| **Prompt Modification** | Appends reasoning suffix | None |
| **Use Case** | Step-by-step reasoning | Information retrieval |
## Export Configuration
Helper states are available via the cyclic_agent package:
```python
from cyclic_agent import State, CyclicExecutor
__all__ = ["State", "CyclicExecutor"]
```
Direct imports for specific helpers:
```python
from cyclic_agent.cot import CoT
from cyclic_agent.search import Search
```
资料来源：[cyclic_agent/__init__.py:1-6]()
## Best Practices
1. **Use Helpers as Building Blocks**: Compose helpers within custom states rather than subclassing them.
2. **Type Your Callbacks**: Always specify callback signatures for type safety.
3. **Handle Edge Cases**: Implement error handling in `next()` for production use.
4. **Document State Transitions**: Use type aliases (e.g., `ReachableStates`) to document possible transitions.
5. **Maintain State Immutability**: Helper states should not mutate shared state; use callbacks for side effects.
---
---
## Doramagic Pitfall Log
Project: xingjianll/cyclic-agent
Summary: Found 6 potential pitfall items; 0 are high/blocking. Highest priority: capability - 能力判断依赖假设.
## 1. capability · 能力判断依赖假设
- Severity: medium
- Evidence strength: source_linked
- Finding: README/documentation is current enough for a first validation pass.
- User impact: 假设不成立时，用户拿不到承诺的能力。
- Suggested check: 将假设转成下游验证清单。
- Guardrail action: 假设必须转成验证项；没有验证结果前不能写成事实。
- Evidence: capability.assumptions | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | README/documentation is current enough for a first validation pass.
## 2. maintenance · 维护活跃度未知
- Severity: medium
- Evidence strength: source_linked
- Finding: 未记录 last_activity_observed。
- User impact: 新项目、停更项目和活跃项目会被混在一起，推荐信任度下降。
- Suggested check: 补 GitHub 最近 commit、release、issue/PR 响应信号。
- Guardrail action: 维护活跃度未知时，推荐强度不能标为高信任。
- Evidence: evidence.maintainer_signals | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | last_activity_observed missing
## 3. security_permissions · 下游验证发现风险项
- Severity: medium
- Evidence strength: source_linked
- Finding: no_demo
- User impact: 下游已经要求复核，不能在页面中弱化。
- Suggested check: 进入安全/权限治理复核队列。
- Guardrail action: 下游风险存在时必须保持 review/recommendation 降级。
- Evidence: downstream_validation.risk_items | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | no_demo; severity=medium
## 4. security_permissions · 存在评分风险
- Severity: medium
- Evidence strength: source_linked
- Finding: no_demo
- User impact: 风险会影响是否适合普通用户安装。
- Suggested check: 把风险写入边界卡，并确认是否需要人工复核。
- Guardrail action: 评分风险必须进入边界卡，不能只作为内部分数。
- Evidence: risks.scoring_risks | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | no_demo; severity=medium
## 5. maintenance · issue/PR 响应质量未知
- Severity: low
- Evidence strength: source_linked
- Finding: issue_or_pr_quality=unknown。
- User impact: 用户无法判断遇到问题后是否有人维护。
- Suggested check: 抽样最近 issue/PR，判断是否长期无人处理。
- Guardrail action: issue/PR 响应未知时，必须提示维护风险。
- Evidence: evidence.maintainer_signals | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | issue_or_pr_quality=unknown
## 6. maintenance · 发布节奏不明确
- Severity: low
- Evidence strength: source_linked
- Finding: release_recency=unknown。
- User impact: 安装命令和文档可能落后于代码，用户踩坑概率升高。
- Suggested check: 确认最近 release/tag 和 README 安装命令是否一致。
- Guardrail action: 发布节奏未知或过期时，安装说明必须标注可能漂移。
- Evidence: evidence.maintainer_signals | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | release_recency=unknown