☰
手写最小Coding Agent:从ReAct循环到生产级骨架
2026/10/8 11:05:49 网站建设 项目流程

前阵子我用 Claude 和 Codex 这类工具处理了一个旧仓库的 issue,看着它自己读代码、改文件、跑测试,我就在想:这些东西的内部到底是怎么跑的?后来我去翻了 Pi 这类生产级 Coding Agent 的骨架,说实话,把它的模块拆开之后你会发现,核心原理一点都不神秘。这篇文章我想从零开始,手写一个最小可用的 Coding Agent,用它把 Pi 生产级 Agent 的骨架讲明白,顺便分享我实际跑任务时踩过的坑。

适合谁看?如果你写过 Python、用过 ChatGPT 或 Claude 的 API,但还没搞清楚 Agent 的 loop、工具调用、上下文管理这些概念,这篇文章就是给你准备的。我会把代码直接贴出来,你复制过去改一改就能跑。

1. 动手前先想清楚:Coding Agent 的本质是"带工具的循环"

很多人第一次接触 Coding Agent 会把想复杂了,以为里面有什么神奇的规划算法或者自动推理引擎。其实拆开来看,一个 Coding Agent 的核心就是四样东西:模型、工具、状态、控制循环。

模型负责"思考",也就是根据当前状态决定下一步做什么;工具负责"动手",比如读文件、执行命令、搜索代码;状态负责"记忆",记录模型看到了什么、做过什么;控制循环负责"调度",不断重复"思考 -> 调用工具 -> 把结果喂回去 -> 再思考"这个过程,直到任务完成。

这个套路在学术界有个名字,叫 ReAct(Reasoning + Acting),本质上是把推理和行动交替进行。Coding Agent 只是把 ReAct 循环的应用场景聚焦到了"代码库"上:模型需要能够浏览代码、理解代码、修改代码,并且要能看到自己改动的结果。

1.1 从 Pi 的骨架里看到的抽象

我看了 Pi 的实现思路(它是一款面向生产环境的开源 Coding Agent,主打让模型自主完成编码任务),它的骨架其实就是一句话:任务输入 -> 上下文感知 -> 工具调用 -> 结果反馈 -> 再决策。

但这五个环节每一个在实际落地时都有大量工程细节。比如"上下文感知"不是简单把整个仓库塞给模型,而是需要按需检索:用户问的是一个文件里的一个函数,系统就去定位这个文件、提取函数附近代码、补上相关依赖,而不是把 10 万行代码全部发给模型。这一步直接决定了 Coding Agent 能不能处理真实规模的仓库。

"工具调用"也不是简单让模型输出一段文字,而是需要一套严格的工具定义协议。模型输出的是"我想调用这个工具,参数是这些",系统需要解析这个结构、校验参数、执行工具、把结果转换成文本再回传给模型。这里任何一个环节出错,整个循环就会断掉。

1.2 最小可用版本的边界

明确了原理之后,要给自己划一条"最小"的边界。我的目标不是实现一个能和 Pi 或者 Codex 媲美的产品,而是实现一个能跑通完整闭环的最小系统。它需要满足三个条件:

  1. 能接入一个代码库(至少在本地目录上操作)。
  2. 能调用至少三个工具(列目录、读文件、执行命令)。
  3. 能自我纠错(工具调用失败或者模型 JSON 格式错误时,系统能把错误信息反馈给模型,让它重试)。

满足这三条,你就拥有了一个 Coding Agent 的最小核。后续加语义搜索、并发、沙箱、权限控制,都是在这些骨架上长肉。我见过很多人一上来就研究复杂的 agent 框架,结果连最基本的循环都没跑通,反而被框架的抽象层绕晕了。从最小的循环开始,是理解这类系统最稳妥的路径。

2. 最小骨架的模块划分:照着 Pi 的思路画一张蓝图

动手写代码之前,先把模块边界画清楚。我参考 Pi 的分层方式,把最小 Coding Agent 拆成六个模块:

