一个智能体（Agent）产品的外壳千差万别，但拆到底都是同一个循环在转圈。我用 Python 从零复刻了 WorkBuddy 的核心能力（约 9000 行后端 + 700 行前端、160 项测试），这篇文章把每个功能点「怎么实现的、代码长什么样、业界怎么做」一次讲透。

引言：为什么要"复刻"而不是"调库"

市面上 Agent 框架已经很多了：LangGraph、OpenAI Agents SDK、CrewAI……如果只是"跑起来一个能干活的 Agent"，接个框架半天就够。但框架把最关键的东西替你抽象掉了——循环怎么转、上下文怎么预算、工具报错怎么回喂。这些恰恰是 Agent 开发的核心竞争力所在。

所以我选择从零手写：不追求功能全，追求每个模块都能讲清楚"为什么这样做"。代码里写满了踩坑注释，测试对应每条护栏。项目分层如下：

┌──────────────────────────────────────────────────┐
│  交互层    CLI / FastAPI + SSE / React 前端        │
├──────────────────────────────────────────────────┤
│  编排层    core/loop.py（ReAct）core/runtime.py    │
│           planning/plan_mode.py automations/      │
├──────────────────────────────────────────────────┤
│  能力层    core/registry.py tools/ skills/         │
│           memory/ mcp/ core/approval.py           │
├──────────────────────────────────────────────────┤
│  模型层    core/provider.py（OpenAI 兼容适配）      │
├──────────────────────────────────────────────────┤
│  基础设施  core/context.py（预算）sandbox/          │
│           storage/（SQLite）core/events.py（SSE）  │
└──────────────────────────────────────────────────┘

两个核心模块（loop.py 的 ReAct 循环、context.py 的上下文预算）决定整个系统的效果与成本，其余全部是插件式挂载。下面按重要性逐个展开。

项目源码：

• github
• gitee

一、ReAct 主循环：整个系统的心脏

思想

ReAct（Yao et al., 2022）的洞见是推理和行动必须交织：只推理（CoT）没有手，全靠幻觉填补事实；只行动没有脑，工具报错就卡死。现代实现里，论文的 Thought/Action/Observation 文本约定已升级为 function calling 协议字段：

论文概念	现代协议对应
Thought	assistant 消息的 content
Action	tool_calls 结构化字段
Observation	role: "tool" 消息回填
Final Answer	一轮响应 tool_calls 为空 = 循环出口

实现

core/loop.py，一个 for turn in range(1, max_turns+1) 的异步循环，每轮四步：

for turn in range(1, deps.max_turns + 1):
    assembled = ctx.build()                     # ① 组装上下文（含预算裁剪）
    response = await deps.provider.chat(        # ② 模型决策（流式，token 即时 emit）
        assembled.messages, tools=tool_schemas or None,
        stream=True, on_token=on_token,
    )
    ctx.add(Message(role=Role.ASSISTANT,
                    content=response.content or None,
                    tool_calls=response.tool_calls or None))

    if not response.has_tool_calls:             # ③ 没工具调用 = 任务完成
        answer = response.content
        break

    for call in response.tool_calls:            # ④ 执行工具，结果回喂上下文
        result, denied = await _execute_call(call, deps, approve=approve)
        ctx.add(Message(role=Role.TOOL,
                        content=result.for_llm(limit=deps.max_tool_output_chars),
                        tool_call_id=call.id))

三道护栏缺一不可，否则线上会烧掉大量 token 甚至死循环：

# ① 最大轮次：max_turns=24，用尽置 stop_reason="max_turns"
# ② token 预算：ContextBuilder 强制 + build() 后一道保险
if assembled.total_tokens > deps.context_max_tokens:
    stop_reason = "context_limit"
    break
# ③ 重复调用检测：滑动窗口，工具名+参数签名连续 3 次全同即中止
recent_signatures.append(call.signature())
if len(recent_signatures) >= 3 and len(set(recent_signatures[-3:])) == 1:
    stop_reason = "repeated_call"
    break

两个新手必踩的坑，代码里专门处理了：

1. 每个 tool_call 必须紧跟一条 role=tool 消息——哪怕参数解析失败、用户拒绝执行，也要补写一条把原因告诉模型，否则 OpenAI 兼容 API 直接 400；
2. 模型出错不抛异常，而是"教它怎么改"：

