Agent 究竟是什么

在写一行代码之前，必须先把"什么是 Agent"想清楚。市面上对 Agent 的描述充满了拟人化的词——"自主"、"思考"、"规划"——这些词都没错，但它们容易让你以为里面藏着某种魔法。其实没有。Agent 的本质是一段循环，加上一些工具，仅此而已。

1.1LLM 是一个无状态的函数

这是整本手册的基石，请先记住它：

把这件事想透，许多看似神秘的现象就一目了然了。"模型为什么会忘记十轮前的事？"因为你没把那十轮喂回去。"为什么模型会自相矛盾？"因为它根本不知道之前的自己是谁。把 LLM 看作 f(text) -> text，剩下的一切都是你在外面拼装的工程问题。

1.2从 Chat 到 Agent，本质上多了什么

把 Chat 和 Agent 摆在一起对照看，你会发现差距其实只在一处。

关键差别就一句话：Agent 在 LLM 之外，套了一个"自动继续"的循环，并允许 LLM 的输出触发外部动作。

1.3所谓"自主性"，是 runtime 给的

这是初学者最容易误解的地方。我们常说"Agent 自己决定下一步做什么"——但 LLM 哪有能力"决定"？它只能输出文本。真正的"决定"发生在你写的 runtime 里：

1. LLM 输出一段文本，里面可能是答案，也可能是某种结构化的请求（"我想调用 read_file 这个工具，参数是 ..."）。
2. 你的 runtime 解析这段文本。如果是普通答案，结束；如果是工具请求，runtime 替它执行，再把结果塞回上下文，让 LLM 继续推理。
3. 这个循环重复，直到 LLM 不再要求调用工具，或者达到某个停止条件。

1.4这本手册的整体地图

有了上面的认知地基，整本手册的路径就清楚了。我们会一层一层往上盖：

• Ch.2 核心循环：写出最简单的 while，跑通"思考-行动-观察"。
• Ch.3 工具调用：给 LLM 装上"手脚"，让它能读写文件、查 API。
• Ch.4 记忆管理：当对话变长，上下文窗口不够用时怎么办。
• Ch.5 状态管理：复杂任务需要显式状态，不能依赖 LLM 的"记性"。
• Ch.6 文件管理：文件是天然的状态载体，也是 Agent 工程的灵魂。
• Ch.7 完整实现：把前面所有东西拼成一个真正能跑的 Agent。
• Ch.8 进阶：规划、反思、多智能体——继续往专家路线走。

核心循环 The Agent Loop

整个 Agent 工程里，有一段代码占据了绝对的核心地位——它就是那个最朴素的 while 循环。这一章我们把这个循环从 5 行写到 30 行，沿途把 ReAct 范式、停止条件、错误恢复这些"细节"逐个谈清楚。看完之后，你会发现"做一个 Agent" 远比想象中简单。

2.1世界上最小的 Agent

不夸张地讲，一个能用的 Agent 大概长这样：

# 最小可工作的 Agent，约 15 行
messages = [
    {"role": "system", "content": SYSTEM_PROMPT},
    {"role": "user",   "content": user_goal},
]

while True:
    response = llm.complete(messages, tools=TOOLS)
    messages.append(response.message)

    if response.tool_calls:
        for call in response.tool_calls:
            result = execute_tool(call.name, call.args)
            messages.append({"role": "tool", "content": result})
    else:
        print(response.content)
        break

就是这个东西。剩下的所有复杂度——记忆、状态、文件、规划——都是在这个骨架上长出来的修饰。 把它默背下来，你已经懂了 60% 的 Agent。

2.2ReAct：让 LLM 先想再做

2022 年 Yao 等人提出的 ReAct (Reasoning + Acting) 是早期 Agent 设计中最重要的一篇论文。它的洞察简单到令人发指：

区别在哪里？看下面这个对比：

现代 API（Anthropic、OpenAI 都支持）内置了 native function calling，模型可以在结构化输出动作之外，再附带自由文本。ReAct 的精神已经被吸收进 API 设计里。你只需要在 system prompt 里鼓励它先解释再行动，就拿到了 ReAct 的好处。

2.3停止条件，是个会反复咬你的问题

看上面那个 while True，你应该立刻警觉：什么时候 break？如果 LLM 一直要求调用工具，你就一直被它牵着走，分分钟把 API 账单烧穿。所以停止条件是 Agent runtime 最先要写好的边界。

实战中至少要设这五道闸：

1. 正常结束：LLM 不再返回 tool_calls，意味着它认为任务完成了。这是 happy path。
2. 最大迭代数：硬限制循环跑多少轮（比如 25 轮）。超过就强制终止。
3. 最大 token 预算：累计消耗超过 X token 就停。直接和钱挂钩，最实用。
4. 最大耗时：墙钟时间超过 Y 分钟就停。防止某个慢工具拖死整个 agent。
5. 致命错误：连续 N 次工具调用失败、API 抛 500、权限拒绝……这些立即终止并上报。

2.4把闸门加进 Agent Loop

更新一下我们的最小 Agent：

# 加了停止条件的 Agent Loop
MAX_STEPS = 25
MAX_TOKENS = 100_000

def run_agent(user_goal):
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user",   "content": user_goal},
    ]
    total_tokens = 0

    for step in range(MAX_STEPS):
        response = llm.complete(messages, tools=TOOLS)
        total_tokens += response.usage.total_tokens
        messages.append(response.message)

        if total_tokens > MAX_TOKENS:
            return "[Stopped] token budget exceeded"

        if not response.tool_calls:
            return response.content   # 正常结束

        for call in response.tool_calls:
            try:
                result = execute_tool(call.name, call.args)
            except Exception as e:
                result = f"[Tool error] {e}"   # 错误也喂回去，让 LLM 自我纠正
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": result,
            })

    return "[Stopped] max steps reached"