模块职责最小实现要求
Task Parser把用户自然语言任务解析成内部指令可以直接透传,不做复杂解析
Context Collector为模型收集仓库上下文(文件列表、文件内容、检索结果)简单实现为"列目录 + 读文件"
Agent Core控制循环,负责调度模型和工具一个 while 循环 + 终止条件
Tool Registry工具注册与调用中心装饰器注册 + 按名字调用
Model Client封装大模型 API,负责对话历史管理OpenAI 兼容接口封装
State Store保存会话状态、历史消息、中间结果内存 list 即可

模块之间互相依赖的方向要控制好:Agent Core 依赖 Tool Registry 和 Model Client,但 Tool Registry 不依赖 Agent Core;Context Collector 可以被 Tool Registry 复用,也可以被 Agent Core 直接调用。这样设计的好处是:当你把模型从 A 换到 B 时,只需要修改 Model Client 一个模块;当你加一个新工具时,只需要在 Tool Registry 里注册一个函数。

2.1 为什么接口设计比功能实现更重要

我第一次写 Agent 的时候,把所有逻辑堆在两个文件里:一个文件里又调 API 又解析 JSON 又执行命令。改起来特别痛苦,比如想把"模型调用"从 GPT 换成别的模型,就得在好几个函数里改代码。

参考 Pi 的骨架之后我意识到,接口设计是 Coding Agent 最值得花心思的地方。核心接口其实只有三个:

  • 模型接口:generate(messages) -> str。输入是一组消息(system、user、assistant、tool),输出是模型生成的文本。不管底层是 GPT、Claude 还是本地模型,都被封装成这一个函数。
  • 工具接口:register(name, description, parameters, fn)和call(name, **kwargs) -> str。所有工具都以"名字 + 描述 + 参数 JSON Schema + 执行函数"的形式注册。
  • 循环接口:run(task, max_steps) -> str。接收任务,返回最终结果。循环内部是模型和工具交替调用。

用这几个接口把所有模块串起来之后,整个系统的复杂度一下子就降下来了。你不需要去理解"Agent 框架里怎么管理计划",你只需要保证这三个接口之间的数据格式是稳定的。

2.2 熟悉一下 GitHub 上 Pi 的代码结构演进路径

如果你去看 Pi 这个项目的目录,早期版本的代码结构其实相当朴素,就是一个agent.py加一个tools/目录。它的演进路径给我很大的启发:先跑通最小闭环,再逐步加功能。而且它的每个功能模块,比如终端工具、文件编辑工具、记忆模块,都是可以独立开关的插件,而不是耦合在核心循环里的硬编码逻辑。

这种演进思路值得我们借鉴。如果一开始就想着实现"多 Agent 协作""自动规划"这些高级功能,大概率会陷入过度设计。先把一个循环跑通,再考虑扩展,这才是务实路线。

3. 手写代码:一个能跑的最小 Coding Agent

下面进入正题,写代码。我的实现用 Python,核心逻辑不到 200 行。先说明一点:这里我不会依赖任何 Agent 框架,只用标准库和requests,这样你能看到每一行代码的作用。

3.1 基础数据结构

# datatypes.py from dataclasses import dataclass, field from typing import Optional @dataclass class Message: role: str # "system" | "user" | "assistant" | "tool" content: str

这个Message类是整个对话历史的基本单位。系统提示词是一个 Message,用户任务是一个 Message,模型生成的回复是一个 Message,工具执行的结果也会被包装成一个 Message 追加回去。

@dataclass class ToolResult: output: str error: Optional[str] = None

工具执行结果统一转成字符串,方便塞回给模型。不要直接返回结构化数据,因为模型的输入是文本。这是 Coding Agent 实现里一个很容易忽略的点:工具返回的字典、列表、异常对象,最后都要变成人类可读的字符串。

3.2 工具注册表

# tool_registry.py import json from typing import Callable, Any class ToolRegistry: def __init__(self): self._tools = {} def register(self, name: str, description: str, parameters: dict, fn: Callable): self._tools[name] = { "description": description, "parameters": parameters, "fn": fn, } def get_schemas(self) -> list: schemas = [] for name, meta in self._tools.items(): schemas.append({ "type": "function", "function": { "name": name, "description": meta["description"], "parameters": meta["parameters"], }, }) return schemas def call(self, name: str, **kwargs) -> ToolResult: if name not in self._tools: return ToolResult(output="", error=f"Unknown tool: {name}") try: result = self._tools[name]["fn"](**kwargs) return ToolResult(output=str(result)) except Exception as e: return ToolResult(output="", error=str(e))