# _execute_call 里：非法 JSON 参数 → 转成一条指导性失败文本回喂
if call.parse_error:
    return ToolResult.failure(
        f"你调用 `{call.name}` 时传入的参数不是合法 JSON：{call.parse_error}\n"
        "请用正确的 JSON 对象重新调用（注意：键名和字符串都要用双引号）。")
# 幻觉出不存在的工具 → 把可用清单报给它
if tool is None:
    return ToolResult.failure(f"工具 `{call.name}` 不存在。可用工具: {available}。请重新选择。")

还有一个细节：模型既没内容也没工具调用时不退出，注入一条"请继续"轻推，最多 2 次（stop_reason="max_nudges"）——真实模型偶尔会"哑火"，直接退出体验很差。

业界怎么做

同一范式，产品级 Agent（Claude Code / Codex CLI / WorkBuddy）都是 while 循环直实现——Claude Code 官方说过核心 loop 约 1500 行。差异全在循环外围：

• 轮次：产品级支持 50+ 轮长任务，配合上下文压缩（见下一节）；
• 并行工具调用：一轮响应可含多个 tool_calls，框架并发执行。本项目是顺序 for 循环，这是最该演化的差异点（asyncio.gather + 按原顺序回填，约 20 行）；
• 框架派：LangGraph 把循环显式化为状态机（agent 节点 ↔ tools 节点），换取 checkpoint / 时间旅行 / 显式人工介入点；OpenAI Agents SDK 用 handoff 把"多智能体"也规约进循环；
• 训练层：o1/o3、DeepSeek-R1 这类推理模型在 RL 时就把"工具使用 + 环境反馈"内化进权重，模型自发涌现 ReAct 行为。这也是为什么 2022 年靠正则抠 Action: 的脆弱文本协议已被淘汰。

二、工具系统：一个能力 = 一个类

实现

tools/base.py + core/registry.py。核心设计：Pydantic 入参模型自动生成 JSON Schema，加能力不改循环：

class Tool(ABC, Generic[TArgs]):
    name: ClassVar[str]
    description: ClassVar[str]
    args_model: ClassVar[type[BaseModel]]
    timeout: ClassVar[float] = 30.0
    dangerous: ClassVar[bool] = False       # 高危标记 → 接人工审批

    @property
    def schema(self) -> dict[str, Any]:     # 直接就是 OpenAI function calling 格式
        return {"type": "function", "function": {
            "name": self.name,
            "description": self.description,
            "parameters": self.args_model.model_json_schema(),  # ★ 自动推导
        }}

写一个真实工具长这样（tools/file_ops.py）：

class ReadFileArgs(BaseModel):
    path: str = Field(..., description="文件路径，可以是相对于工作区的相对路径")
    offset: int = Field(default=0, description="从第几行开始读")
    limit: int = Field(default=300, description="最多读取多少行，建议不要一次读太多以免浪费上下文")

class ReadFile(Tool[ReadFileArgs]):
    name: ClassVar[str] = "read_file"
    description: ClassVar[str] = "读取文本文件内容并返回带行号的内容。文件很大时请用 offset/limit 分段读取。"
    ...

注意 limit 的 description 里藏着"建议不要一次读太多以免浪费上下文"——description 是模型唯一的决策依据，教模型用好工具比调 temperature 有效得多。这是我这个项目里反复验证的第一性经验。

registry.execute() 是统一执行闸口，它的模块注释写着整个系统的哲学：

"""最重要的一条约定：execute 永远不抛异常。
工具内部的任何异常都会被转成 ToolResult.failure，回喂给模型让它重试或改策略。
一个在工具里崩掉的 Agent，就不是 Agent 了——它连自我修正的机会都没有。"""

try:
    args = tool.args_model.model_validate(raw_arguments or {})
except ValidationError as exc:
    return ToolResult.failure(f"调用 `{name}` 的参数校验失败：\n{...}")
try:
    return await asyncio.wait_for(tool.run(args), timeout=limit)   # 超时掐断
except TimeoutError:
    return ToolResult.failure(f"工具 `{name}` 执行超时（超过 {limit} 秒）。请换更轻量的方式。")
except Exception as exc:
    return ToolResult.failure(f"工具 `{name}` 内部异常：{type(exc).__name__}: {exc}")

参数校验失败、超时、内部崩溃，全部转成结构化错误文本回喂，循环永不中断。

