第 2 章 · Agent 执行循环

2.1 Agent 的定义

先给一个严格的定义，避免术语漂移：

Agent = LLM + 循环 + 工具 + 终止条件。

在一个循环里，模型被反复调用，每次它可以选择调用工具（由宿主代码执行并把结果回灌）或给出最终回答；循环在模型不再要工具、或触及预算时结束。

这个定义把行业里两个常被混淆的概念分开了：

• Workflow（工作流）：控制流由代码写死（先调 A 再调 B），模型只负责其中的单步生成。LLM 是流水线上的工人；
• Agent：控制流由模型决定（它自己选择下一步调什么工具）。LLM 是包工头。

这个区分来自 Anthropic《Building effective agents》，是全部设计决策的第一分叉口：能 workflow 解决的不要上 agent——agent 的可控性、可预测性、成本都显著更差，但它是唯一能处理"步骤无法预先确定"的问题的形态。

2.2 最小可运行 Agent

下面是一个完整的 Agent，约 100 行，无任何框架。逐段精读比任何教程都有效。

"""minimal_agent.py — 一个完整的最小 Agent"""
import json
from dataclasses import dataclass, field
from openai import OpenAI

client = OpenAI()

# ---------- 1. 工具：真实执行层 ----------
def read_note(path: str) -> str:
    with open(path, encoding="utf-8") as f:
        return f.read()

def write_note(path: str, content: str) -> str:
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return f"已写入 {path}（{len(content)} 字符）"

def list_notes(dir_path: str = ".") -> str:
    import os
    return "\n".join(os.listdir(dir_path))

TOOLS_IMPL = {"read_note": read_note, "write_note": write_note, "list_notes": list_notes}

TOOLS_SCHEMA = [
    {"type": "function", "function": {
        "name": "list_notes",
        "description": "列出目录下的所有笔记文件名。",
        "parameters": {"type": "object", "properties": {
            "dir_path": {"type": "string", "description": "目录，默认当前目录"}}}}},
    {"type": "function", "function": {
        "name": "read_note",
        "description": "读取一个笔记文件的完整内容。",
        "parameters": {"type": "object", "properties": {
            "path": {"type": "string", "description": "文件路径"}}, "required": ["path"]}}},
    {"type": "function", "function": {
        "name": "write_note",
        "description": "把内容写入笔记文件（整体覆盖）。",
        "parameters": {"type": "object", "properties": {
            "path": {"type": "string"}, "content": {"type": "string"}},
            "required": ["path", "content"]}}},
]

# ---------- 2. Agent 状态 ----------
@dataclass
class AgentResult:
    output: str
    tool_calls: int = 0
    rounds: int = 0

def run_agent(task: str, max_rounds: int = 15, max_tool_calls: int = 30) -> AgentResult:
    messages = [
        {"role": "system", "content":
            "你是笔记管理助手。完成任务后用一段话总结做了什么。"
            "工具失败时不要放弃，先理解错误再换方法。"},
        {"role": "user", "content": task},
    ]
    tool_calls_total = 0
    for round_no in range(1, max_rounds + 1):
        resp = client.chat.completions.create(
            model="gpt-4o", messages=messages, tools=TOOLS_SCHEMA)
        msg = resp.choices[0].message

        if not msg.tool_calls:                       # 终止条件 A：模型给出最终答案
            return AgentResult(msg.content, tool_calls_total, round_no)

        messages.append(msg)                         # assistant 消息必须先回灌
        for tc in msg.tool_calls:                    # 工具调用逐个执行（生产上可并行）
            if tool_calls_total >= max_tool_calls:   # 终止条件 B：工具调用预算
                result = json.dumps({"error": "预算耗尽，请基于已有信息给出最终回答"},
                                    ensure_ascii=False)
            else:
                tool_calls_total += 1
                try:
                    impl = TOOLS_IMPL[tc.function.name]
                    result = impl(**json.loads(tc.function.arguments))
                except Exception as e:               # 工具失败也是"结果"，回灌让模型自救
                    result = json.dumps({"error": f"{type(e).__name__}: {e}"},
                                        ensure_ascii=False)
            messages.append({"role": "tool", "tool_call_id": tc.id,
                             "content": str(result)})

    return AgentResult("[达到轮数上限被强制终止]", tool_calls_total, max_rounds)  # 终止条件 C

if __name__ == "__main__":
    r = run_agent("把笔记 a.txt 和 b.txt 的内容合并成 summary.txt")
    print(r.output, f"\n[工具调用 {r.tool_calls} 次，{r.rounds} 轮]")

这段代码包含了 Agent 的全部骨架，后面所有章节都是在给它的某一部分"加工程"：

代码位置	对应主题	章
messages 无限增长	上下文压缩	03
TOOLS_SCHEMA 描述工程	工具与 MCP	05
except Exception 回灌错误	工具错误设计	05
max_rounds / max_tool_calls	预算与止损	本章 2.4
写文件无隔离	沙箱	06

请把这段代码跑起来，然后故意做三个实验：把 messages.append(msg) 删掉（模型立刻困惑）；把工具异常改成直接 raise（一次失败整个任务死掉）；把 max_rounds 改成 100 并给一个模糊任务（观察死循环烧钱）。这三个实验比十篇文章更能建立直觉。

2.3 设计模式谱系

在最小循环之上，行业沉淀了几种模式。它们不是框架特性，而是 prompt 结构 + 循环结构的组合。