get_schemas()返回的是标准的工具定义,可以直接塞给支持 function calling 的模型 API。call()里面用try-except把异常捕获并转成error字段,这样模型就能看到错误信息并自我纠错。

接下来定义三个最基础的工具:列目录、读文件、执行命令。

# builtin_tools.py import os import subprocess def ls(path: str = ".") -> str: """List directory contents.""" return "\n".join(sorted(os.listdir(path))) def read_file(path: str, max_chars: int = 8000) -> str: """Read a file and return its content, truncated.""" with open(path, "r", encoding="utf-8") as f: content = f.read() if len(content) > max_chars: return content[:max_chars] + f"\n...[truncated {len(content) - max_chars} chars]" return content def run_command(command: str) -> str: """Run a shell command and return stdout+stderr.""" try: result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30) output = result.stdout if result.stderr: output += "\n[stderr]\n" + result.stderr return output[:6000] except subprocess.TimeoutExpired: return "ERROR: command timed out after 30s"

这些工具都非常朴素,但已经足够演示完整的循环。read_file的截断逻辑很重要,后面我会在踩坑部分详细讲为什么必须截断。run_command用timeout防止模型调一条无限循环的命令把整个 Agent 卡死。

3.3 模型客户端封装

# model_client.py import requests import json class ModelClient: def __init__(self, base_url: str, api_key: str, model: str): self.base_url = base_url self.api_key = api_key self.model = model def generate(self, messages: list, tools: list) -> str: url = self.base_url.rstrip("/") + "/chat/completions" payload = { "model": self.model, "messages": [{"role": m.role, "content": m.content} for m in messages], "tools": tools if tools else None, "temperature": 0, } headers = {"Authorization": f"Bearer {self.api_key}"} resp = requests.post(url, json=payload, headers=headers, timeout=120) resp.raise_for_status() data = resp.json() return data["choices"][0]["message"]["content"]

我故意没有用某个具体厂商的 SDK,而是直接调 OpenAI 兼容的 HTTP 接口,这样你可以把base_url换成任何兼容的供应商。temperature=0是为了让模型输出更稳定,Coding Agent 场景下我基本不用随机采样。

3.4 Agent Core:控制循环

这是整个 Coding Agent 的心脏。

# agent_core.py import json import re from datatypes import Message from tool_registry import ToolRegistry from model_client import ModelClient SYSTEM_PROMPT = """你是一个运行在用户代码仓库中的编码代理。 你可以调用以下工具来了解代码库:ls, read_file, run_command。 规则: 1. 每次回复必须是一个 JSON 对象,格式如下: {"thought": "你在这个步骤的思考", "tool": "要调用的工具名或null", "args": {"参数名": "参数值"}} 2. 如果任务还没完成,继续调用工具。 3. 如果任务已经完成,或者你确定无法继续,tool 必须为 null,并在 thought 中给出最终答案。 4. 不要编造工具执行结果,所有信息必须来自工具。 """ class AgentCore: def __init__(self, model_client: ModelClient, registry: ToolRegistry, max_steps: int = 20): self.model_client = model_client self.registry = registry self.max_steps = max_steps def _format_messages(self, messages): return [Message(role=m["role"], content=m["content"]) for m in messages] def _parse_response(self, text: str): """解析模型输出,容忍 markdown 代码块包裹。""" text = text.strip() if text.startswith("```"): text = re.sub(r"^```(?:json)?\s*", "", text) text = re.sub(r"\s*```$", "", text) return json.loads(text) def run(self, task: str): messages = [ Message(role="system", content=SYSTEM_PROMPT), Message(role="user", content=task), ] for step in range(1, self.max_steps + 1): print(f"--- Step {step} ---") try: raw_output = self.model_client.generate(messages, self.registry.get_schemas()) except Exception as e: return f"模型调用失败: {e}" messages.append(Message(role="assistant", content=raw_output)) try: parsed = self._parse_response(raw_output) except json.JSONDecodeError: # 把错误反馈给模型,让它重新输出合法 JSON messages.append(Message( role="user", content="你的输出无法解析为合法 JSON,请严格按照指定格式重新输出。", )) continue tool_name = parsed.get("tool") if tool_name is None: return parsed.get("thought", "任务完成") args = parsed.get("args", {}) result = self.registry.call(tool_name, **args) if result.error: print(f"工具 {tool_name} 出错: {result.error}") else: print(f"工具 {tool_name} 返回: {result.output[:200]}") messages.append(Message( role="user", content=f"工具 {tool_name} 执行结果:\n{result.output}\n错误信息:{result.error or '无'}", )) return f"达到最大步数 {self.max_steps},任务未完成,请增加 max_steps 或简化任务。"