业界怎么做

同构。Anthropic 的 tool use 规范、OpenAI function calling 都走 JSON Schema。行业额外卷的三点：

• 结果截断策略：MCP 官方建议客户端对超长工具输出做头尾截断（本项目 8000 字符）；
• description 工程：Anthropic 文档专门讲如何写工具描述，社区实践是"把何时用/何时不用/参数示例全写进 description"；
• 协议统一：MCP（2024.11）解决"工具供给侧"——工具不该每个 Agent 各写一遍。

三、上下文工程：决定 Agent 多聪明、多贵

实现

core/context.py。上下文窗口是 Agent 的"工作内存"，所有信息都在抢这块地。预算公式：

system = self._build_system_message()   # base_prompt + 记忆/技能各 section
budget = int(self.max_tokens * self.reserve_ratio) - system.tokens - SAFETY_MARGIN

裁剪顺序体现优先级观——build() 里三层递进（先蒸馏旧工具、再截超长、最后丢老消息）：

# 第零步：旧轮次工具输出折叠成短存根（历史蒸馏，下一小节详述）

# 第一步：超长工具输出做「首尾保留式」截断（头 70% 尾 20%，中间省略）
head = int(self.max_tool_output_chars * 0.7)
tail = int(self.max_tool_output_chars * 0.2)
content = message.content[:head] + "\n...[中间 N 字符已省略]...\n" + message.content[-tail:]

# 第二步：仍超限则丢最早的消息，死守最近 6 条底线
while total > budget and len(history) > self.keep_recent_messages:
    removed = history.pop(0)
    total -= removed.tokens

为什么工具输出先截？它是最常见的上下文爆炸源（一次 cat 大文件就没了），而头尾往往含关键信息（错误在头部、结论在尾部）。system 消息永不裁——丢了 Agent 就忘了自己是谁。

一个真实的坑：历史没丢，但 Agent 还是"不聪明"

上线多轮对话后收到反馈：让 Agent 规划一周行程，三轮下来它每轮都在重新搜索，像没听懂前面的话。第一反应是"历史丢了"——查下来发现根本不是：

第1轮 完成：turns=9  tokens≈23166
第2轮 完成：turns=7  tokens≈44317   ← 单调增长，历史完整加载
第3轮 完成：turns=6  tokens≈70359

token 一路涨到 70k，一条"上下文超限，已丢弃"日志都没有，62 条消息全在库里。真正的病因是信噪比崩塌：

消息类型	条数	字符量	占比
tool（网页搜索/抓取正文）	34	~37000	94%
assistant（含行程结论）	25	~2400	6%
user（原始需求）	3	54 字	0.1%

用户的真实需求和已经查到的结论，被稀释在 94% 的原始网页噪声里。模型不是忘了，是对自己历史里的结论不信任，宁可重新搜一遍"求证"。

修法不是加窗口，是历史蒸馏：只保留最近 N 个用户轮次的完整工具输出，更早轮次的工具转写折叠成短存根——

def _stub_tool_output(self, message: Message) -> tuple[Message, bool]:
    content = message.content or ""
    if len(content) <= self.tool_stub_chars:
        return message, False
    stub = Message(
        role=message.role,
        content=(
            content[: self.tool_stub_chars]
            + f"\n...[该工具输出产生于较早轮次，其余 {len(content) - self.tool_stub_chars} 字符已省略；"
            "如需其中信息，请重新调用该工具获取]"
        ),
        tool_call_id=message.tool_call_id,  # 必须保留，否则助手消息缺应答，API 直接 400
    )
    return stub, True

以 role=USER 为轮次分界，build() 时算出"最近 compact_turns 轮"的起始下标，之前的工具消息走存根。三条纪律：用户消息和助手结论永不折叠（那是对话的语义主干）；tool_call_id 必须保留（OpenAI 协议硬约束）；存根里明确写"需要可重新调用"（给模型留出路，而不是单纯删信息）。

效果直接体现在日志上——同样三轮对话：

修复前:  23166 → 44317 → 70359 tokens，每轮 6~9 turns，反复重新搜索
修复后:   5465 →  5588 →  5777 tokens，第 2、3 轮 turns=1 steps=0（零工具调用）

第 2、3 轮不再搜索，直接引用历史结论作答。上下文治理省下来的不只是 token，还有模型的"注意力"。

