Skip to content

从零构建一个最小 Agent

本页速览 不用任何框架,只用 LLM API 手写一个能跑的最小 Agent:50 行 Agent Loop 起步,逐版加上错误回传、步数上限、todo 规划和上下文压缩,每步附完整可运行代码与预期运行轨迹。

从零构建一个最小 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 的全部本质。一轮迭代里只有四件事:

  1. 把「对话历史 + 工具 schema」发给模型;
  2. 模型要么返回最终文本,要么返回一个或多个 tool_calls(函数名 + JSON 参数);
  3. 你的代码执行对应函数,把结果以 role: "tool" 消息塞回历史;
  4. 回到第 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 跑简单任务没问题,跑复杂任务会暴露两个致命伤:

  1. 任何异常都会让程序整个崩掉。 模型生成了非法 JSON 参数、传了不存在的参数名、调了不存在的工具——json.loads 或 DISPATCH[name] 直接抛异常,loop 终止,前面的工作全丢。
  2. 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 还差得远,但每一步扩展都是一次真实的学习。按推荐顺序:

  1. 危险操作确认:给 run_shell 加一个需要人工输入 y/n 才执行的写命令名单。想明白「哪些操作需要人审」是 Human-in-the-Loop 的核心问题。
  2. 记忆文件:让 Agent 把重要结论写入 memory.md,每轮把文件内容注入 system prompt。对比「压缩对话」和「外置记忆」两条路线的优劣,参见 记忆系统。
  3. 子代理:加一个 spawn_agent(task) 工具——内部就是再调一次 run(),只把最终结果带回主对话。体会「子代理的上下文隔离为什么能保护主对话不被污染」,详见 多 Agent 架构。
  4. 重试与限流:给 API 调用加指数退避(429/5xx 时 sleep 后重试)。不加这个,你的 Agent 在真实网络环境里活不过一天。
  5. 流式输出:把 stream=True 接上,逐字打印。用户体验的本质改变,实现只要十几行。
  6. 写 eval:攒 10 个任务 + 预期结果,每次改 prompt 后跑一遍看通过率。没有 eval 的 prompt 调优都是玄学,方法见 评测实践。
  7. 接 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 responseassistant 的 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 解析参数失败模型(尤其弱模型)生成了截断/非法 JSONv1 的 run_tool 已兜底;频繁出现就换更强的模型或开 strict 模式

调试手段就三条,但够用:

  1. 打印每一步的 trace(本页所有 [tool] 输出)。Agent 调试的第一要务是让隐形的循环可见——这也是 可观测性 的原始形态。
  2. 把完整 messages 落盘成 JSONL,一步一行。出问题时直接翻最后几条,比猜强一百倍,还能攒下来当 eval 数据集。
  3. mock 掉 LLM 测工具执行器:写一个假 client 返回固定的 tool_calls,单测你的 dispatch、安全边界和错误回传。工具层是纯函数,不需要为测它花 API 钱。

到这里,你手里已经有一个能跑、能自愈、会规划、不爆 context 的 Agent 雏形。更重要的是:以后再看任何框架的源码,你看到的将不再是魔法,而是「哦,这段我写过」。

参考资料 ​