100 行手写 Agent 循环

专栏：Agent 工程 · 第 4 / 18 篇
AgentAgent 循环源码对照

上一篇你完成了 function calling 的一次往返。今天把它装进 while 循环——这就是 agent 的全部本体，100 行以内。然后打开两个生产级系统的源码，看同一件事的工业版。

:::info 学习目标 完成本篇后你能够：从零写出并调试一个带工具循环的 mini-agent；解释工业版循环在并发、转向、状态机上的五个增强点各自防什么问题。 前置：第 3 篇的裸 API 往返已跑通。预计时长：60~90 分钟（含练习）。 :::

:::note 本章术语速查（新手建议先读）

  • Agent 循环：agent 的核心骨架——问模型 → 模型要调工具 → 执行 → 结果给它 → 再问……直到完成。
  • ReAct：一种经典模式（推理+行动交替），现代 agent 循环的雏形。
  • tool_calls：模型回复里的”工具调用清单”——它想用什么工具、什么参数（注意：只是想法，执行是你的事）。
  • 终止条件：让循环停下来的规则，比如”模型不再调工具""超过 25 步""用户取消”。
  • 错误回填：工具执行失败时，把错误信息（而非崩溃）发回给模型，让它换个办法。
  • Steer（转向）：任务运行到一半时用户补充指示，agent 能接收并调整。 :::

100 行实现

完整可运行。存为 mini_agent.py，填入你的 KEY：

import json, subprocess, sys, urllib.request

API = "https://api.deepseek.com/chat/completions"
KEY = "<你的 KEY>"

TOOLS = [
    {"type": "function", "function": {
        "name": "run_bash",
        "description": "在工作目录执行一条 bash 命令，返回 stdout+stderr。用于查看文件、搜索代码、运行测试。",
        "parameters": {"type": "object", "properties": {
            "command": {"type": "string", "description": "要执行的命令"},
        }, "required": ["command"]},
    }},
    {"type": "function", "function": {
        "name": "read_file",
        "description": "读取文本文件内容。用于查看具体文件。按内容搜索请用 run_bash + grep。",
        "parameters": {"type": "object", "properties": {
            "path": {"type": "string", "description": "绝对路径"},
        }, "required": ["path"]},
    }},
]

SYSTEM = "你是一个谨慎的编码助手，工作目录为当前目录。先探索再修改；任务完成后直接给出总结，不要调用工具。"

def chat(messages):
    req = urllib.request.Request(API, data=json.dumps({
        "model": "deepseek-chat", "messages": messages, "tools": TOOLS,
        "temperature": 0.2,
    }).encode(), headers={"Content-Type": "application/json", "Authorization": f"Bearer {KEY}"})
    return json.load(urllib.request.urlopen(req))["choices"][0]["message"]

def execute(name, args):
    if name == "run_bash":
        p = subprocess.run(args["command"], shell=True, capture_output=True,
                           text=True, timeout=60)
        out = (p.stdout or "") + (p.stderr or "")
        return (out or "(no output)")[:4000]
    if name == "read_file":
        with open(args["path"]) as f:
            return f.read()[:8000]
    return f"unknown tool: {name}"

def run(task, max_steps=25):
    messages = [{"role": "system", "content": SYSTEM},
                {"role": "user", "content": task}]
    for step in range(1, max_steps + 1):
        reply = chat(messages)
        messages.append(reply)
        calls = reply.get("tool_calls")
        if not calls:
            return reply["content"]            # 模型不再调工具 → 任务完成
        print(f"[step {step}] {reply['content'] or '(thinking…)'}")
        for call in calls:
            args = json.loads(call["function"]["arguments"])
            print(f"  → {call['function']['name']}({args})")
            try:
                result = execute(call["function"]["name"], args)
            except Exception as e:
                result = f"Error: {e}"         # 失败也要回填，让模型自我纠正
            messages.append({"role": "tool", "tool_call_id": call["id"],
                             "content": result[:4000]})
    return "达到步数上限，任务未确认完成"

if __name__ == "__main__":
    print(run(" ".join(sys.argv[1:]) or "统计当前目录 Python 代码总行数"))

检查点：跑起来应该看到什么

$ python mini_agent.py 统计当前目录 Python 代码总行数，并列出最大的三个 .py 文件
[step 1] 我先用 find 统计…
  → run_bash({'command': 'find . -name "*.py" | xargs wc -l'})
[step 2] 行数已统计，再排序找最大的三个…
  → run_bash({'command': 'find . -name "*.py" | xargs wc -l | sort -rn | head -4'})
[step 3] 当前目录共有 1234 行 Python 代码，最大的三个文件是 …

三件事必须符合预期：每步打印工具调用；失败的工具返回 Error 文本而循环继续；模型不再调工具时循环结束。如果第一步就没调工具，先检查工具描述（第 5 篇的主题提前预演）。

逐段拆解：每个设计决定防什么