注意两个细节：

1. 工具报错时不要直接 raise，而是把错误信息也作为 observation 喂回 LLM。这样它有机会自我纠正——"哦，文件不存在，那我换个路径试试"。
2. 循环用 for 不用 while。 for step in range(MAX_STEPS) 比 while True 安全得多——天然就有上限。

2.5System Prompt 该写些什么

System prompt 是你和 Agent "签的合同"。它该包含：

• 角色与目标：你是谁，要干什么。一两句话讲清。
• 可用工具的使用建议：什么时候用什么工具（API 已经把工具 schema 给模型了，这里讲策略）。
• 输出规范：什么时候停下、停下时输出什么。
• 边界与限制：不要做什么。

给一个最小可用的模板：

你是一个善于完成具体任务的 AI 助手。你能调用一组工具来读写文件、
搜索信息、执行计算。

工作方式：
1. 收到目标后，先在 thought 中分析需要哪些步骤；
2. 选择并调用合适的工具；
3. 仔细阅读工具返回的结果，再决定下一步；
4. 任务完成时，给出一份清晰的总结，不要再调用工具。

要求：
- 每次只调用必要的工具，不要重复读同一个文件；
- 工具报错时分析错误原因并尝试不同方案，不要盲目重试；
- 任务无法完成时，明确说明原因，而不是凭空编造结果。

工具调用 Tool Use

如果 Agent 的"大脑"是 LLM，那"手脚"就是工具。一个没有工具的 Agent，就只能在自己的语言世界里打转——它能写代码，但不能运行；它能读你说的话，但读不了一个真实的文件。工具，是 Agent 和世界的接口。这一章把工具调用从原理到工程实践讲透，包括一个反直觉的事实：所谓"LLM 调用工具"，从头到尾都是你在替它调用。

3.1工具调用的真相：LLM 没"调用"任何东西

请回忆 Ch.1：LLM 只能输出文本。所以当你看到模型"调用了 read_file"，真实发生的事情是这样：

1. 你在 API 请求里附带了一份 tools schema——告诉模型有哪些工具、每个工具叫什么名字、接什么参数。
2. 模型生成 token 的时候，被训练成会输出一段特殊格式的内容，本质上是个 JSON：
   {"name": "read_file", "arguments": {"path": "report.md"}}
3. API 把这段 JSON 单独拎出来，作为 tool_calls 字段返回给你。
4. 你的代码看到这个字段，调用真实的 read_file() 函数，拿到结果。
5. 你把结果（又是一段文本）作为新的 message 加进对话历史，再调用一次 API。
6. 模型这次看到了文件内容，可以继续推理。

3.2工具的 Schema：模型只能看到 description

当你定义一个工具，需要给 API 提供这样一份描述：

tool_schema = {
    "name": "read_file",
    "description": (
        "读取工作目录中的一个文本文件并返回其完整内容。"
        "适用于需要查看文件内容、提取信息或分析代码的场景。"
        "如果文件不存在或非文本格式，会返回错误信息。"
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "path": {
                "type": "string",
                "description": "相对于工作目录的文件路径，如 'src/main.py'",
            }
        },
        "required": ["path"],
    },
}

写 description 的几个心法：

• 说能力，也说适用场景。"读取文件" 是能力；"适用于查看代码、提取信息" 是场景。两者都要给。
• 说失败行为。"文件不存在会返回错误信息" 这一句能让模型懂得在错误时该怎么办。
• 说边界。"只能读 1MB 以下的文本文件"——明说限制，模型不会乱来。
• 参数 description 同样关键。 path 是绝对路径还是相对路径？给个例子最稳。

3.3工具粒度：每个工具该做多大的事

这是个工程审美问题，但有几条经验法则：

合适的粒度通常是：每个工具完成一个"有意义的最小动作"。读一整个文件、发一个 HTTP 请求、运行一段 Python——这些都是好的粒度。一次 IO + 简单加工，差不多就是一个工具该承担的事。

3.4错误反馈：把报错当 observation 喂回去

这是 Agent 工程里"教会模型自我纠正"的核心技巧。来看一个对照：

更进一步：在错误信息里附带修复提示，效果更好。例如：

except FileNotFoundError:
    available = os.listdir(work_dir)
    return f"[Error] 文件 {path} 不存在。当前目录下有: {available[:10]}"

这样 LLM 不仅知道错了，还顺手得到了恢复所需的信息。错误信息越"有用"，Agent 的鲁棒性越强。

3.5工具注册表：把工具组织起来

工具一多，散在代码各处就难管了。常用模式是搞一个 tool registry：

# 一个简易但够用的工具注册表
class ToolRegistry:
    def __init__(self):
        self._tools = {}

    def register(self, name, fn, description, input_schema):
        self._tools[name] = {
            "fn": fn,
            "schema": {
                "name": name,
                "description": description,
                "input_schema": input_schema,
            },
        }

    def schemas(self):
        return [t["schema"] for t in self._tools.values()]

    def dispatch(self, name, args):
        if name not in self._tools:
            return f"[Error] unknown tool: {name}"
        try:
            return self._tools[name]["fn"](**args)
        except Exception as e:
            return f"[Error] {type(e).__name__}: {e}"