配套做了两件事：web 类工具输出源头收紧（搜索结果摘要 300→200 字符，web_fetch 默认 8000→4000，参数 description 里写明"别调大，长文本会挤占后续所有轮次的预算"）；降级 token 估算修正——原来的 len//2 对中文严重偏乐观（汉字在主流 BPE 里约 1 字 ≈ 1 token 而非 0.5），tiktoken 离线拿不到 BPE 文件时，预算守护的"尺子"本身就准不了，改成按 CJK / ASCII 分类计数。

坑的续集：跨轮治好了，单轮内还在爆

蒸馏上线后同一个行程任务再测，暴露出对称的问题："安排 7 天大兴安岭行程"只有一条用户消息，Agent 在一轮里连搜 6-7 次网页，每次输出都在"最近 N 个用户轮"之内——跨轮蒸馏帮不上忙，上下文照样堆到撞线。

所以蒸馏补了第二条规则，两条互补：

def _recent_tool_boundary(self) -> int:
    """轮内蒸馏：全局只保留最近 K 条工具输出全量，更早的一律折叠。"""
    tool_indices = [i for i, m in enumerate(self.messages) if m.role is Role.TOOL]
    if len(tool_indices) <= self.keep_recent_tool_outputs:
        return -1
    return tool_indices[-self.keep_recent_tool_outputs]

跨轮规则管"用户说过几句话"，轮内规则管"工具跑了多少次"，build() 里任一命中即折叠。同一批修复还有两件配套事：

• 模型窗口封顶。CONTEXT_MAX_TOKENS=100000 配 deepseek-chat（真实窗口 64k）是错配——我们护栏"合法"，API 直接 400。加了内置 MODEL_CONTEXT_WINDOWS 表按模型名推断（支持中转站变体名的模糊匹配），软预算取 min(自设值, 真实窗口)。"宣称窗口 ≠ 有效窗口"这件事上一节已经用信噪比证明过，这里是它在配置层的镜像。
• 撞线前抢救。护栏③原来直接硬停提示"开新会话"，现在先做一次 build(aggressive=True)——把全部工具输出（含最近轮）折叠成存根重拼，语义主干原样保留，救得回来就继续跑。真超窗口是很糟糕的体验，但绝大多数时候只是工具输出肥，折一下就过去了。

还有一处修的是"误报"：以前丢弃消息和蒸馏折叠共用一条"上下文超限"提示，每轮正常降噪都吓用户一次。现在分成两条——丢弃=警告，折叠=信息，措辞明确写"保留结论，需要时模型可重新调用工具"。最后给系统提示词加了两条行为纪律：及时沉淀结论（每次工具调用后把关键发现写进自己的回复文本，因为工具原文会被折叠而结论永生）、搜索适可而止（同一主题最多搜 2-3 次，够决策就开始产出）。上下文工程和提示词工程在这里咬合：机制负责"挤掉噪声"，纪律负责"让被挤掉的确实是可以丢的"。

业界怎么做

行业术语叫 context engineering。Claude Code 的 auto-compact 是代表实现：接近窗口上限（约 92%）时让模型自己写一份摘要替换历史——相比"丢最早消息"的静默降级，丢弃会丢决策依据，摘要是保语义的有损压缩。长任务还会用"外部记忆文件"把中间状态卸载到磁盘（todo list、笔记），需要时再读回。另外注意 Claude Code / Codex 的压缩阈值看的是会话总长度，不区分轮次边界——单轮内烧到阈值照样压，这也呼应了我们的"轮内蒸馏"补刀。深度研究类产品（Perplexity 等）对搜索密集型任务的正规解法是 fan-out：每天行程派一个子 Agent 各烧各的上下文，主线只收 300 字结论。

历史蒸馏可以看作 auto-compact 的零成本降级版：不额外花一次 LLM 调用写摘要，只把最肥、最过期的工具转写折叠掉，保住语义主干。实践中这个"穷人版"已经能解决绝大部分多轮退化问题——真金白银的 LLM 摘要，留到上下文确实逼近上限时再上。

四、Provider 适配层：把厂商差异收敛在一层

实现

core/provider.py。选 OpenAI 兼容协议的原因：DeepSeek / 通义 / 智谱 / Ollama 都实现了同一套 chat.completions，切换模型只改配置不改代码。