这里有几个关键设计:

  1. 每次模型回复都会追加到messages里,保证对话历史连续。如果解析失败,把错误信息作为新的 user 消息塞回去,让模型自行修复。这一步就是"自我纠错"的雏形。
  2. 工具执行结果和错误信息一起返回给模型。即使工具调用失败,模型也能根据错误信息调整策略,而不是直接崩溃。
  3. print输出用于调试,你可以改成日志,方便追踪每一步发生了什么。

这个循环就是 Pi 这类生产级 Agent 的最小形态了。真实产品会在这个循环里加很多细节,比如:工具结果超长时怎么压缩、模型输出不稳定时怎么重试、循环超过多少步要强制停止、怎么让用户中途打断。但核心骨架就是这样。

3.5 组装起来

# main.py from tool_registry import ToolRegistry from model_client import ModelClient from agent_core import AgentCore from builtin_tools import ls, read_file, run_command def main(): registry = ToolRegistry() registry.register( "ls", "List directory contents", {"type": "object", "properties": {"path": {"type": "string", "description": "目录路径,默认当前目录"}}}, ls, ) registry.register( "read_file", "Read a file's content", {"type": "object", "properties": {"path": {"type": "string", "description": "文件路径"}, "max_chars": {"type": "integer", "description": "最多读取的字符数,默认8000"}}}, read_file, ) registry.register( "run_command", "Run a shell command", {"type": "object", "properties": {"command": {"type": "string", "description": "要执行的命令"}}}, run_command, ) client = ModelClient( base_url="https://api.openai.com/v1", # 改成你的兼容接口地址 api_key="your-api-key", model="gpt-4o-mini", ) agent = AgentCore(model_client=client, registry=registry, max_steps=15) result = agent.run("请统计当前目录下 Python 文件中 TODO 注释的数量,并列出每个文件的数量。") print("\n最终结果:\n", result) if __name__ == "__main__": main()

到这一步,你已经有了一个完整的、能跑的最小 Coding Agent。它可以在任意本地目录上自主列目录、读文件、跑命令,然后根据观察结果继续决策。

4. 实测:让最小 Agent 完成一次真实仓库任务

理论说完了,来点实际的。我在一个包含多个 Python 文件的小仓库里跑了一下上面这个 Agent,任务是:"统计当前目录下 Python 文件中 TODO 注释的数量,并列出每个文件的数量。"

4.1 预期的 Agent 行动路径

如果这个任务交给一个经验丰富的人类开发者,你大概会这么做:先看看目录结构,找到 Python 文件,然后逐个打开或者用grep搜索 "TODO" 字样,最后汇总成报告。

模型如果足够聪明,也应该走类似的路径。第一次跑的时候,模型是这样行动的:

  • 第一步:调用ls查看目录结构。
  • 第二步:调用run_command执行grep -n "TODO" --include="*.py" -r .直接搜索。
  • 第三步:根据搜索结果,再决定是否需要读文件来确认上下文。
  • 第四步:汇总结果,输出最终报告。

实际上,模型在第三步产生了分歧。它搜到了一些 TODO,但无法确定某些 TODO 是注释里的还是字符串里的,于是它打算用read_file去看具体文件。这个行为让我挺惊讶的,因为我没有在系统提示词里教它"不确定时要读原文",它自己学会了。