# 注册一个工具
registry = ToolRegistry()
registry.register(
    name="read_file",
    fn=lambda path: open(path).read(),
    description="读取工作目录中的文本文件并返回内容...",
    input_schema={"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]},
)

这样在主循环里就只剩一句 registry.dispatch(call.name, call.args)，干净利落。

3.6常备工具清单

一个通用 Agent 最少应该具备这些工具：

• 文件操作四件套：read_file、write_file、list_files、delete_file。详见 Ch.6。
• 代码执行：run_python 或 run_bash。让 Agent 自己写代码自己跑，是放大能力的关键一步。
• 搜索：web_search（如果你有搜索 API）或 grep_files（搜本地）。
• HTTP 请求：http_get 之类。访问任何 REST API 都靠它。
• 用户交互：ask_user(question)。当 Agent 不确定时，让它主动问，比瞎猜强得多。

记忆管理 Memory

"Agent 怎么有记忆？" 这是新手最常问的问题，也是最容易被市面上各种"记忆框架"绕晕的话题。但只要记住 Ch.1 的那句话——LLM 是无状态的——所谓"记忆管理"就只是一件事：你怎么决定每一轮把哪些文本塞进 context window。这一章会把这件事的所有花样讲清楚。

4.1记忆的三种类型

借用认知科学的分类，Agent 的记忆可以分成三层。这个分类不是为了显得高大上，而是因为不同类型的信息要用不同方式管理。

Working Memory · 工作记忆

就是你 messages 数组里那些东西——system prompt、用户输入、LLM 回复、工具结果。一切都在 context window 里，每次 API 调用都被完整地塞进去。它非常"快"（模型一眼就看到），但极其昂贵——窗口就那么大，token 就那么多钱。

Episodic Memory · 任务记忆

本次任务跑了 30 轮，前 20 轮的细节已经不重要，但20 轮里的关键产出（生成的代码、调研结论、决策记录）必须保留。这些东西从 working memory "毕业"出来，落到磁盘上（文件、scratchpad、数据库），需要时再读回来。

Semantic Memory · 长期记忆

跨任务、跨会话的东西。用户的名字、偏好的语言、上次 debug 的项目结构……这些不属于任何一次具体任务，但每次都可能用到。通常存在外部存储（关系数据库、向量数据库），通过特定工具或注入方式让 Agent 用到。

4.2Working Memory 撑不下了怎么办

典型场景：Agent 跑到第 15 轮，messages 数组已经膨胀到 6 万 token，下一次调用就要爆窗口了。这时有四种经典策略：

策略 A · 滑动窗口（Sliding Window）

最朴素：只保留最近 N 条消息。早期的扔掉。简单粗暴，但会丢失早期上下文。适合不需要早期信息的场景（如客服对话）。

def trim_to_window(messages, keep=20):
    # 永远保留 system prompt 和最初的用户目标
    head = messages[:2]
    tail = messages[-keep:]
    return head + tail

策略 B · 摘要压缩（Summarization）

当消息超过阈值，把前面的若干条让 LLM 自己摘要成一段，替换原文。这样信息密度大幅提升，token 占用骤降。

def compress_old_messages(messages, llm, threshold=30):
    if len(messages) <= threshold:
        return messages

    old = messages[2:-10]   # 留头留尾，压缩中间
    summary = llm.complete([
        {"role": "system", "content": "把以下对话压缩成一段事实摘要，保留所有关键决策和数据。"},
        {"role": "user",   "content": str(old)},
    ]).content

    return messages[:2] + [
        {"role": "system", "content": f"[早期对话摘要] {summary}"}
    ] + messages[-10:]

注意：摘要本身要花一次 LLM 调用，所以触发频率别太高。一般每 20-50 轮触发一次。

策略 C · 向量检索（RAG-style）

把所有历史消息（或文档片段）嵌入向量库，每轮根据"当前在做什么"动态检索最相关的几条塞进 context。适合知识量大但单次只用一小部分的场景。

但要注意：你的 API key 只支持 completion，不一定支持 embedding。如果没有 embedding API，可以用关键词检索（BM25、ripgrep）代替，效果也不错。

策略 D · 把信息搬到文件里

这是最 underrated 但最实用的策略。**当某个工具返回 1MB 的文本，不要把它整段塞进 messages**——写到一个文件里，只在 messages 里留下"我把结果写到了 result_001.json"。下次 LLM 需要时，让它自己用 read_file 读。

4.3实战推荐：混合策略

真实项目里没人只用单一策略。Carrey 你做平台时大概率会用类似下面这种组合：

1. 固定头部：system prompt + 用户原始目标，永远不变。
2. 动态压缩：早期对话每 N 轮被摘要替换。
3. 近期原文：最近 10-15 轮保留完整原文（保证细节）。
4. 文件 offload：工具返回的大块内容写文件，messages 里只放文件名 + 一句摘要。
5. 跨任务知识：用户偏好等放在 DB / memory file 里，必要时通过工具或注入读取。

# 一个把所有策略组合起来的 Memory Manager
class MemoryManager:
    def __init__(self, llm, max_messages=30, max_tokens=60_000):
        self.llm = llm
        self.max_messages = max_messages
        self.max_tokens = max_tokens
        self.messages = []

    def add(self, msg):
        self.messages.append(msg)
        self._maybe_compress()

    def _maybe_compress(self):
        if len(self.messages) <= self.max_messages:
            return
        # 留前 2 条（system + goal）+ 后 10 条原文，中间摘要
        head, middle, tail = self.messages[:2], self.messages[2:-10], self.messages[-10:]
        summary = self._summarize(middle)
        self.messages = head + [{"role": "system", "content": f"[摘要] {summary}"}] + tail

    def _summarize(self, msgs):
        prompt = [
            {"role": "system", "content": "把以下消息压缩成一段事实记录，保留所有决策、关键数据、错误。"},
            {"role": "user",   "content": json.dumps(msgs, ensure_ascii=False)},
        ]
        return self.llm.complete(prompt).content

4.4记忆管理的几条铁律

1. 永远保留 system prompt 和初始用户目标。任何压缩都不能动这两条，否则 Agent 会忘了自己在干什么。
2. 压缩要保留"决策"和"错误"。可以丢失的是冗余对话，不可丢失的是"我们已经决定走 A 方案"、"上次尝试 X 失败了"。
3. 压缩本身有成本。摘要 = 一次 LLM 调用 + 一次完整上下文输入，比想象中贵。别压得太频繁。
4. 大对象走文件，不走 context。一个工具返回 5000 行 JSON？写到文件里，让模型按需读。
5. 压缩有损。重要事实可能被摘要丢掉。关键信息（如"用户的 API key 是 sk-xxxx"）不要依赖摘要保留，单独存。

状态管理 State

"记忆"是给 Agent 看的，"状态"是给你看的。它们容易被混淆，但本质完全不同：记忆是叙事性的（关于发生过什么），状态是结构性的（关于现在是什么样）。一个长任务的 Agent 没有显式状态，就像一个不写日志的程序——出问题时你只能祈祷。

5.1为什么不能依赖 LLM 的"内部记性"

初学者常犯的错：把任务状态藏在对话历史里，指望模型自己"记住"现在做到哪一步了。这在简单任务上能跑，但稍微复杂一点就翻车。原因有三：

1. LLM 不擅长精确状态追踪。它擅长语言生成，但让它回答"现在子任务 3 是 in_progress 还是 done"，准确率惨不忍睹。
2. 压缩会丢状态信息。Ch.4 讲的摘要压缩，很可能把"第 5 步的中间结果"压没了。
3. 不可观测、不可调试。Agent 跑了 100 轮，出问题时你没法点开"看看现在的状态"——一切都在文本里漂浮。

5.2需要追踪哪些状态

取决于你的任务，但通常包括：

• task：原始目标、当前任务状态（pending/running/completed/failed）、最终结果。
• plan：分解后的子任务列表、当前在哪一步、每个子任务的状态。
• budget：已用步数、已用 token、已用时间。和 Ch.2 的停止条件呼应。
• workspace：工作目录、产出的文件、scratchpad（临时草稿）。Ch.6 详谈。
• history：发生过的关键事件——错误、重要决策、关键工具调用。可以理解为 Agent 的"日志"。

5.3一个简单的 State 实现

from dataclasses import dataclass, field
from enum import Enum
import time, json

class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    DONE    = "done"
    FAILED  = "failed"

@dataclass
class Subtask:
    id: int
    description: str
    status: TaskStatus = TaskStatus.PENDING
    result: str = ""

@dataclass
class AgentState:
    goal: str
    status: TaskStatus = TaskStatus.PENDING
    plan: list = field(default_factory=list)
    current_step: int = 0
    tokens_used: int = 0
    started_at: float = field(default_factory=time.time)
    workspace_dir: str = "./workspace"
    events: list = field(default_factory=list)

    def log(self, kind, msg):
        self.events.append({"t": time.time(), "kind": kind, "msg": msg})

    def snapshot(self):
        """把状态渲染成一段文字给 LLM 看"""
        plan_str = "\n".join(
            f"  [{s.status.value}] {s.id}. {s.description}" for s in self.plan
        ) or "  (尚未规划)"
        return f"""
[Agent 当前状态]
目标: {self.goal}
总状态: {self.status.value}
当前步骤: {self.current_step}
已用 token: {self.tokens_used}
计划:
{plan_str}
""".strip()

    def persist(self, path):
        """把状态序列化到磁盘，用于 checkpoint"""
        with open(path, "w") as f:
            json.dump(self.__dict__, f, default=str, indent=2, ensure_ascii=False)

注意三个关键方法：

• log()：记录事件。出问题时回放整个 Agent 的"心路历程"。
• snapshot()：把状态变成给 LLM 看的文字。这是状态注入 context 的方式。
• persist()：写到磁盘。配合下一节的 checkpoint。

5.4状态注入：让 LLM 知道"现在到哪了"

状态是给你看的——但 LLM 也得知道当前进展，才能做出对的决策。做法是：每轮 LLM 调用前，把状态 snapshot 插进 system message。

def build_messages(state, history):
    return [
        {"role": "system", "content": SYSTEM_PROMPT + "\n\n" + state.snapshot()},
        *history,
    ]

这样模型每一轮看到的 system 都是最新的状态，再也不会"忘了自己规划了什么"。

5.5Checkpoint：能恢复的 Agent 才是工程级的

长任务跑到一半，机器崩了、网络断了、API 限流了——如果你没做 checkpoint，所有进度作废，钱白烧。所以：

def run_with_checkpoint(goal, checkpoint_path="./state.json"):
    # 1. 尝试恢复
    if os.path.exists(checkpoint_path):
        state = AgentState.load(checkpoint_path)
        print(f"[Resume] 从 step {state.current_step} 恢复")
    else:
        state = AgentState(goal=goal)

    # 2. 主循环
    while state.status == TaskStatus.RUNNING:
        step_one(state)
        state.persist(checkpoint_path)   # ← 关键的一行

5.6状态机思维：把 Agent 想成一个 FSM

更进一步，你可以把整个 Agent 看作一个有限状态机。状态转移由 LLM 的输出驱动。这种视角的好处是：你能画出一张状态图，明确每个状态下"允许做什么"、"不允许做什么"。

这种思维方式不是必须的（很多 Agent 没显式 FSM 也能跑），但当你做多 Agent或复杂工作流时（比如你的群聊式 Agent 平台），状态机思维会救你的命——你的 agent runtime 本质上就是一个分布式状态机。

文件管理 Files

文件管理常常被新手低估——他们觉得"不就是 read_file / write_file 两个工具吗"。但实际上，文件系统是 Agent 工程里最被低估的设计杠杆。它同时是：内存的延伸、状态的载体、Agent 之间的通信介质、人类介入的接口。这一章告诉你为什么以及怎么做。

6.1文件，是 Agent 的"外脑"

回忆 Ch.4 我们提过的"把大对象搬到文件里"。把这个想法推到极致，你会发现 Agent 设计可以有一个截然不同的范式：

6.2工作目录的设计

不要让 Agent 满世界乱写文件。给它一个明确的工作目录（workspace），并约定好结构：

workspace/
├── goal.md              # 用户最初给的目标，永远不变
├── plan.md              # Agent 自己写的当前计划
├── scratchpad.md        # 临时草稿、思考过程
├── inputs/              # 用户提供的输入文件
│   └── ...
├── results/             # 工具产生的中间结果
│   ├── data_001.json
│   └── ...
└── outputs/             # 最终给用户的产出
    └── ...

把这套结构告诉 Agent（写在 system prompt 里），它会自然地按这个约定办事。这远比每次都让 LLM 自己决定"文件存哪"要稳。

6.3必备的文件工具

四件套，加两个进阶工具：

# 基础四件套
def read_file(path):
    """读取文本文件。超大文件会被截断到前 50KB。"""
    full_path = _safe_path(path)
    with open(full_path) as f:
        content = f.read(50_000)
    return content

def write_file(path, content):
    """把内容写入文件。如果父目录不存在会自动创建。"""
    full_path = _safe_path(path)
    os.makedirs(os.path.dirname(full_path), exist_ok=True)
    with open(full_path, "w") as f:
        f.write(content)
    return f"已写入 {path}，{len(content)} 字符"

def list_files(path="."):
    """列出目录下的文件与子目录。"""
    full_path = _safe_path(path)
    items = []
    for name in sorted(os.listdir(full_path)):
        p = os.path.join(full_path, name)
        size = os.path.getsize(p) if os.path.isfile(p) else "-"
        kind = "DIR" if os.path.isdir(p) else "FILE"
        items.append(f"{kind:5} {size:>10} {name}")
    return "\n".join(items)

def delete_file(path):
    full_path = _safe_path(path)
    os.remove(full_path)
    return f"已删除 {path}"

# 进阶两件
def grep_files(pattern, path="."):
    """在文件中搜索关键词，避免读全文。"""
    # ...用 ripgrep 或 Python re 实现，返回匹配行

def edit_file(path, old_str, new_str):
    """精确替换：把 old_str 替换为 new_str（必须唯一）。"""
    """避免每次改一个字符都要重写整个文件。"""

这里的 _safe_path 至关重要——下面专门讲。

6.4沙盒隔离：不要让 Agent 写穿你的硬盘

如果你不做路径校验，Agent 一旦发飙（或者被 prompt injection），可以 read_file("/etc/passwd")、write_file("../../../etc/cron.d/evil", ...)。这不是假想威胁，是非常现实的。

def _safe_path(path):
    workspace = os.path.realpath(WORKSPACE_DIR)
    full = os.path.realpath(os.path.join(workspace, path))
    if not full.startswith(workspace + os.sep) and full != workspace:
        raise ValueError(f"路径 {path} 越界，不在 workspace 内")
    return full

注意要用 os.path.realpath——它会解析符号链接和 ..。直接 startswith 字符串比较会被 ../ 绕过。

更强的隔离方式：容器。把 Agent 跑在 Docker 里，宿主机挂载只读，只有 workspace 目录可写。这样即便所有 Python 层校验都失效，影响也限制在容器内。生产环境强烈推荐。

6.5让 LLM 知道 workspace 长什么样

有了文件系统，下一个问题是：LLM 怎么知道现在 workspace 里有哪些文件？

两种做法：

1. 被动型：LLM 想知道就自己 list_files()。简单但费一次工具调用。
2. 主动型：每轮 LLM 调用前，runtime 自动把 workspace 的目录树附加到 system message。LLM 永远看到最新结构。

实战推荐主动型，配合 Ch.5 的状态注入：

def render_context(state):
    tree = render_tree(state.workspace_dir, max_depth=3)
    return f"""
{state.snapshot()}

[工作目录]
{tree}
""".strip()

这样 LLM 每次都看到："哦，我在 results/ 下面已经有了三个 json，outputs/ 还空着"——它能基于事实做决策，而不是基于回忆。

6.6文件作为"任务接力"的介质

这一节稍微往前一步，给你的多 agent 平台埋个伏笔。

当多个 agent 协作时，最干净的"沟通方式"不是直接互发消息，而是通过共享文件接力：

• Agent A 完成调研，把结论写到 research.md。
• Agent B 接到任务"基于 research.md 写报告"，读取后处理。
• Agent C 接到任务"review 这份报告"，读取后给反馈，写到 review.md。

这种"文件即消息"的模式有三个巨大好处：

1. 异步天然：A 不需要在线，B 直接读它的产物。
2. 可审计：每个 agent 的产出都有持久化痕迹。
3. 人可介入：人类随时打开文件查看、修改，无缝插入工作流。

你那个群聊式 Agent 平台，如果把"任务的中间产物"都落成文件，配合一个简单的事件机制（A 写完文件触发 B），整个协作就清晰多了。

完整实现 Putting It Together

理论讲完了，工具一个一个剖了，是时候把它们焊在一起。这一章给你一份大约 220 行的可运行 Python，把前六章所有概念——loop、tool registry、memory manager、agent state、file manager——组装成一个能跑的最小 Agent。读完你可以直接 python agent.py。

7.1项目结构

先看整个 Agent 的代码组织——刻意保持极简，6 个模块对应前面 6 章：

agent/
├── llm.py          # LLM 客户端（薄薄一层）
├── tools.py        # ToolRegistry + 标准工具
├── memory.py       # MemoryManager
├── state.py        # AgentState
├── files.py        # FileManager + 沙盒
└── agent.py        # 主循环——把上面的串起来

每个文件 30~50 行。下面把核心代码给你——为了篇幅省略了一些 import 和异常分支，但骨架完整，你照搬就能跑。

7.2组件一：LLM 客户端

用 OpenAI SDK 兼容协议——这样你可以用 OpenAI、DeepSeek、Moonshot、本地 vLLM，只要换 base_url 和 model name。

# llm.py
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("LLM_API_KEY"),
    base_url=os.getenv("LLM_BASE_URL", "https://api.deepseek.com/v1"),
)
MODEL = os.getenv("LLM_MODEL", "deepseek-chat")

