外观
从零构建一个最小 Agent
学 Agent 最忌讳的一件事:框架学了一堆,真问「Agent 到底是什么」,答不上来。这页的解法很直接——不用 LangChain、不用任何框架,只用 LLM API 和一个 while 循环,从零手写一个能跑的 Agent。写完你会明白:所谓 Agent 框架,封装的不过是你今天写的这几百行代码,外加一堆工程细节。
全页分四版迭代,每版解决一个真实痛点:
v0 最小 Agent Loop → 能跑,但会死循环、会崩
v1 工具错误回传 + 步数上限 → 能自愈,不会失控
v2 todo 规划工具 → 长任务不跑偏
v3 上下文压缩 → 长会话不爆 context window准备工作只有三样:Python 3.10+、pip install openai(1.x 版本)、一个支持 tool calling 的模型 API Key。代码用的是 OpenAI Chat Completions 格式,这是事实标准——OpenAI 官方、DeepSeek、通义千问、Kimi 等 OpenAI 兼容端点都能直接跑,换个 base_url 就行。
一、先想清楚:Agent 的最小内核是什么
剥掉所有框架的包装,Agent 的内核是一个循环:
┌──────────────────────────────────────────┐
│ │
▼ │
┌───────────┐ tool_calls ┌────────────┐ │
│ LLM │───────────────▶│ 执行工具 │ │
│ (带着工具 │ │ (你的代码) │ │
│ 和上下文) │◀───────────────┴────────────┘ │
└───────────┘ 把工具结果塞回 messages │
│ │
└── 不再调工具 = 给出最终回答 → 退出循环 ─────┘这就是 Agent Loop 的全部本质。一轮迭代里只有四件事:
- 把「对话历史 + 工具 schema」发给模型;
- 模型要么返回最终文本,要么返回一个或多个
tool_calls(函数名 + JSON 参数); - 你的代码执行对应函数,把结果以
role: "tool"消息塞回历史; - 回到第 1 步,直到模型不再调工具。
一个关键认知
LLM 从来不「执行」工具。它只是生成一段结构化文本说「我想调用 read_file,参数是 {"path": "notes.txt"}」。真正执行的是你的 Python 代码——所以工具的权限边界、超时、错误处理全是你的责任,模型管不了。这也是 工具与 MCP 和 Agent 安全 两页反复讲边界控制的原因。
二、v0:50 行的 Agent Loop
下面是完整可运行的 v0。它有两个真实工具:read_file(读文件)和 run_shell(执行白名单内的只读命令)。注意两个安全设计——这是底线,不是可选优化:
read_file把路径强制限制在workspace/内,防目录穿越(模型给出../../etc/passwd会被拒);run_shell用命令白名单 +shell=False(shlex.split后数组传参),杜绝注入和写操作。
python
# mini_agent_v0.py —— 一个能跑的最小 Agent
# 依赖:pip install openai;环境变量 OPENAI_API_KEY
import json
import shlex
import subprocess
from pathlib import Path
from openai import OpenAI
client = OpenAI() # 兼容端点示例:OpenAI(base_url="https://api.deepseek.com", api_key="...")
MODEL = "gpt-5-mini" # 任何支持 tool calling 的模型都行
# 工具只允许在这个目录里活动,防止模型乱翻磁盘
WORKSPACE = Path("./workspace").resolve()
SYSTEM_PROMPT = """你是一个文件操作助手。
- 用 read_file 查看文件内容,用 run_shell 执行只读命令(ls/cat/grep/find/wc 等)。
- 所有路径都相对工作目录 workspace/。
- 回答前先收集足够的信息;信息够了就直接用中文给出最终答案,不要再调工具。
"""
# 工具 schema:模型靠 description 决定何时调、怎么填参数,写清楚就是生产力
TOOLS = [
{
"type": "function",
"function": {
"name": "read_file",
"description": "读取工作目录下一个文本文件的内容",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "相对 workspace 的文件路径"},
},
"required": ["path"],
},
},
},
{
"type": "function",
"function": {
"name": "run_shell",
"description": "执行一条只读 shell 命令(白名单:ls/cat/grep/find/pwd/wc/head/tail/echo)",
"parameters": {
"type": "object",
"properties": {
"command": {"type": "string", "description": "要执行的命令"},
},
"required": ["command"],
},
},
},
]
ALLOWED_CMDS = {"ls", "cat", "grep", "find", "pwd", "wc", "head", "tail", "echo"}
def safe_path(p: str) -> Path:
"""把模型给的路径限制在 WORKSPACE 内,防目录穿越"""
full = (WORKSPACE / p).resolve()
if not str(full).startswith(str(WORKSPACE)):
raise ValueError(f"路径越界: {p}")
return full
def read_file(path: str) -> str:
p = safe_path(path)
if not p.is_file():
return f"error: 文件不存在 {path}"
return p.read_text(encoding="utf-8", errors="replace")[:4000] # 截断,保护 context
def run_shell(command: str) -> str:
argv = shlex.split(command)
if not argv or argv[0] not in ALLOWED_CMDS:
return f"error: 命令被拒绝(白名单外): {command}"
try:
out = subprocess.run(argv, cwd=WORKSPACE, capture_output=True, text=True, timeout=10)
return (out.stdout + out.stderr)[:4000] or "(无输出)"
except subprocess.TimeoutExpired:
return "error: 命令超时(10s)"
except FileNotFoundError:
return f"error: 命令不存在 {argv[0]}"
DISPATCH = {"read_file": read_file, "run_shell": run_shell}
def run(task: str) -> str:
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": task},
]
while True: # v0 故意不加任何护栏,先观察会发生什么
resp = client.chat.completions.create(model=MODEL, messages=messages, tools=TOOLS)
msg = resp.choices[0].message
# 模型的原始回复必须进历史,tool 消息靠 tool_call_id 与之配对
messages.append(msg.model_dump(exclude_none=True))
if not msg.tool_calls: # 停止条件:模型不再调工具 = 给出最终回答
return msg.content
for tc in msg.tool_calls: # 一轮可能并行调多个工具
args = json.loads(tc.function.arguments)
result = DISPATCH[tc.function.name](**args)
print(f" [tool] {tc.function.name}({args}) -> {result[:60]}...")
messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})
if __name__ == "__main__":
WORKSPACE.mkdir(exist_ok=True)
(WORKSPACE / "notes.txt").write_text("本月目标:跑通 mini agent\n预算:3000 元\n", encoding="utf-8")
print(run("workspace 里的 notes.txt 写了多少预算?"))跑起来的典型轨迹(每次细节不同,但骨架一致):
$ python mini_agent_v0.py
[tool] run_shell({'command': 'ls'}) -> notes.txt...
[tool] read_file({'path': 'notes.txt'}) -> 本月目标:跑通 mini agent...
notes.txt 里写的预算是 3000 元。几个值得停下来想的点:
msg.model_dump(exclude_none=True)不能省也不能乱改。 Chat Completions 协议要求:assistant 的tool_calls消息和后续role: "tool"消息通过tool_call_id一一配对,少一条、顺序错了都会 400。这是新手最高频的报错。- 停止条件就是「模型不再调工具」。 不需要显式的
finish工具——模型输出纯文本时循环自然结束。这也是为什么 system prompt 里要写「信息够了就直接给答案」。 - 不要给 GPT-5 系列这类推理模型传
temperature,API 会直接报错;不写这个参数,让模型用默认值。 - 成本:mini/nano 级模型 2026 年中的价格大约是百万输入 token 几毛美元量级,跑通这页全部实验通常花不到一分钱人民币,放心跑。
为什么用 Chat Completions 而不是更新的 Responses API
OpenAI 官方文档现在主推 Responses API(工具 schema 拍平、内置 tool search 等新特性),但 Chat Completions 是所有 OpenAI 兼容端点的最大公约数——学会这套消息结构,你换任何厂商都不用改代码。生产上新项目可以评估 Responses API。
三、v1:工具错误回传与步数上限
v0 跑简单任务没问题,跑复杂任务会暴露两个致命伤:
- 任何异常都会让程序整个崩掉。 模型生成了非法 JSON 参数、传了不存在的参数名、调了不存在的工具——
json.loads或DISPATCH[name]直接抛异常,loop 终止,前面的工作全丢。 while True没有刹车。 模型可能陷入「调工具 → 结果不满意 → 换个参数再调」的死循环,烧 token 烧到天荒地老。
v1 的修法是加一层统一分发器,把所有异常变成字符串回传给模型,再加上步数上限:
python
MAX_STEPS = 12 # 一个任务 12 步内搞不定,大概率是跑偏了
def run_tool(name: str, raw_args: str) -> str:
"""工具执行的统一入口:任何异常都变成字符串回传给模型,而不是让程序崩掉"""
fn = DISPATCH.get(name)
if fn is None:
return f"error: 未知工具 {name},可用工具: {list(DISPATCH)}"
try:
args = json.loads(raw_args)
except json.JSONDecodeError:
return f"error: 参数不是合法 JSON: {raw_args[:200]}"
try:
return fn(**args)
except TypeError as e:
return f"error: 参数不匹配: {e}"
except Exception as e:
return f"error: 工具内部异常: {type(e).__name__}: {e}"
def run(task: str) -> str:
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": task},
]
for step in range(MAX_STEPS):
resp = client.chat.completions.create(model=MODEL, messages=messages, tools=TOOLS)
msg = resp.choices[0].message
messages.append(msg.model_dump(exclude_none=True))
if not msg.tool_calls:
return msg.content
for tc in msg.tool_calls:
result = run_tool(tc.function.name, tc.function.arguments) # 异常在这里被吸收
print(f" [step {step}] {tc.function.name} -> {result[:60]}...")
messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})
return f"(达到 {MAX_STEPS} 步上限,强制停止。最后一条消息:{messages[-1]})"改动只有两处,但行为变化非常大。故意给它一个坑:把 notes.txt 删掉再提问,你会看到类似这样的轨迹:
[step 0] read_file -> error: 文件不存在 notes.txt...
[step 1] run_shell -> notes.txt 不存在?让我用 ls 看看...
[step 2] read_file -> error: 路径越界 ../notes.txt...
[step 3] (给出最终回答)notes.txt 目前不存在……这就是「错误回传」的威力:模型读得懂错误信息,会自己换策略。error: 文件不存在 它就去 ls;error: 路径越界 它就改回相对路径。你什么都不用教,只要把错误写清楚。
错误消息是写给模型看的,不是写给人看的
error: 未知工具 xxx,可用工具: [read_file, run_shell] 这种格式比 KeyError: 'xxx' 有效得多——前者直接告诉模型下一步该干什么。设计工具返回值的第一个原则:让模型能从错误中恢复。这也是 Agent 设计原则 里「错误即信息」的落地。
步数上限则是另一道保险:它不是用户体验问题,是成本和安全问题。一个失控的 Agent 在一小时内烧掉几十美元是真实发生过的事故。MAX_STEPS 到顶后把最后状态带出来,方便你排查它卡在哪。
四、v2:加 todo 规划工具
任务一复杂(比如「把 workspace 里所有 .py 文件的 TODO 注释汇总成一份报告」),v1 会出现典型症状:干到一半忘了目标、重复读同一个文件、或者读了一个文件就匆忙交卷。原因很朴素——计划只存在于模型的「短期记忆」里,而每轮对话它都要从茫茫上下文中重新推断自己该干嘛。
解法是给模型一个外置的「记事本」:todo 工具。这是 Claude Code、Cursor 等 coding agent 的标配,实现却简单得不像话——它甚至不「做」任何事,只是把任务清单写进上下文:
python
TODO_STATE: list[dict] = []
def todo_write(todos: list) -> str:
"""全量覆盖任务清单。todos: [{"content": "...", "status": "pending|in_progress|completed"}]"""
TODO_STATE.clear()
TODO_STATE.extend(todos)
mark = {"pending": " ", "in_progress": "~", "completed": "x"}
lines = [f"[{mark.get(t.get('status'), ' ')}] {t.get('content', '')}" for t in TODO_STATE]
return "当前任务清单:\n" + "\n".join(lines)
TODO_TOOL = {
"type": "function",
"function": {
"name": "todo_write",
"description": "更新任务清单。开始工作前列出步骤,每完成一步把对应项标记为 completed",
"parameters": {
"type": "object",
"properties": {
"todos": {
"type": "array",
"items": {
"type": "object",
"properties": {
"content": {"type": "string"},
"status": {"type": "string", "enum": ["pending", "in_progress", "completed"]},
},
"required": ["content", "status"],
},
},
},
"required": ["todos"],
},
},
}
TOOLS.append(TODO_TOOL)
DISPATCH["todo_write"] = todo_write同时在 system prompt 里加一句:
- 对于需要 3 步以上的任务,先用 todo_write 列出计划,每完成一步就更新清单状态。预期轨迹变成:
[step 0] todo_write -> 当前任务清单: [~] 找出所有 .py 文件 [ ] 提取 TODO 注释 [ ] 汇总报告
[step 1] run_shell(find . -name "*.py") -> ./a.py ./b.py...
[step 2] todo_write -> [x] 找出所有 .py 文件 [~] 提取 TODO 注释 [ ] 汇总报告...
[step 3] run_shell(grep -rn TODO .) -> ./a.py:3:# TODO: 处理空文件...
[step 4] todo_write -> [x] [x] [x] ...
[step 5] (最终回答)共找到 2 个文件、3 条 TODO……为什么一个「什么都不干」的工具能明显提升表现?两个机制:
- 目标外化。计划从「模型每轮重新脑补」变成「白纸黑字躺在上下文里」,模型每轮都能读到「我现在进行到哪一步」。这是 Planning 一节讲的 plan-and-execute 的最小实现。
- 注意力锚点。Transformer 对上下文首尾的内容关注更强,一份反复刷新的清单持续把「当前步骤」推到最近的位置,对抗长上下文里的「迷失在中间」(lost in the middle)效应。
判断标准:什么时候值得加 todo
经验法则:任务预期超过 3-5 步、或中间产物需要跨多步引用时,加;一问一答式任务别加——它只是徒增 token 开销和一步多余的工具调用。工具不是越多越好,模型在工具之间做选择也是要消耗「注意力预算」的。
五、v3:上下文压缩
跑长任务你迟早撞上这堵墙:工具结果一条条堆进 messages,几十轮后 token 数逼近模型的 context window 上限,要么报错要么贵得离谱。v3 加最后一个机制:超过阈值时,自动把早期对话总结成摘要。
python
MAX_CONTEXT_TOKENS = 6000 # 演示用小阈值;生产里按模型实际上限的 70-80% 设
KEEP_RECENT = 6 # 压缩后保留最近 N 条消息原文
def approx_tokens(messages: list) -> int:
"""粗估 token 数:约 4 字符 ≈ 1 token。要精确就用 tiktoken,教学上这个够了"""
return sum(len(json.dumps(m, ensure_ascii=False)) for m in messages) // 4
def compact(messages: list) -> list:
"""把早期对话压缩成摘要:保留 system + 摘要 + 最近 N 条原文"""
boundary = len(messages) - KEEP_RECENT
# 不能在 assistant(tool_calls) 和它的 tool 回复之间切,否则协议报错。
# 向前回退到一条非 tool 消息为止。
while boundary > 1 and messages[boundary].get("role") == "tool":
boundary -= 1
old, recent = messages[1:boundary], messages[boundary:]
resp = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": "你是上下文压缩器。把对话历史压缩成给下一个 Agent 看的简报,"
"必须保留:用户目标、已完成的步骤、关键文件路径与结论、待办事项。"},
{"role": "user", "content": "请压缩以下对话:\n" + json.dumps(old, ensure_ascii=False)[:8000]},
],
)
summary = resp.choices[0].message.content
print(f" [compact] {approx_tokens(messages)} tokens -> 压缩为摘要")
return [messages[0], {"role": "user", "content": f"【前情摘要】\n{summary}"}] + recent在 loop 的每次迭代开头加一行接线:
python
for step in range(MAX_STEPS):
if approx_tokens(messages) > MAX_CONTEXT_TOKENS:
messages = compact(messages)
resp = client.chat.completions.create(model=MODEL, messages=messages, tools=TOOLS)
# ……其余与 v1 相同注意 compact 里那个 while 回退——这是真刀真枪的坑:如果你把一条 tool 消息留在 recent 开头、而它配对的 assistant tool_calls 消息被压掉了,API 会报 tool_call_id 找不到配对。压缩点必须落在消息配对的边界上。
两个诚实的告诫:
- 压缩是有损的。 Claude Code 的用户就实际遇到过压缩后「忘掉」项目指令、甚至把摘要里的「下一步建议」当成用户指令自动执行的问题(见参考资料里的 issue)。所以我们的实现里 system prompt 永远不被压缩;真正关键的事实(结论、路径、决策)应该写进文件,而不是只留在对话历史里——这正是 v2 的 todo 和后面练习里 memory 文件存在的意义。
- 阈值别贴脸上限。 留 20-30% 余量给摘要生成和后续输出。贴满再压会陷入「太长所以要压缩,太长所以压不动」的死锁(Claude Code 也踩过)。
这套机制的工业版远比这里复杂——分层摘要、选择性保留、按重要性淘汰——属于 上下文工程 的范畴,但核心思想你已经亲手实现了。
六、练习清单:接着该加什么
这个 ~150 行的骨架离一个「产品级」Agent 还差得远,但每一步扩展都是一次真实的学习。按推荐顺序:
- 危险操作确认:给
run_shell加一个需要人工输入y/n才执行的写命令名单。想明白「哪些操作需要人审」是 Human-in-the-Loop 的核心问题。 - 记忆文件:让 Agent 把重要结论写入
memory.md,每轮把文件内容注入 system prompt。对比「压缩对话」和「外置记忆」两条路线的优劣,参见 记忆系统。 - 子代理:加一个
spawn_agent(task)工具——内部就是再调一次run(),只把最终结果带回主对话。体会「子代理的上下文隔离为什么能保护主对话不被污染」,详见 多 Agent 架构。 - 重试与限流:给 API 调用加指数退避(429/5xx 时 sleep 后重试)。不加这个,你的 Agent 在真实网络环境里活不过一天。
- 流式输出:把
stream=True接上,逐字打印。用户体验的本质改变,实现只要十几行。 - 写 eval:攒 10 个任务 + 预期结果,每次改 prompt 后跑一遍看通过率。没有 eval 的 prompt 调优都是玄学,方法见 评测实践。
- 接 MCP:把一个
read_file换成调用真实 MCP server。你会发现 MCP 客户端干的活和你手写的 dispatch 一模一样,只是协议标准化了,见 工具与 MCP。
做完 1-3,你就已经理解了 Claude Code 这类产品 80% 的骨架——剩下的 20% 是打磨,见 Claude Code 案例。把项目整理成 README 发出去,就是一个合格的 作品集项目。
七、常见 bug 与调试技巧
按踩坑频率排序:
| 症状 | 根因 | 解法 |
|---|---|---|
400:tool_call_id did not have response | assistant 的 tool_calls 没进历史,或 tool 消息少回/顺序错 | 检查 msg.model_dump() 那行;压缩时注意配对边界(见 v3) |
| 模型反复调同一个工具 | 工具报错但错误消息没说清怎么办,或结果里没它要的信息 | 改进错误文案(v1);在 system prompt 里给「失败 N 次就换思路」的指令 |
| 模型不调工具直接瞎编 | system prompt 没要求先查证,或工具 description 写得像摆设 | 明确「必须先读文件再回答」;把 description 当 prompt 写 |
temperature 参数报错 | GPT-5/o 系列推理模型不支持该参数 | 删掉,用默认值 |
| 上下文爆了 | 工具输出没截断,一次 cat 进几万字 | 所有工具输出截断(本页统一 4000 字符);上 v3 压缩 |
Windows 下 ls/cat 报「命令不存在」 | 白名单命令是 Unix 的 | 换成跨平台实现(纯 Python 的 list_dir/read_file),或允许 dir/type |
| JSON 解析参数失败 | 模型(尤其弱模型)生成了截断/非法 JSON | v1 的 run_tool 已兜底;频繁出现就换更强的模型或开 strict 模式 |
调试手段就三条,但够用:
- 打印每一步的 trace(本页所有
[tool]输出)。Agent 调试的第一要务是让隐形的循环可见——这也是 可观测性 的原始形态。 - 把完整
messages落盘成 JSONL,一步一行。出问题时直接翻最后几条,比猜强一百倍,还能攒下来当 eval 数据集。 - mock 掉 LLM 测工具执行器:写一个假
client返回固定的tool_calls,单测你的 dispatch、安全边界和错误回传。工具层是纯函数,不需要为测它花 API 钱。
到这里,你手里已经有一个能跑、能自愈、会规划、不爆 context 的 Agent 雏形。更重要的是:以后再看任何框架的源码,你看到的将不再是魔法,而是「哦,这段我写过」。
参考资料
- OpenAI Function Calling 官方指南 —— 工具 schema、tool_calls 协议、strict 模式与 Responses API 的权威说明。
- How to Build an AI Agent from Scratch in Python (2026) —— 与本页同思路的英文教程,约 60 行核心循环。
- sergenes/mini_agent —— 只用 OpenAI SDK 和一个 while 循环的最小 Agent 开源实现,配套 Medium 系列文章。
- codereindeer-dev/minimal-agent —— 单文件 Python Agent,每个 commit 加一个概念(工具、记忆、子代理),适合做练习清单的对照。
- TheSeydiCharyyev/build-your-own-agent —— 「build-your-own-x」风格的 Agent 组件教程精选索引。
- Claude Code issue:压缩后丢失 CLAUDE.md 指令 —— 上下文压缩有损性的真实案例,印证 v3 的告诫。
- 阿里云:通过 OpenAI 兼容模式调用通义千问 —— 本页代码切换到国产模型端点的参考。