4.2 真实的运行输出节选

下面是我实际跑出来的输出片段(已经简化掉中间的思考内容):

--- Step 1 --- 工具 run_command 返回: client.py g2.py utils.py --- Step 2 --- 工具 run_command 返回: utils.py:12: # TODO: 重构这个函数的命名 client.py:45: # TODO: 需要补充超时重试逻辑 client.py:78: # TODO: 处理连接池耗尽的情况 --- Step 3 --- 工具 read_file 返回: (这里读取了 client.py 的部分内容) --- Step 4 --- 工具 run_command 返回: (执行了更精确的 grep,确认 TODO 行号) 最终结果: 当前仓库中 Python 文件的 TODO 注释统计如下: - utils.py: 1 处 - client.py: 2 处 共 3 处 TODO 注释。

说实话第一次跑通的时候,我盯着终端看了好一会。这个最小实现没有语义搜索、没有 memory、没有并行工具调用,但完成这个任务已经足够了。这说明一个很关键的事实:Coding Agent 的核心能力不来自于复杂的架构,而来自于"模型 + 工具 + 反馈循环"这三者的有效组合。

4.3 实测中暴露的三个问题

测试当然不是一帆风顺的。我遇到了三个典型问题,每个都值得单独说一说。

第一个问题是 JSON 解析失败。模型偶尔会输出一段带解释文字和代码块的回复,而不是纯 JSON。我在_parse_response里加了容忍逻辑,但还是会遇到格式错乱的情况。处理方式就是把错误反馈给模型重试,这比用正则硬解析要可靠得多。

第二个问题是工具结果太长导致上下文爆炸。有一次模型调用run_command执行了一个输出很长的命令,我把完整输出塞回给模型,直接导致下一次模型调用因为超出上下文限制而失败。后来我在工具函数的max_chars和输出截断上加上了更激进的限制。

第三个问题是最危险的:陷入死循环。模型在某个任务上反复调用ls,就是不做决策。如果没有max_steps保护,这个循环会一直调用 API,烧掉不少钱。从这个角度看,最大步数限制不是可选项,而是必需品。

5. 从最小版到生产级:Pi 骨架里的工程加固点

跑通了最小闭环,再看 Pi 这类生产级 Coding Agent,你会发现它们多出来的东西并不是"更聪明的模型",而是围绕这个循环做的大量工程加固。我梳理了五个最重要的加固方向。

5.1 上下文管理与 Token 预算

最小版本里,每次模型调用都是把完整历史发过去。历史越长,Token 消耗越大,延迟越高,而且模型可能会被早期无关信息干扰。生产级 Agent 至少会做三件事:

  1. 截断:超长工具结果不完整回传,而是摘要或只保留关键部分。
  2. 裁剪历史:超过一定轮次后,把早期对话压缩成摘要。
  3. 上下文检索:不是把整个仓库读进来,而是根据任务动态检索相关文件。

我见过一个很实用的做法:把 ToolResult 超过 2000 字的内容自动截断,并附带一句"[结果过长已截断,如需完整内容请针对性读取文件]"。模型能理解这个提示,并且会转而用更精准的方式去读取它想要的片段。这比无限扩大上下文窗口要经济得多。

5.2 沙箱与安全边界

最小版里,run_command直接用shell=True执行任意命令。这在你自己电脑上跑没问题,但生产级系统绝对不能这么干。Pi 这类项目里,命令执行通常会包在容器、虚拟机或者至少是一个受限的工作目录里。原因很简单:模型可能被诱导执行危险命令,或者模型自己"灵机一动"执行了删除操作。沙箱的意义不是防恶意攻击,而是防止模型犯低级错误造成不可逆损失。

退一步讲,即便不加沙箱,也至少要加一层"高危命令确认"机制:把rm -rf、git push、pip install这类命令拦下来让用户确认。我在代码里没有加这个,但在真实项目里我会强烈建议加。

5.3 流式输出与用户体验