def chat(messages, tools=None):
    """一次性调用，返回完整 message dict。"""
    resp = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=tools,
        tool_choice="auto" if tools else None,
        temperature=0.2,
    )
    msg = resp.choices[0].message
    return {
        "role": "assistant",
        "content": msg.content or "",
        "tool_calls": [tc.model_dump() for tc in (msg.tool_calls or [])],
    }

7.3组件二：工具与注册表

沿用 Ch.3 的设计，但加上 file 工具的接入：

# tools.py
import json, subprocess, requests
from files import read_file, write_file, list_files, edit_file

class ToolRegistry:
    def __init__(self):
        self.fns = {}; self.schemas = []

    def register(self, name, fn, description, parameters):
        self.fns[name] = fn
        self.schemas.append({
            "type": "function",
            "function": {"name": name, "description": description,
                         "parameters": parameters},
        })

    def call(self, name, args_json):
        try:
            args = json.loads(args_json) if isinstance(args_json, str) else args_json
            result = self.fns[name](**args)
            return str(result)[:4000]   # 截断保护
        except Exception as e:
            return f"[tool error] {type(e).__name__}: {e}"

# 标准工具集
def build_default_registry():
    r = ToolRegistry()
    r.register("read_file", read_file,
        "读取 workspace 内文件的全部内容",
        {"type": "object",
         "properties": {"path": {"type": "string"}},
         "required": ["path"]})
    r.register("write_file", write_file,
        "把 content 写到 workspace/path（覆盖）",
        {"type": "object",
         "properties": {"path": {"type": "string"},
                        "content": {"type": "string"}},
         "required": ["path", "content"]})
    r.register("list_files", list_files,
        "列出 workspace/path 下的文件，默认根目录",
        {"type": "object",
         "properties": {"path": {"type": "string", "default": "."}}})
    r.register("finish", lambda reason="": f"FINISH:{reason}",
        "任务已完成时调用，runtime 会停止循环",
        {"type": "object",
         "properties": {"reason": {"type": "string"}}})
    return r

