从零手搓生产级 AI Agent：六个机制拼出单 Agent 底座

语言模型是个很奇怪的东西。

你问它一个问题，它给你一段回答。你让它写代码，它给你一段代码。你让它「帮我重构这个文件」，它告诉你「好的，建议你这样改…」然后列出一堆建议，等你手动复制粘贴。

它不会自己打开文件，不会自己跑命令，不会自己看结果然后决定下一步。它只会做一件事：生成下一段内容。

模型是大脑，但它没有手，也没有记忆。让它从「会说话的程序」变成「会干活的 Agent」，你需要一套机制——它叫 Agent Loop。

这篇文章，我从一个空的 <code>while True</code> 开始，一步步加工具、加持久化、加提示词、加压缩、加容错。六个机制拼完，你就有了一个能自己干活、重启后对话还在、碰到 API 挂了也不崩的生产级单 Agent 底座。

这是 Hermes Agent 教学仓库第一阶段（S01-S06）的全部内容。不追源码细节，只讲核心机制和最小实现。每个机制都按一个套路走：它是什么 → 为什么需要 → 最小实现 → Hermes 做了哪些独特设计。

---

机制一：Agent Loop —— 没有循环，就没有 Agent

它是什么

Agent Loop 就是一段代码，反复做三件事：

1. 把消息发给模型
2. 如果模型说要调工具，就执行工具，把结果塞回消息列表
3. 回到步骤 1，直到模型说「我说完了」

用一句话表达就是：<code>messages → model → tool_calls → tool_result → next turn</code>。

为什么需要

因为模型本身不会「执行→观察→推理」。它只会生成文本。你让它「读一下这个文件然后改」，它不会自己去找文件、打开、读、改。它只会生成一段文字：「建议你打开某某文件，然后这样改…」

循环是让模型动手的底座。没有循环，模型就是一个只能聊天的程序。

最小实现

二十行代码就够了：

from openai import OpenAI

client = OpenAI(base_url="https://openrouter.ai/api/v1", api_key="...")

def run_conversation(user_message, system_prompt, tools, max_iterations=90):
    messages = [{"role": "user", "content": user_message}]

    for i in range(max_iterations):
        # 每次调用都把 system prompt 拼在最前面
        api_messages = [{"role": "system", "content": system_prompt}] + messages

        response = client.chat.completions.create(
            model="anthropic/claude-sonnet-4",
            messages=api_messages,
            tools=tools,
        )

        assistant_msg = response.choices[0].message

        # 把 assistant 回复写回历史
        messages.append({
            "role": "assistant",
            "content": assistant_msg.content,
            "tool_calls": [
                {"id": tc.id, "function": {"name": tc.function.name, "arguments": tc.function.arguments}}
                for tc in (assistant_msg.tool_calls or [])
            ] or None,
        })

        # 没有 tool_calls → 结束
        if not assistant_msg.tool_calls:
            return {"final_response": assistant_msg.content, "messages": messages}

        # 执行每个工具，结果写回
        for tool_call in assistant_msg.tool_calls:
            output = run_tool(tool_call.function.name, tool_call.function.arguments)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": output,
            })

    return {"final_response": "达到最大迭代次数", "messages": messages}

就这么多。核心骨架就五步：拼装消息 → 调 API → 写回 assistant 消息 → 有 tool_calls 就执行 → 继续。

三条规则：

1. 工具结果必须写回消息历史。不写回，模型下一轮看不到执行结果，等于白干。
2. system prompt 每次 API 调用时拼在最前面。它不应该存在 messages 列表里，否则会被持久化、被压缩、被重复。
3. 必须设迭代上限。Hermes Agent 默认 90 次。没有上限，模型可能在某个循环里永远出不来。

Hermes 的独特设计

同步循环 + 异步桥接。 循环本身是普通 <code>def</code>，不是 <code>async def</code>。为什么？因为大部分工具（读文件、写文件、跑命令）是同步的，只有少数（网络请求、浏览器操作）需要 async。如果整个循环都改成 async，所有同步工具也要包一层，堆栈变复杂，调试变困难。Hermes 的选择：循环保持同步，遇到 async 工具时扔给一个持久化的事件循环去执行，执行完把结果拿回来继续。

messages vs api_messages。 Hermes Agent 内部维护了两份消息。<code>messages</code> 是完整账本，什么都有（包括 reasoning 字段、token 计数等内部状态）。<code>api_messages</code> 是每次 API 调用前从 messages 清洗出来的副本，只留模型看得懂的字段。简单说，messages 是底稿，api_messages 是每次寄出去的信件。

