★ FEATURED ARTICLE 前阵子我用 Claude 和 Codex 这类工具处理了一个旧仓库的 issue看着它自己读代码、改文件、跑测试我就在想这些东西的内部到底是怎么跑的后来我去翻了 Pi 这类生产级 Coding Agent 的骨架说实话把它的模块拆开之后你会发现核心原理一点都不神秘。这篇文章我想从零开始手写一个最小可用的 Coding Agent用它把 Pi 生产级 Agent 的骨架讲明白顺便分享我实际跑任务时踩过的坑。适合谁看如果你写过 Python、用过 ChatGPT 或 Claude 的 API但还没搞清楚 Agent 的 loop、工具调用、上下文管理这些概念这篇文章就是给你准备的。我会把代码直接贴出来你复制过去改一改就能跑。1. 动手前先想清楚Coding Agent 的本质是带工具的循环很多人第一次接触 Coding Agent 会把想复杂了以为里面有什么神奇的规划算法或者自动推理引擎。其实拆开来看一个 Coding Agent 的核心就是四样东西模型、工具、状态、控制循环。模型负责思考也就是根据当前状态决定下一步做什么工具负责动手比如读文件、执行命令、搜索代码状态负责记忆记录模型看到了什么、做过什么控制循环负责调度不断重复思考 - 调用工具 - 把结果喂回去 - 再思考这个过程直到任务完成。这个套路在学术界有个名字叫 ReActReasoning Acting本质上是把推理和行动交替进行。Coding Agent 只是把 ReAct 循环的应用场景聚焦到了代码库上模型需要能够浏览代码、理解代码、修改代码并且要能看到自己改动的结果。1.1 从 Pi 的骨架里看到的抽象我看了 Pi 的实现思路它是一款面向生产环境的开源 Coding Agent主打让模型自主完成编码任务它的骨架其实就是一句话任务输入 - 上下文感知 - 工具调用 - 结果反馈 - 再决策。但这五个环节每一个在实际落地时都有大量工程细节。比如上下文感知不是简单把整个仓库塞给模型而是需要按需检索用户问的是一个文件里的一个函数系统就去定位这个文件、提取函数附近代码、补上相关依赖而不是把 10 万行代码全部发给模型。这一步直接决定了 Coding Agent 能不能处理真实规模的仓库。工具调用也不是简单让模型输出一段文字而是需要一套严格的工具定义协议。模型输出的是我想调用这个工具参数是这些系统需要解析这个结构、校验参数、执行工具、把结果转换成文本再回传给模型。这里任何一个环节出错整个循环就会断掉。1.2 最小可用版本的边界明确了原理之后要给自己划一条最小的边界。我的目标不是实现一个能和 Pi 或者 Codex 媲美的产品而是实现一个能跑通完整闭环的最小系统。它需要满足三个条件能接入一个代码库至少在本地目录上操作。能调用至少三个工具列目录、读文件、执行命令。能自我纠错工具调用失败或者模型 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 CoreContext 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, errorfUnknown tool: {name}) try: result self._tools[name][fn](**kwargs) return ToolResult(outputstr(result)) except Exception as e: return ToolResult(output, errorstr(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, encodingutf-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 stdoutstderr. try: result subprocess.run(command, shellTrue, capture_outputTrue, textTrue, timeout30) 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: fBearer {self.api_key}} resp requests.post(url, jsonpayload, headersheaders, timeout120) resp.raise_for_status() data resp.json() return data[choices][0][message][content]我故意没有用某个具体厂商的 SDK而是直接调 OpenAI 兼容的 HTTP 接口这样你可以把base_url换成任何兼容的供应商。temperature0是为了让模型输出更稳定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(rolem[role], contentm[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(rolesystem, contentSYSTEM_PROMPT), Message(roleuser, contenttask), ] 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(roleassistant, contentraw_output)) try: parsed self._parse_response(raw_output) except json.JSONDecodeError: # 把错误反馈给模型让它重新输出合法 JSON messages.append(Message( roleuser, 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( roleuser, contentf工具 {tool_name} 执行结果\n{result.output}\n错误信息{result.error or 无}, )) return f达到最大步数 {self.max_steps}任务未完成请增加 max_steps 或简化任务。这里有几个关键设计每次模型回复都会追加到messages里保证对话历史连续。如果解析失败把错误信息作为新的 user 消息塞回去让模型自行修复。这一步就是自我纠错的雏形。工具执行结果和错误信息一起返回给模型。即使工具调用失败模型也能根据错误信息调整策略而不是直接崩溃。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 files 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_urlhttps://api.openai.com/v1, # 改成你的兼容接口地址 api_keyyour-api-key, modelgpt-4o-mini, ) agent AgentCore(model_clientclient, registryregistry, max_steps15) 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 至少会做三件事截断超长工具结果不完整回传而是摘要或只保留关键部分。裁剪历史超过一定轮次后把早期对话压缩成摘要。上下文检索不是把整个仓库读进来而是根据任务动态检索相关文件。我见过一个很实用的做法把 ToolResult 超过 2000 字的内容自动截断并附带一句[结果过长已截断如需完整内容请针对性读取文件]。模型能理解这个提示并且会转而用更精准的方式去读取它想要的片段。这比无限扩大上下文窗口要经济得多。5.2 沙箱与安全边界最小版里run_command直接用shellTrue执行任意命令。这在你自己电脑上跑没问题但生产级系统绝对不能这么干。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我的建议很简单先照着这篇文章把最小循环跑通然后把你第一个真实任务跑一跑你会立刻发现哪里需要加固。比如我自己在跑完第一个任务之后第一个想加的功能就是给工具结果加速摘要因为上下文真的消耗得太快了。从最小骨架出发好过从庞大复杂的框架出发这条路我替你验证过了值得走。
阅读完成 · 觉得有帮助？