7.4组件三：记忆管理器

Ch.4 设计的极简版本——窗口 + 摘要：

# memory.py
from llm import chat

class MemoryManager:
    def __init__(self, max_recent=20):
        self.system = []      # 永远保留：[system, goal]
        self.summary = ""     # 滚动压缩的历史摘要
        self.recent = []      # 最近 N 轮
        self.max_recent = max_recent

    def set_system(self, msgs): self.system = msgs

    def add(self, msg):
        self.recent.append(msg)
        if len(self.recent) > self.max_recent:
            self._compress()

    def render(self):
        msgs = list(self.system)
        if self.summary:
            msgs.append({"role": "system",
                         "content": f"[之前进展摘要] {self.summary}"})
        msgs.extend(self.recent)
        return msgs

    def _compress(self):
        # 把前一半压缩成摘要，保留后一半
        half = len(self.recent) // 2
        old, new = self.recent[:half], self.recent[half:]
        prompt = [
            {"role": "system",
             "content": "用 5 行内总结对话进展，保留关键决策、错误、未完成事项。"},
            {"role": "user", "content": str(old)},
        ]
        summary = chat(prompt)["content"]
        self.summary = (self.summary + "\n" + summary).strip()
        self.recent = new

7.5组件四：状态

极简 dataclass——Ch.5 那一套的最小版：

