Related Pages

Related topics: State Base Class, Hello World Example

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.

Sources: README.md:1-8

Prerequisites

Before installing CyclicAgent, ensure your environment meets the following requirements:

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

Sources: cyclic_agent/state.py:1-4

Installation Methods

Standard Installation (Recommended)

Install the package directly from PyPI:

pip install cyclic-agent

Sources: README.md:7-9

Development Installation

For contributing or modifying the source code:

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.

Sources: pyproject.toml

Project Configuration

Environment Variables Setup

Create a .env file in your project root to store sensitive credentials:

COHERE_API_KEY=your_api_key_here

#### Example .env Configuration

# 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

Sources: examples/bilibili_surfer/bilibili_surfer.py:1-20

Core Package Imports

After installation, import the essential components:

from cyclic_agent import State, CyclicExecutor

Sources: cyclic_agent/__init__.py:1-6

Available Exports

State Machine Architecture

The framework uses a state design pattern where each state implements a next() method returning the subsequent state.

graph TD
    A[Initial State] -->|next()| B[State 1]
    B -->|next()| C[State 2]
    C -->|next()| D[State N]
    D -->|next()| A

Sources: README.md:10-16

Creating Your First Agent

Basic Setup Pattern

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"))

Sources: examples/hello_world/hello_world.py:1-16

Implementing States

Each state must inherit from State and implement the next() method:

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)

Sources: examples/hello_world/hello_world.py:18-24

State Base Class Requirements

The State class is a Pydantic BaseModel with an abstract method:

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

Sources: cyclic_agent/state.py:6-12

Running the Executor

Basic Execution

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

Sources: examples/hello_world/hello_world.py:34-40

Executor Configuration

Sources: cyclic_agent/executor.py:4-10

Executor Control Methods

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:

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()

Sources: cyclic_agent/executor.py:4-25

Advanced Setup: Memory Integration

For agents requiring persistent memory, extend the base state class:

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

Sources: examples/bilibili_surfer/bilibili_surfer.py:28-35

Memory Class Implementation

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()))

Sources: 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

Sources: cyclic_agent/state.py:1-2, examples/bilibili_surfer/bilibili_surfer.py:30

Related Pages

Related topics: State Base Class, CyclicExecutor

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). Sources: 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. Sources: 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:

Sources: 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". Sources: README.md:10-11

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:

Sources: cyclic_agent/__init__.py:1-6

Class Hierarchy

classDiagram
    class State~SigT~ {
        <<abstract>>
        +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. Sources: cyclic_agent/state.py:1-13

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

Sources: 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. Sources: cyclic_agent/executor.py:1-45

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

Execution Flow:

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]

Sources: 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. Sources: examples/hello_world/hello_world.py:1-38

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:

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. Sources: examples/bilibili_surfer/fifo.py:1-27

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

Sources: examples/bilibili_surfer/bilibili_surfer.py:1-55

State Inference Pattern

States can use LLM-driven inference to determine the next state dynamically:

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

Sources: 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:

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

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

Sources: examples/bilibili_surfer/fifo.py:1-27

Specialized States

Search State

A generic search state that executes queries and transitions based on results:

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)

Sources: 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:

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_)

Sources: cyclic_agent/cot.py:1-15

Complete Workflow Example

The Bilibili Surfer demonstrates a complete autonomous agent workflow:

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:

Sources: examples/bilibili_surfer/bilibili_surfer.py:1-180

Package Exports

The framework exposes a minimal public API:

from cyclic_agent import State, CyclicExecutor

All core functionality is accessible through these two classes, keeping the API simple and intuitive for developers. Sources: cyclic_agent/__init__.py:1-6

Installation

Install the framework via pip:

pip install cyclic-agent

The package is available on PyPI and can be integrated into any Python project. Sources: 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 Sources: 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.

Related Pages

Related topics: CyclicExecutor, Creating Custom States

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.

Sources: cyclic_agent/state.py:6-9

Related Pages

Related topics: State Base Class, Creating Custom States, Helper States

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.

Sources: 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.

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

Sources: cyclic_agent/executor.py:7-14

API Reference

Constructor

def __init__(self, default_time_interval: float)

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.

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.

executor.kill()  # Clean shutdown

Sources: cyclic_agent/executor.py:16-31

Execution Flow

Main Loop Logic

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

Sources: cyclic_agent/executor.py:33-35

Thread Safety

The executor uses a threading.Lock to protect shared state modifications:

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.

Sources: cyclic_agent/executor.py:10

State Integration

State Base Class

The executor expects states to implement the State[SigT] protocol:

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.