最小版里,每一次模型调用用户都要干等十几秒甚至几十秒,不知道系统在干什么。生产级 Agent 会做流式输出,让模型思考过程像打字机一样实时显示出来,用户能判断它有没有走偏。这个需求看似只是体验优化,实际上很重要:一个完全黑盒的 Agent 用户是不敢放心使用的。哪怕只是把中间步骤打印出来,都算进步。

5.4 并发与多会话

生产级 Agent 通常需要支持多用户、多会话同时运行,这会引出会话隔离、数据库存储、任务队列等一堆问题。Pi 的骨架在处理这个问题时,把"会话状态"单独抽象成了一个存储层,而不是把它放在内存 list 里。这样进程重启、多实例部署、断线恢复都能支持。最小版里我把messages直接放在run()函数里,是为了方便看逻辑,但生产系统必须把状态外置。

5.5 费用与速率控制

这个点很容易被忽略。Coding Agent 一次任务可能要调用几十次模型 API,一次完整跑下来费用可能很高。生产级系统会做 token 级费用统计、单任务预算上限、模型降级策略(比如简单任务用便宜模型,复杂任务才用强模型)。我见过一个项目,就因为忘了加费用上限,某个 Agent 实例在后台空转了一晚上,产生了上千次 API 调用。这个教训很惨痛。

6. 让 Coding Agent 真正好用的几个关键经验

最后聊一些我在实际开发和使用中积累的经验,这些东西不写进代码,但比代码更重要。

6.1 模型选择没有银弹

我测试下来,在 Coding Agent 场景里,模型的能力差距会被循环机制放大。弱模型在第一步就可能格式错误,或者调用了不存在的工具名,然后循环变成"报错 -> 重试 -> 报错",不仅慢还费钱。选模型的原则是:先选你预算内最强的模型跑通流程,再考虑降级。降级时要小心,不是所有模型都擅长严格遵循 JSON 格式输出,格式敏感的任务不要用太弱的模型。

6.2 工具定义的质量直接影响成功率

工具描述不要写得太简单。同样是ls工具,参数描述写"路径"和写"要列出的目录路径,默认当前目录。注意区分绝对路径和相对路径,列出文件时同时显示文件和目录名"效果完全不同。模型的工具选择准确性很大程度上取决于工具描述和参数描述的质量。这也是为什么很多人发现自己写 Agent 效果不如商业产品——不是模型问题,是工具定义不够好。

6.3 永远假设模型会犯错

生产级 Coding Agent 设计的核心原则是"宁可多一步反馈,也不要相信模型一次到位"。常见假设包括:模型可能输出非法 JSON、可能调用不存在的工具、可能传错参数类型、可能陷入重复循环、可能编造工具结果。每一个假设都应该在代码里对应一个防护措施。我的最小实现里只处理了前两种,但你已经能看到这种设计思路带来的稳定性差别。

6.4 不要追求"一次完成大任务"

把一个大任务直接扔给 Agent,期望它一口气完成,是新手最容易踩的坑。我的经验是:把任务拆小,每步只做一件事,然后让 Agent 逐步推进。最小版的 Agent 虽然能在 4 步内完成 TODO 统计任务,但如果你让它"把这个仓库重构一遍",它大概率会陷入混乱。生产级 Agent 会结合任务分解、阶段性校验、用户中途确认来应对这种场景。

写在最后

回到开头的问题:Coding Agent 到底是怎么跑起来的?答案藏在那个简单得有点无聊的循环里:模型看到一个状态,决定下一步做什么,调用工具,观察结果,再决定下一步。Pi 这类生产级 Agent 的骨架再复杂,也逃不出这个循环。它多出来的那些东西——沙箱、上下文管理、流式输出、多会话、费用控制——都是为了让这个循环更稳定、更可控、更经济地运转。

如果你也想做一个自己的 Coding Agent,我的建议很简单:先照着这篇文章把最小循环跑通,然后把你第一个真实任务跑一跑,你会立刻发现哪里需要加固。比如我自己在跑完第一个任务之后,第一个想加的功能就是"给工具结果加速摘要",因为上下文真的消耗得太快了。从最小骨架出发,好过从庞大复杂的框架出发,这条路我替你验证过了,值得走。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询