这层代码里坑最深的是流式 tool_calls 拼接。流式模式下工具调用是分片增量下发的：第一片带 id 和 name，后续每片只有 arguments 的一小段字符串碎片：

buckets: dict[int, dict[str, str]] = {}
async for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.content:
        content_parts.append(delta.content)
        await on_token(delta.content)          # token 即时推给前端
    for piece in delta.tool_calls or []:
        slot = buckets.setdefault(piece.index, {"id": "", "name": "", "arguments": ""})
        if piece.id: slot["id"] += piece.id
        if fn.name:  slot["name"] += fn.name
        if fn.arguments: slot["arguments"] += fn.arguments   # ★ 按 index 累积
tool_calls = [ToolCall.parse(...) for idx, slot in sorted(buckets.items())]

不拼接就会拿到残缺 JSON。ToolCall.parse 再把拼好的字符串安全解析，失败保留 parse_error 供循环层"教模型改正"。

另外两个真实世界的坑：

• 代理字符消毒：终端粘贴/网页抓取混入的落单 surrogate 字符，会一路活到 SDK 的 json.dumps().encode("utf-8") 才炸，报错位置离源头十万八千里——所以在 provider 出口对 str content、多模态 text 段、tool_calls arguments 三处统一 clean_text；
• 视觉输入展开放在 provider 出口而非编排层：Message.to_openai() 只留 _image_paths 标记（types 层保持 IO-free 不碰磁盘），出口处 expand_images_in_payload 把图片文件展开成 [{type:"text"},{type:"image_url",url:"data:..."}] 多模态 parts，且必须在消毒之前（消毒只认 str content）。

业界怎么做

所有框架都有 Provider 抽象层（LangChain 的 ChatModel、Vercel AI SDK 的 provider package），本质相同。各家的私有格式差异（Anthropic 的 content block 数组、Gemini 的 functionDeclarations）靠这层抹平。不支持 function calling 的模型，退化路径是本项目注释里写的"扩展 ReAct 文本格式实现"——LangChain 的 create_react_agent 至今保留这种文本协议实现。

五、流式事件通道：从协程一路推到浏览器

实现

core/events.py 定义 11 种事件（token / turn_start / tool_start / tool_result / approval / plan / todo / info / answer / done / error），api/routes.py 用 SSE 推送。架构上是生产者-消费者解耦：

chat 跑在 asyncio.create_task 后台任务 → 事件塞 asyncio.Queue
SSE 生成器从队列取 → "data: {json}\n\n" 推给浏览器

好处：客户端断开不杀任务（Agent 继续跑完落库）；前端 sse.ts 手写按 \n\n 切帧解析（因为 EventSource 只支持 GET，对话要 POST JSON body）。

配置 REDIS_URL 后事件总线换 Redis Pub/Sub 实现，接口不变——多副本部署时 SSE 连接和 Agent 任务可能不在同一个进程。

业界怎么做

同 SSE/WebSocket。行业补强点：断线重连（Last-Event-ID 续传）、事件持久化回放、消息队列解耦。Vercel AI SDK 的 useChat 把这套 SSE 解析封装成了库——本项目前端手写的部分它替你做了。

六、Human-in-the-loop：审批与规划模式

实现

这是 Web 通道的核心难题：Agent 在协程里执行高危工具，浏览器里的人点"批准"是一个独立的 HTTP 请求——怎么让协程停下来等？

core/approval.py 的答案优雅到值得背下来：跨请求边界传递 asyncio.Future。

class ApprovalBroker:
    def __init__(self, timeout: float = 300.0) -> None:
        self._pending: dict[str, asyncio.Future[ApprovalResult]] = {}

    async def wait(self, request_id: str) -> ApprovalResult:
        """Agent 协程调这个挂起等待。"""
        future = asyncio.get_running_loop().create_future()
        self._pending[request_id] = future
        try:
            return await asyncio.wait_for(future, timeout=self.timeout)
        except TimeoutError:
            return ApprovalResult(False, "审批超时，已按拒绝处理。")   # ★ 超时=拒绝

    def resolve(self, request_id: str, approved: bool, ...):
        """浏览器 POST /api/approve 时调这个唤醒。"""
        future = self._pending.get(request_id)
        future.set_result(ApprovalResult(approved, feedback))