Sources: cyclic_agent/state.py:6-9

Type Parameter Usage

The generic type SigT allows states to receive external signals of any type:

Sources: cyclic_agent/state.py:6

Configuration Options

Interval Selection Guidelines

Sources: cyclic_agent/executor.py:12

Usage Examples

Basic Hello World

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()

Sources: examples/hello_world/hello_world.py:1-26

Complex Agent: Bilibili Surfer

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(), ...)

Sources: examples/bilibili_surfer/bilibili_surfer.py:1-95

Typical Execution Pattern

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:

from cyclic_agent import State, CyclicExecutor

__all__ = ["State", "CyclicExecutor"]

Sources: 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

Sources: cyclic_agent/executor.py:18

Related Components

Sources: cyclic_agent/search.py:1-14, cyclic_agent/cot.py:1-18, examples/bilibili_surfer/fifo.py:1-30

Related Pages

Related topics: State Base Class, CyclicExecutor, Memory Management with Fifo

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. Sources: 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. Sources: cyclic_agent/state.py:8

Key Characteristics

Architecture Components

State Base Class

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:

Sources: cyclic_agent/state.py:7-11

CyclicExecutor

The CyclicExecutor manages the runtime execution loop for state machines:

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

Sources: cyclic_agent/executor.py:1-29

Execution Flow

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. Sources: cyclic_agent/executor.py:27-33

State Machine Structure

Typical State Implementation

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

Sources: examples/hello_world/hello_world.py:13-26

State Transitions

graph LR
    A[AskQuestion] -->|next()| B[AnswerQuestion]
    B -->|next()| A

The cyclic nature allows indefinite chaining of states, enabling continuous autonomous operation. Sources: README.md:27-42

Supporting Components

Memory Management (Fifo)

The Fifo class provides persistent memory for states:

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

Sources: examples/bilibili_surfer/fifo.py:1-27

Base State Pattern

Complex states inherit from a base class providing common functionality:

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

Sources: 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:

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)

Sources: cyclic_agent/cot.py:1-16

Search State

The Search state provides a template for search operations:

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

Sources: cyclic_agent/search.py:1-16

Complex State Machine Example

The Bilibili Surfer example demonstrates a multi-state architecture:

graph TD
    A[BrowsingVideo] -->|next| B[ReadingComments]
    A -->|next| A
    B -->|next| C[PostComment]
    B -->|next| A
    C -->|next| A

Sources: examples/bilibili_surfer/bilibili_surfer.py:60-130

Configuration and Threading

Executor Threading Model

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. Sources: cyclic_agent/executor.py:16-18

Time Interval Configuration

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. Sources: 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.

Related Pages

Related topics: State Base Class, Hello World Example, Bilibili Surfer Example

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." Sources: 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.

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

Creating Basic States

Step 1: Define Your State Class

Extend the State class and implement the next() method:

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:

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)

Sources: examples/hello_world/hello_world.py:14-18

Step 3: Create the State Transition Pair

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

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

Sources: examples/hello_world/hello_world.py:21-26

Step 4: Execute the State Machine

if __name__ == "__main__":
    initial_state = AskQuestion()
    executor = CyclicExecutor(5)  # 5-second interval between transitions
    executor.start(initial_state)

Sources: examples/hello_world/hello_world.py:29-33

State API Reference

State Class

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

Sources: cyclic_agent/state.py:1-9

State Data Model

States inherit from Pydantic's BaseModel, providing automatic validation and serialization:

class AnswerQuestion(State[None]):
    question: str  # State-specific attribute

Sources: 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.

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

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

Sources: cyclic_agent/executor.py:6-12

Thread Safety

The executor uses a threading lock to ensure thread-safe state transitions:

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

Sources: 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:

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

Sources: examples/bilibili_surfer/bilibili_surfer.py:16-22

FIFO Memory Implementation

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)])

Sources: examples/bilibili_surfer/fifo.py:1-19

State-Independent Transitions

States can transition to different next states based on internal logic using pattern matching:

match self._infer_state_helper('BrowsingVideo', 'ReadingComments', 'PostComment'):
    case 'BrowsingVideo':
        return BrowsingVideo(...)
    case 'ReadingComments':
        return self
    case 'PostComment':
        return PostComment(...)

Sources: examples/bilibili_surfer/bilibili_surfer.py:58-63

State Inference Helper

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

Sources: 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:

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

Sources: examples/bilibili_surfer/bilibili_surfer.py:13-14

Complete Example: Two-State Loop

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)

Sources: examples/hello_world/hello_world.py:1-33