ReAct（Reason + Act）

模型的思考过程显式进入上下文。实现上只需改 system prompt：

按以下格式交替进行：
Thought: 我需要先弄清楚 X
Act: 工具名(参数)
（系统返回 Observation）
Thought: 根据 Observation 我知道了 Y，接下来……
……直到你能给出 Final Answer

优点：思考外显，可调试性好（错误推理能被看见），工具选择质量高。缺点：思考本身耗 token，且"思考文本"容易被用户看到（产品上可能不想要）。适合探索型任务——路径不确定、需要边看边决策。

Plan-and-Execute

先让模型产出结构化计划，再逐步执行：

第一阶段（规划器）：
  输出 JSON：{"steps": [{"id": 1, "action": "...", "done_when": "..."}, ...]}

第二阶段（执行器）：按步骤执行，每步是一个小 Agent 循环

第三阶段（可选，重规划）：某步失败时，把"已完成 + 失败原因"喂回规划器重新出计划

优点：长任务不迷失（计划是持久的目标锚点）；计划可以被人审（天然 HITL 挂点）；步骤间可以并行。缺点：计划可能与现实脱节（执行中发现步骤 3 根本不可行）——所以必须配重规划。适合结构相对稳定的长任务。

Reflection

执行完成后追加一轮自我批判：

任务输出: {draft}
请审查：1) 是否完成任务的全部要求 2) 有哪些具体错误 3) 给出修正后的版本

优点：显著提升单任务质量。缺点：延迟和成本接近翻倍；模型自我审查有盲区（它看不见自己的系统性偏差）。适合质量敏感、量少的输出（一封重要邮件、一段关键代码），不适合批量流水线。

Router（路由）

一个轻量模型先分类用户意图，再交给对应的专职 Agent/工作流处理。这是控制成本的第一杠杆：80% 的简单请求被便宜模型消化，复杂请求才进入昂贵的多步循环。

选型速查

任务形态	推荐
步骤固定	Workflow，别用 Agent
步骤不定但单 agent 能扛	ReAct 循环
长任务、易迷失	Plan-and-Execute + 重规划
输出质量 > 延迟成本	+ Reflection
请求异质且量大	Router 分流
子问题独立且可并行	Multi-Agent（第 7 章详细讨论，先记住警告：token 约 15×）

2.4 终止条件与预算：Agent 的"刹车系统"

没有刹车的 Agent 是生产事故。四层防护，全部要有：

1. 自然终止：模型不再发起工具调用（正常路径）；
2. 轮数预算：max_rounds 限制循环次数。经验值：简单任务 8-15 轮，复杂任务 30-50 轮。达到上限后不要直接抛异常，而是注入一条"预算耗尽，请基于已有信息给出最终回答"的用户消息，让模型做体面收尾——半路掐死会丢失它已经收集到的全部信息；
3. 工具调用预算：独立的工具次数上限，防止"每轮只调一次工具"绕过轮数预算；
4. token/费用预算：累计 usage 超阈值即触发收尾。这是唯一能直接控钱的刹车，第 9 章会扩展成配额体系。

死循环的典型形态与对策：

• 重复调用同一工具同一参数：在循环里记录 (tool, args_hash) 历史，检测到重复时注入提示"你已经调用过相同的工具和参数，结果不会变化，请换思路或给出答案"；
• 工具报错后固执重试：同上，把连续失败计数注入上下文；
• 永不终止的礼貌：模型每轮都输出一点内容但永远说"接下来我再……"——轮数预算兜底 + system prompt 明确"完成任务就输出最终答案，不要预告未来动作"。

2.5 流式与并行工具调用

生产级循环还有两个工程细节。

流式 + 工具调用的混合处理（第 1 章 1.3 的延伸）：一轮流式响应里，内容块实时推送前端，工具调用块静默累积，响应结束后统一执行。渲染层需要区分"思考中"与"回答中"，否则用户会看到半截的工具中间产物。

并行工具调用：模型一次输出多个 tool_calls 时，相互独立的可以并发执行（如同时读三个文件），有依赖的串行。并行时用 tool_call_id 严格配对回灌。注意：有副作用的工具（写文件、发请求）默认不要并行——除非你能证明互不冲突，否则并行写是竞态温床。

2.6 循环宿主：工程架构的分水岭

最小 Agent 的循环跑在一个函数里，生产系统的循环宿主要有额外能力：中断（用户叫停）、恢复（进程重启后继续）、持久化（每一轮的状态落盘）、可观测（每轮的决策可回溯）。这决定了循环宿主的形态：

进程内协程（简单，进程死则任务死）
  → 状态机 + 事件溯源（每轮先落盘再执行，崩溃后可重放）
    → 外置执行器（引擎与执行分离，worker 从队列领取步骤，水平扩展）

实现作业：把 2.2 的最小 Agent 改造为可恢复的——每一轮结束后把 messages 与预算计数序列化到本地 JSON；进程被杀后重跑时从断点继续（幂等：已执行的工具调用不重复执行）。完成这个作业，你就理解了所有"长任务 Agent 平台"的核心难点。

深入材料

• Anthropic, Building effective agents（2024，workflow vs agent 谱系的原始出处）
• ReAct 原论文：arXiv 2210.03629（Yao et al., 2022）
• Reflexia（Reflection 的系统性实验）：arXiv 2303.11366
• Anthropic, Effective context engineering for AI agents（2025，与第 3 章衔接）