# state.py
from dataclasses import dataclass, field
import json, time

@dataclass
class AgentState:
    goal: str = ""
    status: str = "running"     # running | done | failed
    step: int = 0
    max_steps: int = 25
    budget_tokens: int = 0
    started_at: float = field(default_factory=time.time)

    def snapshot(self):
        return json.dumps(self.__dict__, ensure_ascii=False, indent=2)

    def should_stop(self):
        return self.status != "running" or self.step >= self.max_steps

7.6组件五：文件沙盒

Ch.6 的安全实现，原样搬过来：

# files.py
import os
WORKSPACE = os.path.realpath(os.getenv("AGENT_WORKSPACE", "./workspace"))
os.makedirs(WORKSPACE, exist_ok=True)

def _safe(path):
    full = os.path.realpath(os.path.join(WORKSPACE, path))
    if not (full == WORKSPACE or full.startswith(WORKSPACE + os.sep)):
        raise ValueError(f"路径 {path} 越界")
    return full

def read_file(path):
    with open(_safe(path), encoding="utf-8") as f:
        return f.read()

def write_file(path, content):
    full = _safe(path)
    os.makedirs(os.path.dirname(full), exist_ok=True)
    with open(full, "w", encoding="utf-8") as f:
        f.write(content)
    return f"已写入 {path}（{len(content)} 字符）"

