本节摘要:Agent 主循环是 harness 唯一不可省略的组件,也是最容易被低估的组件——它不是
while True: call_llm(),而是一台四拍机器:感知(组装上下文:模型这一步该看见什么)、决策(调用模型,拿到文本或工具调用请求)、行动(把工具调用经权限判定后派发执行)、观察(把执行结果回填上下文),然后回到感知,直到模型不再请求工具、给出最终答案。本节先建立四拍模型与状态机视角,再给出约 60 行的最小可运行循环mini_loop.py(本书第一块积木),逐行标注组件归属,最后点出流式输出与并行工具调用两个生产化细节的最小处理方式。
阅读完本节,你应当能够:
mini_loop.py,指出任意一行属于哪一拍。单次 LLM 调用是"一问一答":给上下文,得回复,结束。但真实任务是多步的——修一个 bug 需要先读文件、再搜索、再改代码、再跑测试。两次调用之间的鸿沟在于:LLM API 是无状态的,它不会"记得"上一步做了什么,也不能自己动手做任何事。
填鸿沟的就是主循环。它维护唯一的状态(messages 列表),驱动唯一的动力(模型),并且独占一条分工红线:
模型只负责"决策"(说"我想调用某某工具"),harness 负责"执行"(真的去调、拿结果、决定喂回什么)。模型从头到尾没有碰过你的文件系统——它只是在发电流,harness 把电流变成动作。
| 拍 | 名字 | 输入 | 输出 | 关键点 |
|---|---|---|---|---|
| ① | 感知(Perceive) | 任务、历史、环境状态 | 本步的 messages | 装什么、裁什么(第 3.3 节) |
| ② | 决策(Decide) | messages + 工具 schema | 文本 / tool_calls 列表 | 模型可能一次要调多个工具 |
| ③ | 行动(Act) | tool_calls | 每个调用的结果 | 权限门在此(第 5 章)、沙箱在此(第 6 章) |
| ④ | 观察(Observe) | 执行结果 | 回填的 tool 消息 | 错误也是观察(第 4.3 节) |
四拍即一个 step(步/轮)。循环 = 四拍 + 一个终止判断("模型不再要工具")。这个结构对所有产品成立——第 2 章五个样本的差异只在每拍内部的策略,不在四拍本身。
mini_loop.py:60 行的最小循环# mini_loop.py —— 最小 Agent 主循环(写法示意,以官方文档为准) import json SYSTEM_PROMPT = "你是一个能读写文件的编程助手。完成任务后直接给出最终答案。" TOOLS = [ # ② 工具系统(第 4 章将用注册器自动生成) {"type": "function", "function": { "name": "read_file", "description": "读取指定路径文件的内容", "parameters": {"type": "object", "properties": { "path": {"type": "string", "description": "相对路径"}}, "required": ["path"]}, }}, {"type": "function", "function": { "name": "list_dir", "description": "列出目录下的条目", "parameters": {"type": "object", "properties": { "path": {"type": "string"}}, "required": ["path"]}, }}, ] TOOL_FUNCS = {"read_file": lambda p: open(p, encoding="utf-8").read(), "list_dir": lambda p: "\n".join(__import__("os").listdir(p))} def call_model(messages): # ②' 模型插槽:换模型只改这一个函数 """调用任意兼容 chat 接口的模型。示意:伪代码。""" raise NotImplementedError("接你选择的 SDK,返回带 tool_calls 的消息") def dispatch(name: str, args: dict) -> str: # ③ 行动 func = TOOL_FUNCS.get(name) if func is None: return json.dumps({"error": f"unknown tool: {name}"}, ensure_ascii=False) try: return str(func(**args)) # 权限门应插在这次调用之前(第 5 章) except Exception as exc: return json.dumps({"error": str(exc)}, ensure_ascii=False) def run(task: str, max_steps: int = 30) -> str: messages = [{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": task}] for step in range(max_steps): # 终止条件之一(3.2 节) msg = call_model(messages) # ② 决策 messages.append({"role": "assistant", "content": msg.content, "tool_calls": getattr(msg, "tool_calls", None)}) calls = getattr(msg, "tool_calls", None) or [] if not calls: # 正常终止:模型给最终答案 return msg.content for call in calls: # ③ 行动(逐个派发) args = json.loads(call.function.arguments) result = dispatch(call.function.name, args) messages.append({"role": "tool", # ④ 观察:结果回填 "tool_call_id": call.id, "content": result}) return "已达到最大步数,任务未完成。" # 超限终止
逐行归属:messages 的维护是①感知的载体;call_model 是②决策(被刻意抽成唯一插槽——第 1.1 节"模型易朽"原则);dispatch 是③行动(权限门应插在 func(**args) 之前,第 5 章补上);role: "tool" 回填是④观察;max_steps 与 if not calls 是 3.2 节的主题。
💡 跑通建议:找一段你 SDK 的 function calling quickstart,把
call_model填上,然后给它一个任务如"列出当前目录并读出 README 前几行"。你会看到循环转 2~3 步后自己停下来——第一次亲眼看到四拍转起来,是理解 harness 的顿悟时刻。
流式输出(streaming):②决策往往是几十秒的长思考,终端产品都边生成边显示。最小处理:call_model 改为流式聚合,对循环结构零影响——流式只改变"决策这一拍怎么把字吐给用户",不改变四拍本身。
并行工具调用(parallel tool calls):模型可能一次返回多个 tool_calls(如同时读三个文件)。最小处理如上:for call in calls 逐个执行后统一回填。生产化的升级是用 asyncio.gather 并发执行互不依赖的调用,再按 call.id 对齐回填——注意顺序契约:无论怎么并发,回填顺序与 id 对应关系不能乱,否则模型的"观察"就对不上它的"决策"。
⚠️ 一个常见事故:并行执行时把结果按完成顺序回填,而模型按请求顺序理解——表现为模型把 A 文件的内容当成 B 文件的。按
tool_call_id对齐是硬要求,不是最佳实践。
mini_loop.py:约 60 行、一个模型插槽、一个派发器;权限门与观测的挂点已留好,第 4~5 章逐步填上。tool_call_id 对齐回填。循环转起来了,但"转起来"只是开始。真实任务里模型会空转、会连环撞墙、会在第 40 步才发现方向错了——下一节处理循环的暗面:什么时候停、坏了怎么办。