代码位置设计决定防的问题
for step in range(...)步数上限模型陷入循环烧钱
messages.append(reply)助手消息（含 tool_calls）入历史消息配对断裂，下次请求报 400
except Exception → "Error: ..."失败回填不中断一次工具失败炸掉整个任务
result[:4000]结果截断单个工具结果吃光窗口
if not calls: return模型自主终止无法判断何时收尾

工业版一：Codex 的 run_turn

打开 codex-rs/core/src/session/turn.rs（近 3000 行），骨架与上面同构，增量按”解决什么问题”排列：

  • 循环外围包了三层：ThreadManager（线程生命周期：start/resume/fork/子 agent）→ RegularTask（任务壳，turn 结束后若队列还有输入继续下一轮）→ run_turn（step 循环）。100 行版只有一个函数，工业版把「一次任务」和「一次轮次」分开管理。
  • 流式 + 工具并发：try_run_sampling_request 用 FuturesOrdered 让流读取与工具执行并发——模型吐完 token 之前，前面的工具调用已经在跑。
  • 步进前先看预算：每次采样后检查 token 状态，超限触发 mid-turn 自动压缩（下一篇细讲）。
  • 每步都有”步骤上下文”：capture_step_context 一次性捕获本步可用的工具清单、权限视图——保证一次请求内视图一致。

steer（运行中转向） 是最值得学的增量：CodexThread::start_or_steer_turn 允许用户在模型运行时追加输入，循环每步领取 pending input 并进上下文。dsh 的 inbox 设计完全同构——followup()（开新轮）/ steer()（注入当前轮）/ inject()（只入队不唤醒）三个入口。

工业版二：dsh 的 agent-loop

packages/core/agent-loop/src/agent.ts 的主循环我们在专栏第 7 篇逐行读过，这里只提三个手写版必然缺失的机制：

  1. 相位状态机：Phase = idle | maintenance | running{turn, step}——turn/step 计数单调递增，turn/start 到 turn/end 在 finally 里必写日志，任何退出路径轮次边界都闭合。
  2. inbox 是持久事件：三种输入入口（followup/steer/inject）先落 agent/inbox/spliced 事件再改内存——崩溃后队列可重建。
  3. pre-step 把关：每步进模型前经过 agent/pre-step waterfall，插件可改写或拒绝输入（拒绝则以 blocked 关闭轮次，日志仍记录这次尝试）。

框架对照：LangGraph 的同一段逻辑

工程化框架把这层循环封装掉（专栏立场：先手写再用）：

from langchain.agents import create_agent
agent = create_agent(model="deepseek:deepseek-chat", tools=[run_bash, read_file])
agent.invoke({"messages": [{"role": "user", "content": "统计代码行数"}]})

create_agent 内部就是一个 StateGraph：模型节点 ⇄ 工具节点的条件边循环。手写过一遍后，框架对你不再有魔法。

从 100 行到生产版：差距清单

能力100 行版工业版（谁做的）
消息历史内存 list，进程死即失仅追加事件日志 + JSONL 持久化（dsh session / Codex rollout，第 6 篇）
工具执行串行 for 循环流式并发：FuturesOrdered（Codex）/ 有界滚动池（dsh）
用户中途插话不支持steer：start_or_steer_turn（Codex）/ steer()（dsh）
上下文超限报错或截断mid-turn 自动压缩（两系统，第 7 篇）
失败处理Error 文本回填结构化错误码 + 重试策略 + 渠道/工具熔断（第 9 篇）
权限无审批策略 + OS 沙箱（第 12 篇）
轨迹print事件日志：每步可回放（第 13 篇评测的原料）

这张表就是本专栏剩余所有篇章的地图——每走一篇，抹平一行差距。

常见踩坑

  1. tool 消息没紧跟 assistant 消息：回填顺序错会直接 400——assistant（含 tool_calls）之后必须紧跟对应数量的 tool 消息。
  2. 忘记 append 助手消息：只回填工具结果不追加 reply，tool_call_id 找不到归属。
  3. 结果不截断：一个 cat 大文件 让历史暴涨，后续每步都为它付费。
  4. temperature 忘了调低：评测时同任务每次走不同路径，无法回归。
  5. 把超时当失败中断：timeout 应转成 Error 文本回填，让模型换个命令，而不是抛出终止。

随堂练习（带验收标准）

  1. 跑通基线：用三个不同任务跑通（一个读类、一个执行类、一个混合类）。验收：每个任务步数 ≤ 10，最终回答包含可验证的数字/结论。
  2. 错误自纠实验：把 read_file 改成路径不存在时返回 Error: file not found, did you mean ...——验收：模型下一轮自动改用正确路径；再把错误信息改成空字符串，观察模型开始重复失败调用。体会错误信息质量的 ROI。
  3. 源码对照：打开 codex-rs/core/src/session/turn.rs 找到 run_turn 的 loop，列出工业版比你多的分支，每个标注「防什么」；在 dsh 的 agent.ts 里找到 steer()，回答：steer 的输入为什么进 next-step 而不是 next-turn？

下一篇解决手写版马上会撞上的两个问题：系统提示词怎么写模型才听话、工具描述怎么写模型才肯用。

← 返回文章列表