def list_files(path="."):
    full = _safe(path)
    if not os.path.isdir(full): return f"{path} 不是目录"
    return "\n".join(sorted(os.listdir(full)))

def edit_file(path, old_str, new_str):
    text = read_file(path)
    if text.count(old_str) != 1:
        return f"[error] old_str 必须恰好出现 1 次，实际 {text.count(old_str)} 次"
    write_file(path, text.replace(old_str, new_str))
    return "编辑完成"

7.7主循环：把所有东西串起来

这是最关键的一段，也是整个 Agent 的"心跳"。读完你应该能清晰看到前 6 章的每一个概念都落到了哪里：

# agent.py
import json
from llm import chat
from tools import build_default_registry
from memory import MemoryManager
from state import AgentState
from files import list_files

SYSTEM = """你是一个目标驱动的 agent。每一轮你必须：
1) 先 thought：用一句话说明现在打算做什么、为什么。
2) 再调用一个工具（finish 表示任务完成）。
你可以多次调用工具直到目标达成。所有产物写到 workspace 文件里。
不要在 content 里放大段结果，结果都通过 write_file 落盘。"""

def run(goal: str):
    state = AgentState(goal=goal)
    mem = MemoryManager(max_recent=20)
    tools = build_default_registry()

    mem.set_system([
        {"role": "system", "content": SYSTEM},
        {"role": "user",   "content": f"目标：{goal}"},
    ])

    while not state.should_stop():
        state.step += 1

        # 1) 把当前状态 + 工作目录 注入
        ctx_note = {"role": "system",
                    "content": f"[state]\n{state.snapshot()}\n\n"
                               f"[workspace]\n{list_files('.')}"}
        msgs = mem.render() + [ctx_note]

        # 2) 调 LLM
        reply = chat(msgs, tools=tools.schemas)
        mem.add({"role": "assistant",
                 "content": reply["content"],
                 "tool_calls": reply["tool_calls"]})

        # 3) 没有工具调用 → 视为它在等用户，跳出
        if not reply["tool_calls"]:
            print(f"[step {state.step}] no tool call, stopping.")
            break

        # 4) 执行所有工具调用
        for tc in reply["tool_calls"]:
            name = tc["function"]["name"]
            args = tc["function"]["arguments"]
            result = tools.call(name, args)
            print(f"[step {state.step}] {name}({args[:80]}) → {result[:120]}")
            mem.add({"role": "tool",
                     "tool_call_id": tc["id"],
                     "content": result})
            if result.startswith("FINISH:"):
                state.status = "done"

    print(f"\n[done] status={state.status} steps={state.step}")
    return state

if __name__ == "__main__":
    import sys
    run(sys.argv[1] if len(sys.argv) > 1
        else "调研一下 Rust 和 Go 在并发模型上的差异，写一份 1000 字对比报告到 outputs/report.md")

7.8一次真实运行的轨迹

让上面这个 Agent 跑一个目标"调研 Rust 和 Go 的并发模型差异并写报告"，下面是终端会看到的输出（简化版）。注意观察每一步的因果：

[step 1] list_files(.) → outputs
[step 2] write_file(outputs/plan.md, ...) → 已写入 outputs/plan.md
[step 3] write_file(outputs/rust_notes.md, ...) → 已写入 outputs/rust_notes.md
[step 4] write_file(outputs/go_notes.md, ...) → 已写入 outputs/go_notes.md
[step 5] read_file(outputs/rust_notes.md) → # Rust 并发要点\n- 所有权 + Send/Sync...
[step 6] read_file(outputs/go_notes.md) → # Go 并发要点\n- goroutine + channel...
[step 7] write_file(outputs/report.md, ...) → 已写入 outputs/report.md（约 4200 字符）
[step 8] finish(reason="报告完成") → FINISH:报告完成

[done] status=done steps=8

看出来了吗？Agent 自发地：先列了下当前目录看有什么，写了个计划文件，再分别整理两个语言的笔记到独立文件，最后读回笔记拼成报告。所有中间产物都落盘，messages 里只剩很短的状态。这就是"文件作为外脑"的实战形态。

7.9跑起来需要什么

# 安装
pip install openai

# 配置（DeepSeek 便宜实在，也可以用 OpenAI / Moonshot / 本地 vLLM）
export LLM_API_KEY="sk-..."
export LLM_BASE_URL="https://api.deepseek.com/v1"
export LLM_MODEL="deepseek-chat"
export AGENT_WORKSPACE="./workspace"

# 跑
python agent.py "你的目标..."

进阶之路 Beyond MVP

Ch.7 那 220 行能跑、能干活，但还不够稳，也不够聪明。从这里到真正能放进生产、能多 agent 协作、能被信任去做有副作用的事，中间还有一段路。这一章不写代码，给你一张路线图：四个真正决定 Agent 能力上限的方向，以及每个方向你应该读什么、做什么、避开什么坑。

