第二章 Agent 核心循环

2.1 AIAgent 类设计

AIAgent 是 Hermes Agent 的核心类，定义在 run_agent.py 中。它封装了完整的对话循环——从接收用户消息到调用 LLM、执行工具、组装响应的全过程。

构造函数

class AIAgent:
    def __init__(
        self,
        model: str = "gpt-4o",
        base_url: str = None,
        api_key: str = None,
        max_iterations: int = 90,
        enabled_toolsets: list = None,
        platform: str = "cli",
        session_id: str = None,
        callbacks: dict = None,
        **kwargs
    ):
        ...

关键参数说明

参数	默认值	说明
model	"gpt-4o"	模型标识符，支持 provider 前缀如 anthropic/claude-sonnet-4-20250514
max_iterations	90	单次对话最大迭代次数（防止无限循环）
enabled_toolsets	None	启用的工具集列表，None 表示使用默认集
platform	"cli"	运行平台标识：cli / telegram / discord / slack 等
session_id	None	会话 ID，用于持久化和上下文管理
callbacks	None	回调函数字典（stream_delta, tool_start 等）

两种调用接口

# 简单接口 — 返回纯文本字符串
response: str = agent.chat("解释 Python 的 GIL")

# 完整接口 — 返回详细信息字典
result: dict = agent.run_conversation("分析这段代码的性能")
# result 包含: response, tool_calls, usage, reasoning 等

2.2 主循环详解

run_conversation() 是 Agent 的核心方法，位于 run_agent.py。它实现了一个迭代式工具调用循环——每次 LLM 返回的工具调用都会被执行，结果附加到消息历史中，然后继续调用 LLM，直到不再产生工具调用或达到预算上限。

循环步骤

伪代码表示

def run_conversation(self, user_message: str) -> dict:
    # 1. 输入清洗
    user_message = sanitize_input(user_message)

    # 2. 构建系统提示词
    system_prompt = self._build_system_prompt()

    # 3. 预压缩（如果上下文接近限制）
    self._compress_if_needed()

    # 4. 追加用户消息
    self.messages.append({"role": "user", "content": user_message})

    # 5. 主循环
    while self._budget.check():
        if self._check_interrupt():
            return self._handle_interrupt()

        api_messages = self._build_api_messages()

        response = self._call_llm(api_messages)

        if response.tool_calls:
            for tool_call in response.tool_calls:
                result = self._dispatch_tool_call(tool_call)
                self.messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result
                })
        else:
            return self._assemble_response(response)

    return self._budget_exceeded_response()

2.3 消息格式

Hermes Agent 使用兼容 OpenAI 的消息格式，所有模型提供商的响应都会被规范化为此格式。

消息角色

角色	说明	示例
system	系统提示词，由 PromptBuilder 组装	{role: "system", content: "你是 Hermes..."}
user	用户输入消息	{role: "user", content: "帮我写一个函数"}
assistant	LLM 响应，可能包含 tool_calls	{role: "assistant", content: "...", tool_calls: [...]}
tool	工具执行结果	{role: "tool", tool_call_id: "xxx", content: "执行结果"}

工具调用格式

{
  "role": "assistant",
  "content": null,
  "tool_calls": [
    {
      "id": "call_abc123",
      "type": "function",
      "function": {
        "name": "terminal",
        "arguments": "{\"command\": \"ls -la\"}"
      }
    }
  ]
}

Reasoning（推理链）

对于支持推理的模型（如 Claude 的 extended thinking），推理内容存储在 assistant_msg["reasoning"] 中：

{
  "role": "assistant",
  "content": "根据分析，我建议...",
  "reasoning": "首先需要检查文件是否存在，然后..."
}

2.4 迭代预算

IterationBudget 类控制 Agent 的迭代次数，防止无限循环消耗资源。

核心机制

class IterationBudget:
    def __init__(self, max_iterations: int = 90):
        self.max_iterations = max_iterations
        self._iterations_used = 0
        self._interrupt_requested = False
        self._grace_call_used = False

    def check(self) -> bool:
        """检查是否还有预算继续迭代"""
        if self._interrupt_requested:
            return False
        if self._iterations_used >= self.max_iterations:
            if not self._grace_call_used:
                self._grace_call_used = True
                return True  # 给一次额外的 grace call 来收尾
            return False
        return True

Grace Call 机制

当迭代次数达到上限时，Agent 不会立即停止，而是利用 grace call 机制给 LLM 一次额外的调用来生成最终摘要。这确保了：

1. 用户总能得到一个有意义的响应
2. 中间状态不会被直接丢弃
3. 长任务可以优雅地收尾

2.5 API 重试

Hermes Agent 实现了完善的 exponential backoff（指数退避） 重试机制。

错误分类

退避策略

# 伪代码 — 实际实现在 retry_utils.py
async def call_with_retry(api_func, max_retries=5):
    for attempt in range(max_retries):
        try:
            return await api_func()
        except RateLimitError:
            wait = base_delay * (2 ** attempt) + random_jitter()
            await asyncio.sleep(wait)
        except ContextOverflowError:
            await compress_context()
            continue  # 压缩后立即重试，不计算在重试次数内
        except AuthError:
            raise  # 认证错误不重试

2.6 流式响应

Hermes Agent 支持流式响应，通过 stream_delta_callback 实现逐 token 输出。

流式处理流程

流式集成点

组件	回调	用途
TUI (Ink)	stream_delta_callback	逐字渲染到终端
Gateway	stream_delta_callback	通过 WebSocket/SSE 推送到消息平台
KawaiiSpinner	内部集成	在等待时显示动画，收到 delta 后切换为文本流

KawaiiSpinner

KawaiiSpinner 是 Hermes Agent 的标志性等待动画，在 LLM 思考和工具执行期间显示可爱好看的动画效果。当流式 delta 到达时，Spinner 自动切换为文本渲染模式。

# 回调集成示例
def on_stream_delta(delta_text: str):
    """每收到一个 delta token 就调用"""
    spinner.stop()
    print(delta_text, end="", flush=True)

agent = AIAgent(
    callbacks={"stream_delta": on_stream_delta}
)

---

上一章：第一章 认识 Hermes Agent下一章：第三章 工具系统