Agent 开发学习站
Agent 开发›核心›核心

手写最小 Agent:100 行的完整循环

核心核心

一句话定义

用不到 100 行 Python、不依赖任何框架,从零实现一个具备工具调用、循环驱动、错误纠正与终止保护的完整 Agent——它是全知识库的枢纽:kp-006 的理论在这里落地,kp-021 的框架在这里被祛魅。

为什么重要

"手写过再上框架"是本领域最重要的学习顺序(见学习路径)。框架(LangGraph 等)封装的正是这段代码——没写过之前,框架的图、节点、checkpoint 对你是黑魔法;写过之后,你知道每个抽象对应循环里的哪几行,出问题能沉到底层修。这段代码也适合当脚手架:后续每学一个机制(kp-011 MCP、kp-015 记忆、kp-019 子 Agent)就往上加一块。

前置知识

kp-006(消息流不变式)、kp-009(协议)。需要一个 API key(任意支持函数调用的模型,示例用 Anthropic 风格 SDK,各家的消息结构高度趋同)。

核心概念

实现清单(对照 kp-006 的骨架逐项落地):

  1. 工具注册表:名称 → 真实函数的映射 + schema 列表(kp-010)
  2. 循环:调用模型 → 判断 stop_reason → 分派执行 → 回填 tool_result → 继续
  3. 终止保护:最大轮数 + 异常兜底
  4. 可观测性钩子:每轮打印轨迹(kp-025 的雏形)

原理与机制(完整代码)

python"""minimal_agent.py — 无框架最小 Agent(Anthropic 风格 SDK,~90 行)"""
import json, anthropic

client = anthropic.Anthropic()
MODEL = "claude-sonnet-4-5"

# ── ① 工具:schema + 实现(kp-010 守则)─────────────────────
def read_file(path: str) -> str:
    try:
        return open(path, encoding="utf-8").read()[:8000]   # 截断防爆窗
    except FileNotFoundError:
        return f"错误:文件 {path} 不存在。请先用 list_files 确认路径。"

def list_files(directory: str = ".") -> str:
    import os
    return "\n".join(sorted(os.listdir(directory))[:200])

def write_file(path: str, content: str) -> str:
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return f"已写入 {path}({len(content)} 字符)"

TOOLS_SPEC = [{
    "name": "read_file", "description": "读取文本文件内容",
    "input_schema": {"type": "object",
        "properties": {"path": {"type": "string", "description": "相对路径"}},
        "required": ["path"], "additionalProperties": False}},
}, {
    "name": "list_files", "description": "列出目录下的文件名",
    "input_schema": {"type": "object",
        "properties": {"directory": {"type": "string", "default": "."}},
        "additionalProperties": False}},
}, {
    "name": "write_file", "description": "写入/覆盖文本文件",
    "input_schema": {"type": "object",
        "properties": {"path": {"type": "string"},
                       "content": {"type": "string"}},
        "required": ["path", "content"], "additionalProperties": False}},
}]
TOOLBOX = {"read_file": read_file, "list_files": list_files, "write_file": write_file}

# ── ② Agent 循环(kp-006)───────────────────────────────────
def run_agent(task: str, max_steps: int = 20) -> str:
    messages = [{
        "role": "user",
        "content": (f"{task}\n\n"
                    "工作守则:先 list_files 了解环境再行动;"
                    "完成后用一句话汇报你做了什么。"),
    }]
    for step in range(max_steps):                       # ③ 终止保护
        resp = client.messages.create(
            model=MODEL, max_tokens=4096, tools=TOOLS_SPEC, messages=messages)
        print(f"[step {step}] stop={resp.stop_reason}") # ④ 轨迹

        if resp.stop_reason != "tool_use":              # 模型宣布完成
            return next((b.text for b in resp.content
                         if b.type == "text"), "(无文本输出)")

        messages.append({"role": "assistant", "content": resp.content})
        results = []
        for block in resp.content:                      # 逐个工具调用
            if block.type != "tool_use":
                continue
            fn = TOOLBOX.get(block.name)
            if fn is None:
                out = f"错误:未知工具 {block.name},可用:{list(TOOLBOX)}"
            else:
                try:
                    out = fn(**block.input)
                except Exception as e:                  # 可行动错误(kp-010)
                    out = f"错误:{type(e).__name__}: {e}。请检查参数后重试。"
            print(f"  ↳ {block.name}({block.input}) → {out[:80]}")
            results.append({"type": "tool_result",
                            "tool_use_id": block.id, "content": out})
        messages.append({"role": "user", "content": results})  # ★ 回填
    return "已达最大步数上限,强制终止。"

if __name__ == "__main__":
    print(run_agent("看看当前目录有什么文件,挑一个 txt 文件统计行数,"
                    "把结果写入 summary.txt"))

逐行对照知识点

代码位置对应知识点
TOOLS_SPEC 的 descriptionkp-009 协议 + kp-010 描述写法
max_steps 循环上限kp-006 终止保护(防烧钱死循环)
stop_reason != "tool_use"kp-006 终止条件一(模型宣布完成)
messages.append(...results)kp-006 不变式:结果必须回填
截断 [:8000]kp-003 上下文预算管理
异常转"可行动错误"kp-010 / kp-014 自我纠正的燃料
print 轨迹kp-025 可观测性雏形

直观类比

这 90 行就是一台自动挡汽车的发动机裸机:能开,但没有车身(UI)、没有仪表盘(追踪)、没有安全气囊(权限)。框架卖给你的不是发动机,是整车——但不懂发动机的司机,只会换整车不会修发动机。

实例与案例

建议的三个扩展练习(每个 30-60 分钟):

  1. 加搜索工具:接一个 web_search(可用免费 API),观察模型如何在"本地文件找不到"时转向搜索
  2. 加 todo 工具(kp-013):update_plan(items) 让模型自己维护计划文件,观察长任务迷路率变化
  3. 加审批门(kp-023):write_file 执行前 input("确认写入? [y/N]"),体验 human-in-the-loop 的实现之轻

常见误区

  • 误区一:写完就认为"框架没用"。这 90 行离生产差的是:持久化/断点续跑、并发、权限模型、重试策略、成本核算、观测面板——那才是框架/平台的价值(kp-021/027)。
  • 误区二:不打印轨迹直接跑。没有 trace 的 Agent 调试全靠猜(kp-025)。
  • 误区三:把 max_steps 设很大"让它自己跑"。20 步内完不成的任务通常该重新分解(kp-013)而不是给更多步数。
  • 误区四:同步阻塞执行所有工具。多 tool_use 并行执行(asyncio)在真实场景是 2-3 倍延迟差(kp-027)。

自测题

  1. 删掉 messages.append(...results) 这行会发生什么?为什么?
  2. stop_reason 的哪两个值分别对应循环的哪两个出口?
  3. 为什么工具输出要截断而不是全量回填?

(参考答案:1. 模型看不到结果,会重复请求同一工具直至步数耗尽——不变式被破坏;2. end_turn→模型宣布完成;tool_use→继续循环;3. 上下文预算与注意力稀释(kp-003)。)

与其他知识点的关系

  • 理论骨架 → kp-006;工具层 → kp-009/010;加沙箱 → kp-018
  • 加 MCP → kp-011;加子 Agent → kp-019;进化为框架对比 → kp-021

延伸阅读

  • Anthropic, Tool Use 官方 quickstart(本实现的参考基准)
  • 各家 SDK 的消息结构文档(OpenAI Responses / DeepSeek / GLM——结构高度趋同,迁移成本主要是字段名)