8.1方向一：规划（Planning）

Ch.2 的 ReAct 是"边走边看"。这种打法在简单任务上很灵，但任务一旦超过 10 步、需要回头改前面的决定，就会暴露问题——LLM 容易在中间走偏，绕回原点。

解药是显式规划。常见三种模式，按复杂度递增：

第三种是 HTN / 分层任务网——大任务拆子任务，子任务再拆。在多 agent 系统里，"上层 agent 出规划、下层 agent 执行子任务"基本是默认形态。你那个群聊式平台里"调度官 agent + 执行 agent"的分工，本质就是 HTN。

8.2方向二：反思（Reflection）

LLM 写完一段代码、写完一份报告，你直接交付吗？人类专家不会——他们会回头看一眼。Agent 也应该。

反思的本质是再开一轮 LLM 调用，让它当自己作品的批判者：

producer 模式（执行）   → 写代码 → 输出 v1
critic 模式（自我审视）  → 看 v1 → 列出问题清单
producer 模式（修复）    → 根据清单 → 输出 v2
critic 模式（再审）      → 看 v2 → "ok" 或继续

关键技巧：同一段 prompt 让 LLM 既当作者又当审稿人，效果会大打折扣——它会偷懒说"挺好的"。正确做法是切换角色 prompt，让 critic 那一轮的 system message 明确：

• "你是一个挑剔的代码 reviewer，目标是找出问题。"
• "不要夸奖任何东西，只列出可改进的点，按严重性排序。"
• "如果你找不到问题，就退回最有可能出错的地方说出你的怀疑。"

反思要避免的两个坑：

1. 无限反思：设硬上限（比如最多 2 轮 critic）。否则会反复改鸡毛蒜皮，token 烧穿。
2. 反思污染：critic 的输出别全塞回 producer 的 context——只把"action item 列表"传过去。

8.3方向三：多 Agent 协作

这个对你来说应该最有共鸣——你的平台就是奔着这里去的。我把多 agent 系统按耦合度从低到高分四档：

三种模式不是替代关系，是组合关系。你那个平台最终大概率是混合体：

• 群聊（B）作为人与 agent 共同的可见层——所有动作可观察。
• 文件接力（A）作为大产物的传递通道——别把整份报告灌进群聊。
• 编排（C）处理那些"需要拆分并行做"的子任务——临时召唤一组 worker，做完汇总。

多 agent 工程上最难的不是协议设计，是三个"反"：

1. 反循环：A 给 B 发任务，B 反手把任务又发给 A——必死。要在调度层加环路检测。
2. 反风暴：一个事件触发 N 个 agent 同时回应，把消息总线打爆。每个 agent 要有"是否该接话"的判定，而不是无条件响应。
3. 反失能：当一个 agent 卡死或返回垃圾时，系统不能整体停摆。每个任务要有 owner、超时、降级路径。

8.4方向四：评估（Eval）

这条放最后，但其实最重要。你没办法改进你测不出来的东西。大部分新手做 Agent 都跳过这一步，凭感觉调 prompt，结果就是改对了一个 case 改坏了三个 case，永远在原地打转。

Agent 的 eval 和传统 ML 不一样，因为输出不是单一答案，而是一条轨迹。需要从三个维度看：

第三个维度——稳健性：同一个目标跑 10 次，成功几次？方差多大？Agent 的随机性比你想的大，单次成功完全可能是运气。任何一个 prompt 修改都应该跑 N≥10 次取均值才算数。

实操层面，建议从一开始就建立三层 eval pipeline：

1. 单元测试：每个工具单独测——给定输入是否返回正确输出。最便宜，必须 100% 通过。
2. 轨迹测试：跑 20~50 个典型目标，记录完整轨迹和最终产物，人工标注 + LLM judge。
3. 生产监控：上线后每个 session 自动记录步数、token、失败原因。每周看一次分布。

8.5结语：给你的 6 个月路线

把上面所有东西落到时间线上：

第 1~2 周   把 Ch.7 那 220 行跑通，跑 5 个目标，记录失败 case
第 3~4 周   加 plan.md 显式规划 + 反思一轮，再跑同样的目标
第 5~8 周   把单 agent 嵌进你的群聊框架（OpenClaw runtime）
第 9~12 周  做调度官 agent + 任务认领协议
第 13~20 周 上 eval pipeline，开始基于数据迭代
第 21~24 周 跑真实用户、修真实问题，准备 MVP 发布

这条路上你会遇到的最大陷阱不是技术上的，而是过早抽象——总想一上来就把架构做"对"。但 Agent 这个东西现在还没有"对的架构"，全是 trade-off。第一版能跑、能改、能观察，比"漂亮"重要 100 倍。Claude Code、Cursor 早期都是一坨毛线一样的代码——它们活下来是因为能基于真实使用迭代。

尾声 Coda

读完这 8 章，你应该有一种"原来就这样"的释然——而不是"还有 100 个秘密我不知道"的焦虑。这是对的感觉。Agent 不是黑魔法，是一个 LLM 加上一个执着的 for 循环，加上你愿意为它建造的一切外部世界。

真正把你和 90% 的人区分开的，不是知道这些理论——是你真的去搭了一个，跑挂了，调好，又跑挂了。这本手册到这里截止，但你的 Agent 工程师生涯从你打开 IDE 的那一刻才开始。

愿你的 Agent 跑得稳，长得快，永远
不要在凌晨三点给你发奇怪的工具调用。