loop 发现 tool.dangerous → emit(APPROVAL, request_id) → await broker.wait(id)   # 挂起
浏览器渲染弹窗 → 用户点击 → POST /api/approve(request_id, approved)
broker.resolve → future.set_result → loop 继续

超时按拒绝处理——审批挂起占着一次 LLM 会话，静默放行才是危险选项。

Plan 模式是同一机制的协商版：submit_plan 工具刻意不标 dangerous。区别在于：

• "计划被拒"不该中断循环——反馈意见作为工具结果回喂，模型修订后重新提交，这是正常协商；
• "高危被拒"才中止循环（stop_reason="interrupted"）。

批准后计划变成任务清单（TaskBoard 整表覆盖式更新），注入 ctx.set_section("任务清单", ...) 驱动后续每轮执行。

业界怎么做

叫 permission mode / plan mode。Claude Code 是 allow/ask/deny 三级规则树 + 会话级授权记忆（“这次允许后同类不再问”）；Codex 有 sandbox mode + approval policy 组合。本质都是"异步请求-挂起-回填"，只是规则引擎的复杂度不同。

七、记忆与技能：文件即上下文

实现

记忆（memory/store.py）朴素到极致：两层 Markdown 文件（用户级 ~/.agent/MEMORY.md 跨项目 + 工作区级），会话启动时读进来 ctx.set_section("用户长期记忆", ...) 注入，空则不占预算。写记忆走工具且标 dangerous=True——记忆污染会持续拉低后续所有会话的质量，必须人工确认才允许写入。这条纪律特别容易被忽略。

技能（skills/loader.py）的核心是 渐进式披露：

def catalog(self) -> str:
    """system prompt 里只放目录：名字 + 一句话描述，几十 token。"""
    lines = [f"- `{s.name}`: {s.description}" for s in skills]
    return "当前可用技能（需要时仔细阅读其正文再动手）：\n" + "\n".join(lines)

def match(self, query: str) -> Skill | None:
    """命中才把正文完整注入——技能扩展到几百个也不撑爆上下文。"""

匹配里藏着本项目最大的中文坑：[\w\u4e00-\u9fff]+ 会把"帮我写一份周报"切成一整个 token，和关键词"周报"直接比对必然落空。解法是中文串滑窗切二元组：

def query_tokens(query: str) -> list[str]:
    for raw in re.findall(r"[\w\u4e00-\u9fff]+", query):
        if _CJK.search(raw) and len(raw) > 2:
            tokens.extend({raw[i:i+2] for i in range(len(raw) - 1)})  # 帮我/我写/…/周报
        else:
            tokens.append(raw)

业界怎么做

记忆侧，Claude Code 的 CLAUDE.md / memory bank 同样是"文件即记忆"；规模化玩家（如各类长期记忆产品 MemGPT/Letta）在文件之上加向量检索召回。技能侧，这就是 Anthropic 2025 年推的 Agent Skills 规范（SKILL.md + frontmatter，三层披露：metadata → 正文 → 捆绑资源），我的实现是它的最小可用版。词频匹配是"够用就好"，接口已抽象好，换 embedding 检索调用方无感。

八、MCP 连接器：JSON-RPC over stdio

实现

mcp/client.py 手写了一个最小 MCP 客户端：拉起子进程 → initialize 握手 → tools/list 拉清单 → 每个远程工具包成本地 Tool 注册进 registry——模型侧完全不感知工具是本地还是远程的。

难点是并发响应匹配：多个请求同时在飞、响应乱序回来。解法是 pending-Future 注册表：

# 每个请求带自增 id，挂 Future 进表
req_id = self._next_id; self._next_id += 1
fut = loop.create_future()
self._pending[req_id] = fut
await self._write(envelope)
resp = await asyncio.wait_for(fut, timeout=timeout)

# 后台 _read_loop 逐行读 stdout，按 id 派发
fut = self._pending.get(int(msg_id))
if fut is not None and not fut.done():
    fut.set_result(msg)

眼熟吗？和 ApprovalBroker 是同一个模式——pending 表 + Future 挂起/唤醒，这是本项目复用了两次的核心并发原语。

其他协议细节：newline-delimited 帧（注释特意提醒不是 LSP 的 Content-Length 帧）；initialize 超时给 60s（npx 冷启动慢）；写侧 asyncio.Lock 串行化防帧穿插；服务端反向请求（sampling）按协议必须回 -32601 拒绝，否则对方一直等；stderr 必须持续读掉，不然管道缓冲写满子进程卡死——每一条都是不踩一遍不会知道的坑。