Best Practices

Forward References

Always use from __future__ import annotations for forward references in type hints:

from __future__ import annotations

State Immutability

Return new state instances rather than modifying existing ones to ensure thread safety:

def next(self, signal: None = None) -> AskQuestion:
    return AskQuestion()  # Return new instance

Signal Usage

Leverage the signal parameter for external state control:

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:

model_config = ConfigDict(arbitrary_types_allowed=True)

Execution Flow

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.

Related Pages

Related topics: System Architecture, Bilibili Surfer Example

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

Sources: examples/bilibili_surfer/fifo.py:1-35

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.

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.

Sources: examples/bilibili_surfer/fifo.py:1-35 Sources: examples/bilibili_surfer/bilibili_surfer.py:38-52

Class Reference

Fifo

The core memory management class.

#### Methods

##### __init__()

Initializes the Fifo instance and ensures the log file exists.

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("")

Sources: examples/bilibili_surfer/fifo.py:6-14

##### add(item: str) -> None

Adds a string item to the memory buffer.

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

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)

Sources: examples/bilibili_surfer/fifo.py:16-23

##### 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}.

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

Sources: examples/bilibili_surfer/fifo.py:30-31

##### log_to_file(item: str, timestamp: str) -> None

Appends a timestamped entry to the persistent log file.

def log_to_file(self, item, timestamp):
    with open(self.log_file, 'a') as file:
        file.write(f"{timestamp} - {item}\n")

Sources: examples/bilibili_surfer/fifo.py:25-28

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.

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:

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

Sources: examples/bilibili_surfer/bilibili_surfer.py:29-35

Usage Patterns

Adding Memory Entries

Agents record actions by calling memory.add() with descriptive strings:

# 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}")

Sources: examples/bilibili_surfer/bilibili_surfer.py:88 Sources: examples/bilibili_surfer/bilibili_surfer.py:108

Including Memory in LLM Prompts

The _infer_state_helper method constructs prompts that include memory context:

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.
        """
    )

Sources: examples/bilibili_surfer/bilibili_surfer.py:38-49

Initialization

Fifo is instantiated empty and passed to the initial state:

initial_state = BrowsingVideo(memory=Fifo(),
                              initial_prompt=initial_prompt,
                              co=Client(os.environ.get("COHERE_API_KEY")),
                              credential=Credential(...))

Sources: examples/bilibili_surfer/bilibili_surfer.py:135-142

Behavior Specifications

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.

Related Pages

Related topics: Installation and Setup, Creating Custom States, Bilibili Surfer Example

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. Sources: 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. Sources: README.md

State Machine Diagram

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:

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

Sources: cyclic_agent/state.py:5-9

CyclicExecutor

The executor manages the state machine lifecycle:

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

Sources: cyclic_agent/executor.py:1-8

Example Implementation

AskQuestion State

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)

Sources: examples/hello_world/hello_world.py:13-18

AnswerQuestion State

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

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

Sources: examples/hello_world/hello_world.py:20-25

Execution Flow

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:

COHERE_API_KEY=your_api_key_here

Executor Configuration

executor = CyclicExecutor(5)  # 5-second interval between state transitions

Sources: examples/hello_world/hello_world.py:28-29

Key Takeaways

1. State Pattern: Each state is a Pydantic model that implements next(), returning the subsequent state Sources: cyclic_agent/state.py:5-9

1. Thread Safety: The executor uses threading.Lock for safe pause/resume/kill operations Sources: cyclic_agent/executor.py:4

1. Type Safety: Generic type parameters (State[SigT]) ensure compile-time safety for signal types Sources: cyclic_agent/state.py:5

1. Extensibility: New states can be added by inheriting from State and implementing the next() method

Running the Example

# 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.

Related Pages

Related topics: Creating Custom States, Memory Management with Fifo, Hello World Example

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.

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

Sources: 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.

Sources: 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:

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.
        """
    )

Sources: 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:

type BrowsingVideoReachable = Union[BrowsingVideo, ReadingComments]

Sources: 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

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
                                            )
                      )

Sources: 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:

type ReadingCommentsReachable = Union[BrowsingVideo, ReadingComments, PostComment]

Sources: examples/bilibili_surfer/bilibili_surfer.py:118-119

Required Attributes:

Sources: 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

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'])

Sources: examples/bilibili_surfer/bilibili_surfer.py:141-148

PostComment State

Handles automatic reply generation and posting to Bilibili.

Type Definition:

type PostCommentReachable = Union[BrowsingVideo]

Sources: examples/bilibili_surfer/bilibili_surfer.py:93-94