实例缓存复用。 Gateway 模式下，不是每条消息都新建一个 AIAgent 实例。实际实现维护了一个缓存，按 <code>(model, api_key, provider, toolsets)</code> 的签名复用实例。这么做的关键原因是 Anthropic 的 prompt caching——system prompt 不变才能命中缓存，复用实例 = 省钱省时间。

---

机制二：Tool System —— 加一个工具，只加一个文件

它是什么

工具系统就是把「模型想调什么工具」和「这个工具怎么执行」解耦的一层。模型说「我要搜网页」，注册表找到 <code>web_search</code> 这个名字，调用对应的 handler 函数，返回结果。

为什么需要

最直觉的实现是一个 if/elif 链：

def run_tool(name, args):
    if name == "web_search":
        return do_web_search(args)
    elif name == "read_file":
        return do_read_file(args)
    elif name == "terminal":
        return do_terminal(args)
    # ... 每加一个工具加一个 elif

Hermes Agent 有 50+ 内置工具，还有 MCP 外部工具可以动态加入。每加一个工具就要改这个分发函数，系统很快就失控了。

最小实现

Hermes Agent 把工具系统分成三层，靠导入链连接：

registry.py          (不导入任何工具)
     ^
tools/*.py           (导入 registry，注册自己)
     ^
model_tools.py       (导入所有 tools/*.py，触发注册)
     ^
run_agent.py         (导入 model_tools.py，使用接口)

关键点：注册表在最底层，不依赖任何工具。这是整个设计能工作的前提。

每个工具文件长这样：

# tools/web_tools.py
from tools.registry import registry

def handle_web_search(args, **kwargs):
    query = args.get("query", "")
    # ... 执行搜索 ...
    return json.dumps({"results": [...]})

registry.register(
    name="web_search",
    toolset="web",
    schema={"name": "web_search", "description": "Search the web", "parameters": {...}},
    handler=handle_web_search,
    is_async=True,
    requires_env=["SERP_API_KEY"],
)

编排层做的事很简单——把所有工具模块导入一遍就行了：

# model_tools.py
_modules = [
    "tools.web_tools",
    "tools.terminal_tool",
    "tools.file_tools",
    "tools.vision_tools",
    # ... 20+ 模块
]
for mod in _modules:
    importlib.import_module(mod)

Python 的 <code>import</code> 会把模块里的顶层代码执行一遍——每个工具文件末尾的 <code>registry.register(…)</code> 就是顶层代码。所以导入 = 注册。加一个新工具只需要往列表里加一行字符串，循环不用动，注册表不用改，编排层不用改。

Hermes 的独特设计

is_async 标记 + 持久化事件循环。 工具注册时声明自己是同步还是异步的。注册表的 <code>dispatch()</code> 看到 <code>is_async=True</code> 就自动走异步桥接。工具文件和核心循环都不需要关心这个细节。

toolset 开关 + check_fn。 工具可以带一个 <code>check_fn</code>，比如检查环境变量有没有 API key。没有 key 的工具不会出现在给模型的 schema 列表里——模型根本不知道这个工具存在，自然不会调用它。

MCP 外部工具也注册进同一个注册表。 编排层在导入内置工具后，还会调用 <code>discover_mcp_tools()</code> 发现 MCP 外部工具，注册进同一个注册表。对循环来说，内置工具和外部工具没有区别，都是 <code>registry.dispatch(name, args)</code>。

---

机制三：Session Store —— SQLite+WAL 不是优化，是 Gateway 的基本需求

它是什么

把 messages 从内存搬到 SQLite。进程退出，对话还在。下次启动，从数据库读回来继续。

为什么需要

到了 s02，Agent 已经能调工具了。但 messages 只在内存里。进程退出，一切归零。

用文件系统（每个 session 一个 JSON 文件）能解决吗？能解决一部分。但三个问题处理不了：事务保证（写到一半崩溃了怎么办）、并发搜索（想搜历史消息得遍历所有文件）、以及 Gateway 模式下的并发读写——多个平台的消息同时到达，都往同一个数据库写。

Hermes Agent 选了 SQLite。不是因为「SQLite 比文件高级」，而是因为它同时解决了这三个问题。

最小实现

两张表：

conn = sqlite3.connect("state.db")
conn.execute("PRAGMA journal_mode=WAL")  # 开 WAL 模式

conn.executescript("""
CREATE TABLE IF NOT EXISTS sessions (
    id TEXT PRIMARY KEY,
    source TEXT NOT NULL,
    started_at REAL NOT NULL
);

CREATE TABLE IF NOT EXISTS messages (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id TEXT NOT NULL REFERENCES sessions(id),
    role TEXT NOT NULL,
    content TEXT,
    timestamp REAL NOT NULL
);
""")

然后三件事：创建 session → 每轮对话后增量写入 → 下次启动时读回历史。五行代码接进循环就行。

# 启动时
session_id = create_session(conn, source="cli")

# 每轮对话后
new_messages = result["messages"][len(old_messages):]
add_messages(conn, session_id, new_messages)

# 下次启动
history = get_session_messages(conn, session_id)

Hermes 的独特设计

WAL 模式。 注意这一行：<code>PRAGMA journal_mode=WAL</code>。SQLite 默认模式下一个写操作会阻塞所有读操作。WAL 模式下，写操作写到日志文件，读操作继续读主数据库——写不阻塞读。但写和写之间还是要排队。

随机退避解决写锁冲突。 Hermes 把 SQLite 超时设短（1 秒），然后在应用层做随机退避重试。确定性的等待时间在高并发下会导致队列效应（所有人等同样长的时间），随机间隔自然打散了竞争。

system_prompt 缓存到 session 表。 这个细节要配合 s04 的 prompt 缓存理解。Gateway 每条消息创建新的 AIAgent 实例，如果每次都重新组装 system prompt，可能因为 MEMORY.md 被改导致 prompt 变化，Anthropic 的 prompt cache 就失效了。所以 Hermes 第一次组装后把 system prompt 存到 session 表，后续实例直接读缓存。

---

机制四：Prompt Builder —— system prompt 不是硬编码字符串

它是什么

Agent 每次调 API 时，system prompt 是从六七个来源按优先级组装出来的，不是一段写死的字符串。

为什么需要

Agent 需要知道自己是谁（人设）、用户偏好什么（记忆）、项目有什么规则（项目配置）、有哪些技能能用、现在几点——这些信息分布在不同的文件里。如果硬编码在一起，改一处就要改代码，换个项目更麻烦。

最小实现

六层组装，按顺序拼：

def build_system_prompt(soul, memory, skills, project_context):
    parts = [soul]

    if memory:
        parts.append(f"# Memory\n{memory}")
    if skills:
        parts.append(f"# Skills\n{skills}")
    if project_context:
        parts.append(f"# Project Context\n{project_context}")

    parts.append(f"Current time: {datetime.now().strftime('%Y-%m-%d %H:%M')}")

    return "\n\n".join(parts)

六层分别是：

1. 人设（SOUL.md）—— Agent 是谁，什么风格
2. 行为指导——怎么用工具，模型特定的行为规范
3. 记忆（MEMORY.md + USER.md）——用户偏好和长期记忆
4. 技能清单——已安装技能的索引
5. 项目配置（HERMES.md / AGENTS.md / CLAUDE.md / .cursorrules）——项目规则，按优先级只用第一个找到的
6. 时间戳 + 模型信息

组装一次，缓存复用。同一个 session 的所有 API 调用都用同一份，不重新组装。

Hermes 的独特设计

续接 session 时从 SQLite 读 prompt，不重新组装。 前面 s03 提过，这是配合 Anthropic prompt caching 的底线。prompt 变了，缓存就失效，多花钱。

ephemeral_system_prompt 不进缓存。 有些指令只在本次调用临时加入（比如 Gateway 的临时配置），拼在缓存 prompt 后面，不存库，不进缓存。

记忆注入分两条路。 内置记忆（MEMORY.md / USER.md）进 system prompt。外部记忆提供者（plugin）的内容不进 system prompt，而是注入到 user message 里。因为外部记忆每轮可能不同，放进 system prompt 会破坏缓存。

---

机制五：Context Compression —— 压缩不是把历史缩短

它是什么

当对话消息堆到快撞上下文上限时，自动把中间「已经用过了」的轮次压缩成一段结构化摘要，腾出空间让模型继续干活。

但重点不是「压缩」，而是「压缩后模型还能接着干活」。

为什么需要

Agent 能干活了，上下文也更快膨胀了。读一个大文件，塞进很多文本。跑一条长命令，输出几百行。多轮工具调用后，旧结果越堆越多。

三个后果：模型注意力被旧内容淹没；API 请求越来越重、越来越贵；最终撞上上下文上限，任务中断。

最小实现

三层递进，从便宜到贵：

第 1 层：旧工具输出先裁剪。 不需要 LLM，纯字符串替换。把很久以前的 tool 消息的 content 替换成 <code>[Old tool output cleared]</code>。工具输出通常很长，这一步能先减掉大量 token，不花钱。

第 2 层：保护头尾，只压中间。 头部（任务定义，前几条）不动，尾部（最近约 20K tokens 的工作）不动，只压中间那些「已经用过了」的轮次。

第 3 层：用辅助 LLM 生成结构化摘要。 调一个便宜的辅助模型，把中间部分压缩成一段摘要，替代原来的几十条消息。

摘要不是自由文本，而是有格式的：

## Goal
...
## Progress
...
## Key Decisions
...
## Files Modified
...
## Next Steps
...

这五件事必须保住，否则压缩虽然腾出了空间，却打断了工作连续性。

Hermes 的独特设计

TaskState：把目标+进度搬出消息流。 这才是真正要解决的问题。摘要是有损压缩——它把「已读 s01, s02, s03」摘成「已读若干文件」，下一轮压缩时再浓缩一次，信息熵指数级衰减。模型只能重新规划，甚至重复读已经读过的文件。

TaskState 不进 messages 列表，压缩算法看不见它，永远不会被裁剪或摘要。每轮 API 调用前，TaskState 拼到 system prompt 末尾。模型每一轮都看到完整的目标和 TODO 进度。

边界对齐。 压缩时，<code>assistant.tool_calls</code> 和它的 <code>tool</code> 响应必须配对。如果只留前者、压了后者，下一次 API 请求直接被拒。Hermes 的 <code>find_boundaries</code> 会自动把切点「吸附」到完整配对边界。

Preflight 压缩。 进入主循环之前就检查 token 数。如果已经超了（比如用户从小窗口模型切到大窗口模型），在第一次 API 调用前就压缩。不等 API 报错再处理。

压缩失败早退。 如果压缩后 token 没降到原值的 90% 以下，说明找不到可压的中段，直接抛异常告诉用户「新开 session」，不空转。

---

机制六：Error Recovery —— 错误不是例外，是主循环的正常分支

它是什么

API 调用各种出错——限流、超时、模型不存在、输出被截断——这些不是「意外」，是常态。错误恢复就是把这些情况分类，然后按分类走对应的恢复路径。

为什么需要

Agent 不再是一个 demo，它在真的做事。网络超时、限流、服务抖动、API key 过期、模型被下线——这些随时发生。如果没有恢复机制，第一个错误就直接崩溃。

最小实现

先分类，再恢复。四条路径：

1. 输出被截断（finish_reason: length）
   → 注入续写提示，再试（最多 3 次）

2. 上下文太长（400 / context overflow）
   → 触发压缩（s05），再试

3. 临时故障（429 限流 / 503 过载 / 超时）
   → 退避等待，再试（最多 3 次）

4. 不可恢复（401 认证失败 / 404 模型不存在 / 额度用完）
   → 尝试故障转移到备用模型，或放弃

分类器长这样：

def classify_error(status_code, error_message):
    if status_code == 429:
        return {"reason": "rate_limit", "retryable": True}
    if status_code == 400 and "context" in error_message.lower():
        return {"reason": "context_overflow", "retryable": True, "should_compress": True}
    if status_code in (500, 502, 503):
        return {"reason": "server_error", "retryable": True}
    if status_code in (401, 403):
        return {"reason": "auth", "retryable": False, "should_fallback": True}
    if status_code == 404:
        return {"reason": "model_not_found", "retryable": False, "should_fallback": True}
    return {"reason": "unknown", "retryable": False}

退避重试不能是固定时间——多个 Gateway 会话同时重试会撞在一起。Hermes 用指数递增 + 随机抖动：

def jittered_backoff(attempt, base_delay=5.0, max_delay=120.0):
    delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
    jitter = random.uniform(0, delay * 0.5)
    return delay + jitter

续写提示不能模糊——只写「continue」不够，模型经常会重新总结或重新开头。Hermes 的续写提示是：

CONTINUE_MESSAGE = (
    "Your response was cut off. Continue EXACTLY from where you stopped. "
    "Do not restart, do not repeat, do not summarize what came before."
)

Hermes 的独特设计

多提供商意味着更多错误格式。 单提供商 Agent 只需要处理一个 API 的错误格式。Hermes Agent 支持 OpenRouter、Anthropic、本地端点等多种提供商，每种错误格式和状态码含义不完全一样。分类器要从不同提供商的错误消息中提取出统一的故障原因。

连接健康检查。 进入 <code>run_conversation()</code> 之前，先检查 API 客户端连接是否健康。如果检测到上一轮留下的死连接（比如上次超时后 TCP 连接还挂着），主动清理掉。不等新调用挂在僵尸连接上才发现。

主模型恢复。 故障转移到备用模型后，下一轮开始时自动尝试切回主模型。故障转移是临时的，主模型恢复了就自动回去，不需要用户手动切。

---

六个机制拼出的完整底座

到这里，六个机制全部到位。这张图可以看出它们各管什么事：

┌─────────────────────────────────────────────────┐
│  s06 错误恢复   ← 每一层 API 调用都可能出错      │
├─────────────────────────────────────────────────┤
│  s05 上下文压缩 ← 消息太长时保护连续性            │
├─────────────────────────────────────────────────┤
│  s04 提示词组装 ← 给模型完整的「工作说明书」      │
├─────────────────────────────────────────────────┤
│  s03 会话持久化 ← 消息存到 SQLite，跨重启存活     │
├─────────────────────────────────────────────────┤
│  s02 工具系统   ← 模型的手，按名查表分发执行      │
├─────────────────────────────────────────────────┤
│  s01 Agent 循环 ← 底座：提问→执行→喂结果→继续    │
└─────────────────────────────────────────────────┘

从最小循环到完整底座的变化：

s01: while True → 调模型 → 有工具就执行
s02: + 工具自注册、按名分发、async 桥接
s03: + 消息持久化到 SQLite，WAL 并发安全
s04: + 六层 system prompt 组装，缓存复用
s05: + 上下文压缩，TaskState 保任务连续性
s06: + 错误分类、退避重试、故障转移

六个机制各管各的事，互不越界，但拼在一起就是一个完整的单 Agent 底座。

这个底座说到底就一句话：模型思考，代码给模型提供一个可持久、可管理技能、能扛故障的工作环境。

---

附录：核心概念速查

20 个概念，每个一句话：

概念	一句话解释
iteration	一次 API 调用
iteration budget	本次对话允许调用 API 的总次数上限，可被消耗和分享
finish_reason	模型为什么停下来：stop / tool_calls / length
messages vs api_messages	内部完整账本 vs 每次调 API 前清洗出的副本
自注册模式	工具文件在末尾调用 register()，编排层只管导入
导入链	registry ← tools ← model_tools ← run_agent，不能反向
WAL 模式	SQLite 的日志模式，写不阻塞读
parent_session_id	压缩后新 session 指向旧 session，形成归档链
SOUL.md	人设文件，定义 Agent 的身份和行为风格
HERMES.md	项目级配置文件，告诉 Agent 项目规则
prompt 缓存	system prompt 组装一次缓存复用
三层压缩	旧输出裁剪（免费）→ 找边界 → LLM 摘要（花钱）
TaskState	不进消息流的任务态，目标+进度永不丢失
错误分类	HTTP 状态码 + 错误消息 → 翻译成恢复动作
退避重试	指数递增 + 随机抖动
故障转移	主模型不可用自动切备用，恢复后自动切回

---

最佳实践速查

场景	正确做法	常见错误
加新工具	只加一个文件，不改循环	改 if/elif 分发链
消息持久化	增量写入，只存新增的	全量写入
system prompt	组装一次缓存，每次拼在 api_messages 最前面	放进 messages 列表
上下文太长	三层递进：裁剪 → 找边界 → LLM 摘要	一上来就调 LLM
任务进度	用 TaskState 维护在消息流之外	让摘要承载目标和进度
错误处理	先分类，再恢复，每条路径有预算	所有错误走同一条路

---

第一阶段到此结束。后续还有第二阶段（S07-S11）：记忆系统、技能系统、权限系统、子 Agent 委派、配置系统。看这篇反馈再决定写不写。

---

从零手搓生产级 AI Agent 系列

• 第 1 期：六个机制拼出单 Agent 底座（本文）
• 第 2 期：记忆、技能、权限、委派与配置（待写）
• 第 3 期：跨平台 Gateway 与适配器（待写）
• 第 4 期：MCP、浏览器、语音视觉（待写）
• 第 5 期：技能创作、Hook、强化学习（待写）