业界怎么做

MCP 已是事实标准（OpenAI、Google 均已采纳），官方各语言有 SDK。行业实现和我手写的差异主要是：支持 resources / prompts / sampling 全部四类能力、OAuth 授权流程、streamable-http 传输。断连处理上"单个 server 挂了只降级跳过、不影响整体启动"是各实现的共识。

九、多智能体：spawn_subagent 就是个普通工具

实现

tools/subagent.py 的设计哲学：多智能体不需要改循环。父 Agent 调 spawn_subagent 工具 → 起一个独立上下文的 run_agent 子循环 → 只把最终结论作为一条 tool 输出回喂父上下文。

省 token 的本质就在这里：子 Agent 试错的中间过程（读过的文件、失败的探索）不进父窗口，父上下文只多一条干净的结果。

三个刻意的决策：

• role 参数用枚举（researcher/coder/reviewer）不用自由文本——收敛父模型的发挥空间，行为更稳定；
• task 参数描述里教父模型：“子 Agent 是失忆的同事，看不到你现在的对话，工单必须自包含”；
• 子 Agent 的审批回调接回父通道——防止"套一层子 Agent"绕过人工确认，安全边界不能被委托稀释。

业界怎么做

同 orchestrator-worker 模式（Claude Code 的 Task 工具即如此，Anthropic 的 multi-agent research 系统也是 lead agent + 并行 subagent fan-out）。进阶方向：子 Agent 间消息总线、A2A 协议（Google 2025.4）解决两个 Agent 的循环如何互相对话。

十、安全：沙箱与附件

实现

命令沙箱（sandbox/executor.py）四层防护：

# 1. 命令白名单（frozenset：ls/cat/python/git... 其余一律拒绝）
# 2. 危险模式黑名单（rm -rf /、mkfs、dd if=、fork bomb :(){ ）
# 3. 不使用 shell：argv 数组执行（exec 而非 shell=True）
proc = await asyncio.create_subprocess_exec(*argv, cwd=self.root, ...)
# 4. 固定 cwd + 超时 + 输出上限

第 3 层是关键决策：; rm -rf、反引号、$(...) 注入在 argv 执行模式下根上失效，代价是不支持管道——这一点写进工具 description 告诉模型。docstring 也诚实声明：“防误操作，不防恶意攻击者，真隔离请上容器。”

文件工具同样先过闸：所有路径 _resolve() 检查 resolved.relative_to(root)，越界即拒——防 Agent 误伤系统文件的第一道关。

附件安全（core/attachments.py）三条原则：

• 前端只传附件 id 不传路径，id 过正则白名单 ^[0-9a-f]{12}(__[\w.\-]{1,80})?$，路径由服务端拼接——杜绝 ../ 穿越；
• 图片做 magic bytes 验真，伪装成 .png 的文本会被降级为 other；
• 数据库只存路径引用不存 base64（1MB 图 → 1.4MB base64 会拖垮历史加载）。

这里还埋着我修过的最有教育意义的 bug：文件名清洗的白名单放行了空格，但 id 校验正则的 \w 不含空格——写入口和校验入口是同一份契约，字符集漂移导致"中文名带空格的附件能落盘、永远找不回"。教训：服务端生成又服务端校验的 id，必须配一条 save→resolve 往返测试兜底。

业界怎么做

行业更硬：Codex CLI 用 landlock + seccomp 做 OS 级沙箱，云 Agent 用容器/gVisor/microVM 隔离，网络出口白名单，密钥不进上下文。但"白名单 + 参数化执行 + 人工审批"三件套是所有实现的公共底座，且对"学习项目"而言，诚实声明防护边界比假装安全更有价值。

十一、还有两块外围：定时自动化与多模态

• 定时自动化（automations/scheduler.py）：RFC 5545 RRULE 解析 + asyncio 定时器，触发时开全新会话执行——定时任务之间不共享上下文，避免互相污染和无限膨胀。
• 多模态双向：生成走 images API、失败降级让 LLM 产 SVG（保证永远有产出物）；输入走 vision data URL，超 5MB 的图先压到长边 1568px + JPEG q80 再进 base64（视觉模型内部本来就下采样，几乎不丢语义）。