Attributes:

Sources: 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)

1. Append automatic disclosure footnote
2. Post comment to video via Bilibili API
3. Create and publish a Bilibili "Dynamic" (post) with:

• Video link
• Reply content
• Disclosure notice

1. Return to BrowsingVideo state

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

Sources: 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.

Sources: examples/bilibili_surfer/fifo.py:1-25

Configuration

Core Methods

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

Sources: examples/bilibili_surfer/fifo.py:24-25

Execution Model

The CyclicExecutor manages the state machine lifecycle in a dedicated thread.

Sources: cyclic_agent/executor.py:1-47

Executor Configuration

Control Methods

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)

Sources: cyclic_agent/executor.py:36-44

Data Flow Diagram

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:

Running the Example

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

Sources: examples/bilibili_surfer/bilibili_surfer.py:163-180

State Transition Matrix

Integration with Core Framework

The Bilibili Surfer demonstrates CyclicAgent's extensibility:

1. State Pattern: Each state extends State[None], implementing the next() abstract method

• Sources: cyclic_agent/state.py:1-15

1. Type Annotations: Python 3.12+ type unions define reachable states for type safety

• Sources: cyclic_agent/__init__.py:1-6

1. Pydantic Models: States use BaseModel for automatic serialization and validation

• Sources: cyclic_agent/state.py:5

1. Dependency Injection: States receive dependencies via constructor, enabling testability

Related Pages

Related topics: State Base Class, Creating Custom States

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.

Sources: 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.

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:

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

Sources: 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.

Sources: cyclic_agent/cot.py:1-5

Class Definition

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

Parameters

Implementation

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)

Sources: 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

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.

Sources: cyclic_agent/search.py:1-5

Class Definition

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

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.

Sources: cyclic_agent/search.py:6-12

Subclass Implementation Example

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:

graph LR
    A[Helper State] --> B{Operation Complete}
    B --> C[Exit Callback]
    C --> D[Next State]
    D --> E[CyclicExecutor]
    E --> A

Benefits

Integration with CyclicExecutor

Helper states integrate seamlessly with the executor:

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)

Sources: 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

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

Export Configuration

Helper states are available via the cyclic_agent package:

from cyclic_agent import State, CyclicExecutor

__all__ = ["State", "CyclicExecutor"]

Direct imports for specific helpers:

from cyclic_agent.cot import CoT
from cyclic_agent.search import Search

Sources: 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

Doramagic extracted 6 source-linked risk signals. Review them before installing or handing real data to the project.

1. Capability assumption: README/documentation is current enough for a first validation pass.

• Severity: medium
• Finding: README/documentation is current enough for a first validation pass.
• User impact: The project should not be treated as fully validated until this signal is reviewed.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: capability.assumptions | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | README/documentation is current enough for a first validation pass.

2. Maintenance risk: Maintainer activity is unknown

• Severity: medium
• Finding: Maintenance risk is backed by a source signal: Maintainer activity is unknown. Treat it as a review item until the current version is checked.
• User impact: Users cannot judge support quality until recent activity, releases, and issue response are checked.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: evidence.maintainer_signals | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | last_activity_observed missing

3. Security or permission risk: no_demo

• Severity: medium
• Finding: no_demo
• User impact: The project may affect permissions, credentials, data exposure, or host boundaries.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: downstream_validation.risk_items | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | no_demo; severity=medium

4. Security or permission risk: no_demo

• Severity: medium
• Finding: no_demo
• User impact: The project may affect permissions, credentials, data exposure, or host boundaries.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: risks.scoring_risks | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | no_demo; severity=medium

5. Maintenance risk: issue_or_pr_quality=unknown

• Severity: low
• Finding: issue_or_pr_quality=unknown。
• User impact: Users cannot judge support quality until recent activity, releases, and issue response are checked.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: evidence.maintainer_signals | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | issue_or_pr_quality=unknown

6. Maintenance risk: release_recency=unknown

• Severity: low
• Finding: release_recency=unknown。
• User impact: Users cannot judge support quality until recent activity, releases, and issue response are checked.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: evidence.maintainer_signals | art_2a2d1b4b3cfd487880b7144fa08bc9e2 | https://github.com/xingjianll/cyclic-agent#readme | release_recency=unknown

Community Discussion Evidence

Doramagic exposes project-level community discussion separately from official documentation. Review these links before using cyclic-agent with real data or production workflows.

• AI agents as finite state machine ? : r/LocalLLaMA - Reddit - reddit / searxng_indexed
• README/documentation is current enough for a first validation pass. - GitHub / issue