做 Agent 开发这一两年,我最大的感受是:套一个现成框架跑通 demo 很容易,但当你真的想弄明白 Agent 内部到底在做什么的时候,唯一靠谱的方法就是自己动手把骨架拆开、写一遍。前阵子我参照 pi-agent(这里我用这个代号指代那种代码量不大、结构清晰、很适合作为教学参考的轻量 Agent 实现)从零手写了一个几百行的小 Agent。写完之后,之前用框架时那些模模糊糊的概念——工具调用怎么触发、消息上下文怎么组织、主循环到底在循环什么——一下子全串起来了。这篇文章就是那次实践的全记录,从最小架构拆解到落地代码、再到调试现场踩过的坑。如果你是那种已经会用大模型接口、但还想往 Agent 内部钻一层的人,这篇应该能给你省不少弯路,跟着复刻一个能跑的小 Agent 完全可行。
1. 剥开 Agent 开发的本质:先搞清楚要复刻的东西是什么
1.1 Agent 不是一次调用,而是一个闭环
很多人以为 Agent 就是“调一下大模型接口,把回答拼到应用里”。真这么理解的话,后面写出来的东西顶多算一个带提示词外壳的聊天机器人。Agent 的核心其实是那个“循环”:观察 → 决策 → 行动 → 再观察,直到任务收敛。
我习惯用一个类比跟人解释:Agent 像是一个刚入职的实习生。大脑是大模型,负责想问题、做判断;手是工具,比如查天气、算数、读写文件、调接口;记事本是记忆,用来记住前面已经聊过什么、已经做过什么;而把它串起来的那个人,就是主循环 runner。
实习生接到任务后,会先想“这件事我能不能直接答”,不能的话就去查资料、调工具,查完结果还要回头看一眼结果对不对,再决定下一步。这就是闭环。单次问答不算 Agent,闭环才算。你写 Agent 的时候,本质上是在搭一个让“思考”和“行动”互相反馈的传送带。
Agent 的最小组成可以拆成五块:
- 推理核心:负责理解任务、生成决策,通常是大模型接口。
- 规划能力:把大任务拆成小步骤,或者是简单地决定“下一步调用哪个工具”。
- 记忆模块:短期记忆就是上下文窗口里的历史消息,长期记忆则负责跨会话的知识存储。
- 工具集:暴露给模型的函数,模型通过结构化参数调用。
- 执行与反馈:真正去执行工具、捕获结果或异常,然后喂回上下文。
我这个项目里没有把规划单独做成一个模块,因为对一个小 Agent 来说,规划能力可以先融进提示词和主循环里。实现完你就会发现,很多号称“规划”的功能,其实就是引导模型在步骤里多走几轮。
1.2 参考实现给我的三条设计原则
选定 pi-agent 作为参考,不是因为它的功能有多全,而是因为它“小而完整”。这种项目用来做参考有个好处:你能在几千行甚至更少的代码里完整读完整个 Agent 的生命周期,不会被一堆插件、协议、抽象类淹没。我从里面提炼出三条原则,直接决定了我自己实现的方向。
第一,分层解耦。LLM 封装、工具注册、消息管理、主循环,各自独立成模块。工具哪怕挂在另一个文件里,也不影响主循环的运行。这带来一个直接好处:调试的时候可以单独测“工具层”或“记忆层”,不用每次把整个 Agent 拉起来。
第二,工具与模型解耦。模型不关心工具内部怎么实现。工具注册器只告诉模型“有什么工具、参数是什么、返回值是什么”,剩下的事情由主循环调度。我一开始想图省事,把工具逻辑直接写在主循环里,后来发现每加一个工具主循环就得改一次,非常被动。改成注册制之后,加工具只是往注册表里丢一个函数的事。
第三,一切可替换。模型供应商、记忆策略、工具实现,都可以单独换。因为主循环只依赖接口,不依赖具体实现。这也是我认为参考实现最值得学的地方:它逼着你用面向接口的方式去写 Agent,而不是把所有东西糊成一个类。
2. 动手前的关键决策:语言、骨架与消息模型
2.1 为什么选 Python 和最小依赖
技术选型上我几乎没有犹豫,直接用了 Python。不是因为 Python 性能多好,而是因为 Agent 开发这个场景里,Python 有天然优势:主流大模型服务商基本都优先提供 Python SDK 或兼容的 HTTP 接口,生态调起来最顺;调试方便,改完代码立刻能重跑;核心逻辑真写起来也就一两百行,根本不需要复杂构建系统。
依赖方面我压到最低:只需要 requests 发 HTTP 请求,加上标准库里的 json、dataclasses、logging。这里有一点想强调:写这种教学性质的 Agent,不建议上来就引 Agent 框架。框架本身没错,但它的编排逻辑会盖住你自己对主循环的理解。真等你自己写完一遍主循环,再回去看框架文档,会发现里面的概念都能对号入座了。
这个项目也不需要数据库、不需要缓存中间件。因为目标是“跑通最小闭环”,所有状态都放内存里。等后面要接生产,再去考虑持久化。
2.2 目录结构与模块划分
我实际使用的目录结构长这样:
pi_agent/ ├── agent/ │ ├── __init__.py │ ├── core.py # 主循环入口 │ ├── llm.py # LLM 接口封装 │ ├── tools.py # 工具注册与执行 │ ├── memory.py # 上下文管理 │ └── schemas.py # 消息与工具的数据结构 └── app/ └── main.py # 组装并启动每个文件的职责很清楚:llm.py 只负责跟模型服务打交道,把 messages 列表发出去,把响应结构解析成标准格式;tools.py 维护一个全局工具注册表和统一的执行入口;memory.py 负责截断、压缩消息历史;schemas.py 定义消息、工具描述这些基础类型;core.py 把前面几块串起来跑主循环;main.py 负责注册具体工具、设定模型参数、启动交互。
模块边界之所以重要,是因为 Agent 项目一旦长起来,最让人头疼的问题就是“改一处崩一片”。分离干净之后,我做实验的时候可以直接在 main.py 里临时加工具,完全不用动 core.py。这看起来是很普通的工程习惯,但放到 Agent 这种“高度不确定”的项目里,它就是你排查问题的生命线。
2.3 统一消息模型:一切皆消息
Agent 的上下文本质上就是一份按顺序排列的消息列表。系统提示、用户问题、模型回答、工具调用结果,全部都是这列表里的一个元素。所以我用 dataclass 定义了一个统一的消息结构:
from dataclasses import dataclass, field from typing import Any, Optional @dataclass class ToolCall: name: str arguments: dict @dataclass class Message: role: str # system / user / assistant / tool content: str = "" # 普通文本内容 tool_calls: Optional[list] = field(default_factory=list) # assistant 发起的工具调用 tool_call_id: str = "" # 工具消息关联的调用ID name: str = "" # 工具消息对应的工具名这个结构里比较关键的是 tool_call_id。模型发起一次工具调用时会带一个唯一 ID,我们执行完工具之后,必须把结果以带相同 ID 的 tool 消息放回上下文,模型才能把“调用”和“结果”对应起来。你可以把它理解为邮件来往里的“原信编号”,没有这个编号,模型看到一坨工具结果只会一头雾水。
统一消息模型带来的另一个好处是:LLM 接口封装可以做得极其简单,就是“传入 messages 列表,返回新的消息”。主循环不用关心消息到底是用户说的还是工具返回的,它只需要决定“下一轮要不要继续循环”。这跟之后你去看各种编排框架的文档时的体验是一致的,因为它们内部也基本都是这种消息流。
3. 核心实现:短小但完整的主循环
3.1 写一个能对话的 LLM 封装
LLM 封装是这个项目里最简单但最容易踩坑的一层。我做的事情是:接收一个 messages 列表,调用兼容 Chat 补全接口,把响应的 message 对象原样抽出来返回。代码量不多:
import os import requests def chat_once(messages, tools=None): resp = requests.post( os.environ["LLM_API_ENDPOINT"], headers={"Authorization": f"Bearer {os.environ['LLM_API_KEY']}"}, json={ "model": os.environ.get("LLM_MODEL", "default"), "messages": messages, "tools": tools, "temperature": 0.1, }, timeout=60, ) resp.raise_for_status() data = resp.json() return data["choices"][0]["message"]返回的 message 可能长两种样子:一种是纯文本回复,里面只有 content;另一种是结构化动作,里面带着 tool_calls 字段。无论哪种,主循环都要把它原样追加到 messages 列表里。这里有个细节值得展开:模型返回的 message 会携带它自己的 role,保留这个 role 很重要,因为下一步工具结果会以 role="tool" 的消息接在后面。一串合法的消息顺序应该是 assistant(发起调用)→ tool(调用结果)→ assistant(继续思考)这样交替出现。
temperature 我设成了 0.1。Agent 的工具调用是强逻辑型任务,我不希望模型“发挥创造力”去随机选一个不存在的工具。等以后做创意型 Agent 再调高。
3.2 工具注册器:让函数变成 Agent 的手脚
没有工具动作的 Agent 只是一个问答机器人。为了让模型能够调用函数,我需要把每个函数描述成标准结构、挂到一个注册表里。实现也比较直接:
TOOL_REGISTRY = {} def register_tool(name, description, parameters): def decorator(fn): TOOL_REGISTRY[name] = { "fn": fn, "description": description, "parameters": parameters, } return fn return decorator def run_tool(name, arguments): if name not in TOOL_REGISTRY: raise ValueError(f"Unknown tool: {name}") return TOOL_REGISTRY[name]["fn"](**arguments)使用的时候只要在函数上加装饰器:
@register_tool( name="get_current_time", description="获取当前的日期和时间,返回字符串", parameters={ "type": "object", "properties": {}, }, ) def get_current_time(): return datetime.now().strftime("%Y-%m-%d %H:%M:%S")初次做工具注册的时候,我最容易忽略的是 description 的质量。模型是靠工具名、描述和参数说明来做决策的。你把描述写一行“获取当前时间”,它可能只会在用户明确问时间时才用;你要是写成“获取当前日期和时间,当需要了解现在是几点、今天是几号时使用”,它在很多临近场景下都会正确地主动调用。工具注册的本质是给模型写“使用说明书”,说明书越清晰,模型越不会乱来。
3.3 主循环:Agent 的灵魂
主循环是整个项目里最重要的一部分,放在 core.py 里。它的逻辑可以用伪代码概括:把当前消息列表发给模型;如果模型返回的是文本内容,结束;如果模型返回的是工具调用,逐个执行工具,把结果以 tool 消息追加进列表,然后回到第一步继续。直到模型不再申请调用工具,或者达到最大步数。
def run_agent(messages, max_steps=10): for step in range(max_steps): message = chat_once(messages, tools=get_tool_schemas()) messages.append(message) if not message.get("tool_calls"): return message["content"] for call in message["tool_calls"]: tool_name = call["function"]["name"] arguments = json.loads(call["function"]["arguments"]) try: result = run_tool(tool_name, arguments) output = str(result) except Exception as e: output = f"工具执行失败,请检查参数或换一种方式。错误信息: {e}" messages.append(Message( role="tool", content=output, tool_call_id=call["id"], name=tool_name, ).__dict__) return "达到最大步数,任务未能收敛"这段代码看着简单,但它回答了几个一直困惑我的问题。为什么 Agent 能“自动决定下一步”?因为模型在每一步都会看到所有历史消息,包括上一步工具的执行结果,它根据这些信息生成下一步行动。为什么会有“死循环”?因为模型可能反复调用同一个工具原地打转,所以必须有 max_steps。为什么工具结果要带 tool_call_id?因为它要跟模型发起的这次调用配对。把这些想明白了,你对 Agent 机制的理解基本就通了。
3.4 记忆的取舍:先把短期记忆做好
记忆是 Agent 开发里最容易过度设计的地方。我的建议是先不要碰向量库,先把短期记忆做对。短期记忆就是发往模型的消息列表,它直接受限于上下文窗口。实际写作里,我的 memory.py 只做两件事:一是控制消息数量,超过阈值就把最旧的对话消息裁剪掉,但保留系统提示和最近几轮的工具消息;二是记录当前会话的关键摘要,在系统提示里附上“历史摘要”字段。
截断的代码很简单:
MAX_HISTORY = 20 def trim_messages(messages, max_history=MAX_HISTORY): if len(messages) <= max_history: return messages system = [m for m in messages if m.role == "system"] recent = messages[-max_history:] return system + recent if system else recent但这里有个细节容易漏:裁剪消息时,不能把一轮完整的“assistant 发起调用 + tool 返回结果”拆散。如果 assistant 消息还在列表里,对应的 tool 结果却被裁掉了,模型就会看到一次没有结果的函数调用,然后一脸疑惑。我的处理方式是识别 tool_call_id,把整轮调用链作为一个整体来裁剪。
对于长时记忆(跨会话记住用户偏好、历史知识),我暂时没有做。原因很简单:小 Agent 的阶段还不需要,而且做了之后会增加大量调试成本。等你的 Agent 真的有了“记住事”的需求,再去评估向量库或者摘要库也不迟。
3.5 防御式输出:模型的话不能全信
模型返回的参数是 JSON 字符串,但这不是说它每次都能给出合法的 JSON。偶尔会多一个逗号、少一个引号,甚至把参数名拼错。所以我在 tool 参数解析上做了兜底:
def safe_load_args(raw: str) -> dict: try: return json.loads(raw) except json.JSONDecodeError: # 尝试去掉 markdown 代码块标记 cleaned = raw.strip().removeprefix("```json").removesuffix("```").strip() return json.loads(cleaned)如果还解析失败,就不要再盲目重试了。把“参数解析失败”作为工具消息返回给模型,让它自己重新组织参数。
工具执行本身的异常也有讲究。我不建议直接抛出让整个程序崩溃,而是捕获异常、转成一行清晰的错误描述放回上下文。模型在下一次循环里有很大概率会根据错误信息修正自己的行为。这就是 Agent 的“自我纠错”能力,它依赖的不是模型有什么魔法,而是我们把反馈渠道打通了。
另外提一嘴 Agent 安全。工具注册表里暴露给模型的函数必须是白名单制,别把 delete 文件、改系统配置这种危险操作直接挂上去。模型的判断是不确定的,权限最小化、操作可回溯才稳妥。
4. 实操全记录:用参考骨架跑一个“时间助手”
4.1 最小可运行版本
我实际做的示例 Agent 功能很朴素:能获取当前时间、能做整数加法,并在拿到结果之后用一句话做总结。选择这种任务是因为它需要连续两次工具调用,足够演示主循环的完整过程。
启动前的环境准备:
python -m venv .venv source .venv/bin/activate pip install requests python-dotenv export LLM_API_ENDPOINT="https://api.example.com/v1/chat/completions" export LLM_API_KEY="your_key_here" export LLM_MODEL="your_model"我的 main.py 里注册了上面两个工具,然后直接调用 run_agent。
messages = [ Message(role="system", content="你是一个小助手,可以使用工具回答用户问题。"), Message(role="user", content="现在几点?顺便帮我算一下 123456 + 789"), ] answer = run_agent([m.__dict__ for m in messages]) print(answer)第一次跑的时候有件事情让我印象很深:我以为模型会先回答“我看一下时间”,然后再算加法,结果它在一个 assistant 消息里同时发出了两个 tool_calls。这说明主流模型是支持并行工具调用的。我的主循环是顺序执行这两个调用,再把两次结果依次放回上下文,模型看到结果后,直接生成了一句完整的总结。
4.2 日志观察 Agent 的思考链
为了看清 Agent 每一步在干什么,我给主循环加了逐条日志。一次完整运行的日志长这样:
[step 1] 发送 messages 共 2 条 [assistant] content= tool_calls=[{"name":"get_current_time","arguments":{}},{"name":"add","arguments":{"a":123456,"b":789}}] [tool] get_current_time -> 2025-02-18 15:23:01 [tool] add -> 124245 [step 2] 发送 messages 共 5 条 [assistant] 当前时间是 2025-02-18 15:23:01。另外,123456 与 789 相加的结果是 124245。注意 step 2 发送的消息条数是 5,因为在原来的 2 条基础上,追加了 1 条 assistant 消息和 2 条 tool 消息。这正是闭环在数据层面的体现。我把日志结构写得跟 messages 里的字段一一对应,排查问题时就能直接判断“模型到底有没有看到工具结果”。
这也是为什么我一直强调日志要带完整上下文。很多 Agent 出了问题你不知道是模型没理解,还是工具结果没传回去。日志里把这些字段列清楚,问题定位快十倍。
4.3 错误注入:看模型怎么从反馈里恢复
写 demo 的时候我故意做了破坏性测试。我给 Agent 注册了一个除法工具,然后让它计算 1/0。工具直接抛了 ZeroDivisionError,我的主循环捕获之后返回了“工具执行失败:division by zero”。下一轮模型看到这个反馈,自动换了一个策略——它没有继续请求除零,而是回复“这个计算不合法,0 不能作为除数,建议提供其他数字”,整个过程非常自然。
这说明一个非常重要的事实:Agent 的鲁棒性有很大一部分来自“把错误反馈给模型”这个动作本身。模型不擅长在第一次调用时就保证绝对正确,但它非常擅长根据工具结果的错误信息修正自己。工程上的启示就是:不要在 Agent 工具层做太多“智能化修复”,老老实实把错误抛回去,让模型来决策,往往更省事,也更符合 Agent 的设计哲学。
4.4 与参考完整实现的差异对照
做完精简版之后,我拿它跟 pi-agent 参考实现的完整功能做了个对照。这张表可以帮你判断哪些功能是必须的,哪些是可以以后再补的:
| 功能点 | 参考实现 | 我的精简版 | 结论 |
|---|---|---|---|
| 工具注册机制 | 支持装饰器注册 | 支持装饰器注册 | 必须保留 |
| 主循环 | 支持循环与步数限制 | 支持循环与步数限制 | 必须保留 |
| 工具结果反馈 | 统一 tool 消息 | 统一 tool 消息 | 必须保留 |
| 短期记忆裁剪 | 有历史截断 | 有历史截断 | 必须保留 |
| 多智能体协作 | 支持多个角色协作调度 | 未实现 | 可以后补 |
| 长时记忆 | 支持向量检索 | 未实现 | 可以后补 |
| 并行工具调用 | 支持异步并发 | 顺序执行 | 按需加 |
| 任务规划器 | 单独规划步骤 | 靠提示词完成 | 可以后补 |
| 可观测日志 | 结构化追踪 | 简单日志 | 建议尽快加 |
这份对照给我最大的提醒是:Agent 的核心不是那些炫酷的高级功能,而是“调用工具、收集反馈、继续推理”这个基础闭环。闭环不牢,加再多记忆和规划都是空中楼阁。
5. 常见问题与排查技巧实录
5.1 高频问题速查表
前前后后折腾了不少问题,我把最常见的情况整理成一个速查表,适合贴在工位旁边:
| 现象 | 最常见原因 | 解决思路 |
|---|---|---|
| 模型完全不调用工具 | tools 参数格式不对,或模型不支持 | 检查 tools schema,确认工具描述清晰 |
| 工具结果没生效 | 消息列表丢字段,或 tool_call_id 不匹配 | 按标准消息结构追加 tool 消息并保留 ID |
| 上下文爆炸 | 没有做历史消息裁剪 | 加 MAX_HISTORY,按整轮对话裁剪 |
| Agent 陷入死循环 | 模型反复调用同一工具 | 设 max_steps,检测重复调用,提前终止 |
| 工具参数解析失败 | 模型生成了非法 JSON | 加 JSON 兜底清理,失败信息回喂模型 |
| 接口偶发超时 | 网络波动或模型服务超时 | 加指数退避重试,超时前把降级文案返回 |
5.2 模型“不调用工具”八成是 schema 的问题
我第一次遇到这个问题时,第一反应是“模型怎么这么笨”。后来逐项排查,发现根本不是模型的问题。tools 参数的结构必须完全匹配模型服务要求的格式,一开始我漏掉了 parameters 里的 type:"object" 这一层,模型直接忽略了我的工具。
排查顺序建议是:先用一条英文或中文的短消息测试,确认 tools 参数确实传进了请求体;再检查 functions/tools 的 JSON 结构是否合法;最后确认工具描述的措辞是否足够具体。大多数情况下改一下描述,模型就会用了。
5.3 上下文爆炸:裁剪时要保护“工具调用链”
随着对话变长,消息列表会越积越多。如果不控制,很快会顶到模型上下文窗口上限。最初的粗暴截断是把 messages 直接截掉前 N 条,但这导致了那个前面提到的拆散问题:assistant 消息被保留,而对应的 tool 消息被裁掉,模型看到一次没有结果记录的调用。
我的修复方式是:裁剪时先扫描 assistant 消息里所有 tool_call_id,把这些 ID 对应的 tool 消息标记为“同一轮”,然后把整轮作为一个裁剪单位。这样既控制长度,又保证了每一轮的完整性。
5.4 死循环:步数限制之外再加重复检测
max_steps 是兜底,但更优雅的解法是加重复检测。如果模型连续两三轮都在请求同一个工具、参数也相似,说明它在原地打转。这时候直接把“重复调用,请换一种方式或直接回答用户”作为 tool 消息塞回去,往往能打断循环。
我还在主循环里统计了每个工具的调用次数,超过阈值就把它从“可调用列表”里临时摘掉。这个设计在单个 Agent 项目里可能用不太上,但如果你后面做自主 Agent,这个思路非常有用,因为“越界调用”是真实存在的风险。
5.5 接口超时与异常:重试要带退避,降级要带文案
大模型接口不是银弹,它也会超时、限流、返回 5xx。我最初的实现是 requests 抛异常后整个程序崩掉。后来加了简单的指数退避重试:第一次等 1 秒,第二次等 2 秒,第三次等 4 秒,最多三次。重试之间还会检查剩余步数,如果步数快用完了就不重试了,直接把“服务暂时不可用,请稍后再试”作为答案返回。
这种降级策略看起来简单,但对用户体验和排查效率都很重要。否则一次偶发的网络抖动,就把一个本来正常的 Agent 变成了“坏掉的”应用。
6. 从玩具到可用:我的进阶路线与个人心得
6.1 生产化要补的四件事
demo 跑通之后,我并没有直接把它拿去接真实业务。因为从玩具到可用,中间还差四件事:
第一,可观测性。我给每条工具调用打上了 trace_id,记录完整消息流、工具入参、返回值、耗时。没有这些记录,你根本无法回答“这个 Agent 为什么在线上表现不一致”这个问题。
第二,幂等与容错。工具调用可能会被重试,如果工具本身有副作用(比如发邮件、改订单状态),必须做幂等控制。我的做法是给每次调用生成唯一 request_id,工具内部用这个 ID 做去重。
第三,评估集。Agent 是概率系统,那你就要用“评估集”去衡量它改提示词前后的表现。我准备了十几个典型任务,每次改动都跑一遍,看正确率有没有下降。这比任何自信都有说服力。
第四,安全底线。工具白名单、敏感操作拦截、上下文里不塞密钥,这些是生产环境的红线。模型的不确定性决定了你不能把权限边界交给模型去把握。
6.2 手搓一遍之后,再回去看框架
很多人在“自己写 Agent”和“用框架”之间纠结。我的真实体会是:这两件事不冲突,但顺序很重要。你自己手写过一次主循环、踩过一遍消息结构的坑之后,再去看社区里那些编排框架的文档,每个概念都能跟实际代码对上号。那时候框架对你来说不是黑盒,只是一个帮你省写样板代码的工具。
反过来的顺序就比较痛苦。先被框架养习惯,遇到问题就加依赖、翻文档,始终不知道底层发生了什么。一旦框架的某个抽象跟你预期的场景对不上,排查起来无从下手。
这个 demo 做完之后,我自己还在继续迭代。先把并行工具调用加上,把短期记忆升级成“摘要 + 关键信息”两层,然后做一点简单的任务规划。但不管怎么加,那套主循环骨架我不会动。因为它是整个 Agent 的地基,地基稳了,上面爱盖几层盖几层。
最后再说一个我踩过几次坑之后形成的习惯。写 Agent 代码时,我会把每个工具执行结果都记录到日志里,并且连同当时的消息列表一起存一份快照。这听起来麻烦,但将来你治理 Agent 的时候,会发现这些快照就是最好的调试数据。Agent 的行为是不确定的,你只有保住完整的运行现场,才有可能复盘和优化它。这个习惯让我在后续迭代里省了非常多的时间,你可以直接抄走。