总结：那些不变的与善变的

从 2022 论文到今天的工程，ReAct 的形式变了三次（文本约定 → function calling → RL 内化），框架分化出 while 循环和状态机两派，但信息流骨架从未变过：

模型 = f(目标, 行动历史, 最新观察) → 下一个动作

所有工程差异只在循环之外，可以归纳为四个"转"：

1. 转得起 —— 上下文预算、token 计数降级、事件总线解耦；
2. 转得对 —— 工具 description 工程、结构化错误回喂、中文分词匹配；
3. 转得安全 —— 审批 Future、沙箱四层、附件 id 契约、记忆写入需确认；
4. 转得省 —— 子 Agent 隔离脏活、技能渐进披露、工具输出头尾截断。

如果你也想通过造轮子学 Agent，我的建议是：按依赖顺序读码，然后改坏它再修好。把 max_turns 调成 2 观察截停、删掉某个工具的 description 看模型选错工具、关掉 vision 开关发张图看降级路径——每个护栏都在测试里有对应用例，动手一次胜过读十遍。

项目地址：github.com/Tangugo/mini-workbuddy.（MIT，Python 3.12+ / FastAPI / React）。欢迎拍砖。

了解原理后我们把项目跑起来

git clone https://github.com/Tangugo/mini-workbuddy.git
cd mini-workbuddy
uv sync                      # 安装依赖（要求 Python ≥ 3.12）
cp .env.example .env         # 复制配置模板

编辑 .env，唯一必填项是一个 OpenAI 兼容端点的 API Key（DeepSeek / 通义 / 智谱 / Kimi / 本地 Ollama 均可，.env.example 里列了各家对应的三行配置）：

LLM_API_KEY=sk-xxxxxxxx
LLM_BASE_URL=https://api.deepseek.com
LLM_MODEL=deepseek-chat

CLI 模式——推荐先用它，终端里能逐行看清 ReAct 循环每一轮的思考、工具调用和结果回喂，是理解后文各模块的最快路径：

uv run python main.py check          # 配置自检
uv run python main.py                # 交互式对话
uv run python main.py ask "查一下 asyncio 是什么，写到 notes.md"

Web 模式——完整前端体验（流式渲染、工具过程、审批弹窗、附件上传），两个终端各起一端：

# 终端 1：后端（FastAPI + SSE，http://127.0.0.1:8000，/docs 有交互文档）
uv run python main.py serve

# 终端 2：前端（Vite 把 /api 代理到 :8000，打开 http://127.0.0.1:5173）
cd frontend && npm install && npm run dev

两点省心设计：PostgreSQL 和 Redis 都不是必须的，不配置自动降级为 SQLite / 进程内实现；跑测试验证环境用 uv run pytest（149 项，不需要真实 API Key，provider 全部 mock）。

启动服务

我配置的具备多模态的大模型：qwen3.8-flash

• 后端服务
  
• 前端服务
  

示例

查询天气

• 可以看到智能体是调用了web_search、web_fetch 网络搜索和网络抓取工具获取天气信息
• 已拿到实时数据，再补一下今日最高/最低温。智能体是有loop 循环的，先思考，然后调工具，接着观察，再思考循环

多模态–图片内容识别

• 原素材图
  
• 多模态识图
  

plan 规划模式

• 勾选规划模式
  
• 规划执行过程，可以看到执行了网络搜索、抓取、运行命令、写操作等工具
  
• 输出阿勒泰 7 天行程规划 md 文件
  
• 阿勒泰 7 天行程规划转 html 版本
  
  

文生图

写在最后

• 豆包工作、千问办公、workbuddy、codex、 Claude code 等智能体工具各有优势，没有最好只有是否适合/满足当前自己的需求，无需盲目追求
• AI 发展惊人的快，不是每个人都拥有研究基座大模型的学识和智力，但未来机会肯定属于善于使用 AI 工具的人
• 此次复刻 workbuddy不及其 1%，主要是想借此学习一个智能体核心功能模块

参考资料

• Yao et al., ReAct: Synergizing Reasoning and Acting in Language Models, 2022
• Anthropic, Introducing the Model Context Protocol (MCP), 2024
• Anthropic, Agent Skills 规范与 Claude Code 工程实践公开分享
• OpenAI, Function Calling 与 Tools API 文档
• Google, Agent2Agent (A2A) Protocol, 2025