从 35 行循环到生产级 Agent：7 个原语搭出可靠的多智能体框架

原文来源：Data For Science — 把基本的 LLM 调用循环升级为生产级 Agent 框架：类型化工具、DAG 并行调度、分层记忆、验证层级、角色分离、多维预算与追踪器七大原语逐一拆解。

上一篇讲"基本 Harness"的循环是对的，但太天真了。一个优秀飞行员能打赢狗斗，但没人靠单机打空战——真实作战要任务规划师决定出击顺序、编队并行执行独立任务、燃油预算和返航呼叫在油箱耗尽前强制返航、飞行记录仪让每次任务可复盘、战后评估判断任务是否真的成功。这些都不替代飞行员，而是用结构把整个系统包裹起来，让它更快、更安全、更可调试、更可度量。

Claude Code、Devin、Cursor 这些生产级 Agent 对基本循环做的事一模一样。这篇文章的目标，是把基本 Harness 的每一块都升级到生产形态，同时不把任何机制藏在框架后面。贯穿全程的问题只有一个：

如何把单次 LLM 调用变成一个可靠系统——能规划、能执行、能恢复、能证明自己做对了？

答案是组合。我们构建小而可测试的原语：类型化工具、计划 DAG、分层记忆、验证层级、预算和追踪器，再用一个刻意保持薄的编排器把它们串起来。每个原语的存在，都因为天真 Agent 会以特定、可预测的方式失败：LLM 会编造非法工具参数，所以加 Pydantic 校验的类型化工具；所有东西串行执行，所以加依赖图和并行执行；上下文窗口被垃圾填满，所以加检索预算下的多层记忆；坏输出静默传播，所以加验证层级；一个提示词想干所有事，所以拆成 Planner、Worker、Critic 三个角色；成本失控，所以加多维预算和优雅降级。

贯穿全文的例子

全文用一个城市对比 Agent 做例子：给定城市列表，生成一份对比报告，比较人口、时区和每座城市的短篇叙述。

任务看起来简单得有点侮辱智商，但选它是精心设计的：每个城市属性查询互相独立，三座城市自然分解成 9 个可以同时跑的工具调用；最终报告依赖所有查询完成，远超扁平步骤列表；可以编程检查每座要求的城市是否真的出现在报告里；工具成本差异巨大——人口和时区查询是内存字典读取，而每城摘要和最终聚合都要调 LLM，给预算管理提供了真实的压力。

为了可复现，查询工具读一个模拟字典 CITY_FACTS，笔记本无需网络即可完整复现。LLM 部分可以跑真实 Anthropic 模型，也可以跑确定性 mock——这引出第一个原语。

原语一：可插拔的大脑

每个组件最终都要调 LLM：规划器、摘要器、聚合器、评论家。如果这些调用硬编码某个 SDK，整个 Harness 就不可测试、被厂商锁定。

所以先定义一个基类，抽象各种 LLM 调用 API 的细节：

class LLMProvider:
    """Shared interface. Subclass to plug in a different backend."""
 
    def complete(self, system: str, user: str, role: str = "default") -> str:
        raise NotImplementedError
 
    async def acomplete(self, system: str, user: str, role: str = "default") -> str:
        # Wrap sync call in a thread; works for any SDK.
        return await asyncio.to_thread(self.complete, system, user, role)

同时实现一个 MockProvider 用于测试和调试，返回确定性、感知角色的响应：被要求规划时给出标准计划，被要求摘要时给出模板化的一行摘要，被要求判断时给出基于规则的通过/失败结论。这样开发时能区分"我的编排逻辑错了"和"模型规划得不好"——这也是本文所有实验在任何机器上都能复现的原因。

原语二：类型化工具

基本 Harness 里手写校验工具参数，很快崩盘：每个新工具重复校验逻辑，LLM 永远看不到正式 schema 只能猜参数形状，产生的错误是模型无法自我修正的临时字符串。

升级方案是：把每个工具的参数声明为 Pydantic 模型，一个定义驱动一切：

@dataclass
class TypedTool:
    name: str
    description: str
    args_model: type[BaseModel]     # Pydantic model defining the arg schema
    fn: Callable[..., Any]
    cost_hint: float = 0.0          # relative cost for budget accounting
 
    def schema(self) -> dict:
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": self.args_model.model_json_schema(),
        }
 
    def run(self, raw_args: dict) -> Any:
        args, err = self.validate_args(raw_args)
        if err is not None:
            raise ValueError(err)
        return self.fn(**args.model_dump())

这一举多得：运行时校验、生成 Anthropic/OpenAI 工具调用 API 需要的 JSON Schema、通过 Field(description=...) 给规划器提供文档，还借 cost_hint 挂钩成本核算。执行前失败能避免昂贵的工具调用和潜在副作用——坏计划应该在校验层快速失败，而不是深埋在数据库查询里。

工具注册表放四个工具、三个成本档：人口和时区查询几乎免费（cost_hint=0.1），summarize_city() 每城一次 LLM 调用（cost_hint=1.0），aggregate_report() 是生成最终 Markdown 的重 token 合成调用（cost_hint=2.0）。注意后两个工具内部调用 LLM——LLM 和其他工具没什么不同，Worker 看到统一的工具接口，但某些工具只是子提示词的包装，可以独立缓存、限流或替换内部模型。

原语三：计划是一张图

基本 Harness 每轮执行一个动作。任务严格串行时没问题，但我们的任务有 9 个独立查询喂给一个聚合。while 循环只能一个个跑。**有向无环图（DAG）**显式表达依赖关系，让执行器并发跑所有就绪节点。

于是不再让 LLM 一次要一个动作，而是让 Planner 一次性输出整张图。LLM 声明结构后才执行。规划器也是 LLM，它同样会幻觉结构：依赖不存在的节点 ID、永远无法完成的循环依赖。所以拿到计划的第一件事是验证，避免在坏计划上浪费 token。

ready_nodes() 是调度器的核心：任何时刻返回依赖全部满足的节点集合。三城任务下规划器输出十个节点：九个空依赖的查询节点（全部可并行），一个依赖全部九个的 aggregate_report 终结点。

执行器是层级同步的 DAG 遍历器：

MAX_CONCURRENT = 5  # cap concurrent tool/LLM calls
 
async def execute_dag(dag, tools, on_step=None):
    semaphore = asyncio.Semaphore(MAX_CONCURRENT)
 
    async def run_node(node):
        node.status = NodeStatus.RUNNING
        async with semaphore:
            try:
                node.result = await asyncio.to_thread(tools[node.tool].run, node.args)
                node.status = NodeStatus.DONE
            except Exception as exc:
                node.error = f"{type(exc).__name__}: {exc}"
                node.status = NodeStatus.FAILED
 
    while not dag.is_done():
        ready = dag.ready_nodes()
        if not ready:
            break  # remaining nodes depend on FAILED ancestors
        await asyncio.gather(*(run_node(n) for n in ready))

两个小决定扛起了大部分重量。第一，asyncio.to_thread 把同步工具函数跑在线程池里，永远不用把工具重写成 async def，也不用让 Harness 耦合异步原生 SDK。第二，信号量限制并发——没有它，50 个节点的计划会同时发起 50 个 LLM 调用，立刻撞上速率限制或成本飙升。

这刻意不是带工作窃取和优先级队列的完整动态调度器。对 Agent 工作负载（每个节点是耗时几百毫秒到几秒的 API 调用），层级同步并行已经拿到大部分收益：串行墙钟时间约等于各查询延迟之和；并行约等于最大值加聚合步骤。

原语四：记住该记的东西

天真 Agent 把一切倒进提示词：完整聊天历史、每个工具输出、每个先前任务。这双倍失败——为用不上的 token 付钱，而且无关文本稀释目标时模型性能可测地下降。

生产 Agent 用分层记忆，灵感粗略来自认知科学。工作记忆是始终在上下文中的草稿本：当前目标、计划摘要、最近几个结果。情景记忆（episodic）存过去运行的结果，当过去任务与当前相似时检索。语义记忆存背景事实，同样方式检索但不绑定特定运行。

绝不注入一切；按与当前目标的相似度拉取 top-k 记忆，再在硬字符预算下组装上下文：

def build_context(working: WorkingMemory, store: MemoryStore,
                  budget_chars: int = 4000) -> str:
    pieces = [working.to_prompt()]
    used = len(pieces[0])
 
    # Episodic first (past similar tasks), then semantic (facts)
    for kind in ("episodic", "semantic"):
        for m in store.retrieve(working.goal, k=3, kind=kind):
            snippet = f"[{kind}] {m.content}"
            if used + len(snippet) + 1 > budget_chars:
                return "\n".join(pieces) + "\n(...truncated at budget...)"
            pieces.append(snippet)
            used += len(snippet) + 1
 
    return "\n".join(pieces)

情景记忆优先于语义记忆，因为过去相似任务上的错误通常比通用事实更可操作；预算耗尽时截断是显式的而非静默的。上下文应该被主动组装，而不是被动积累。

相似度函数支持两个后端：Jaccard 相似度零成本但无法处理转述（"法国的著名地标"和"巴黎以埃菲尔铁塔闻名"几乎没有共同词）；用 all-MiniLM-L6-v2 生成的 384 维真实句子嵌入能把转述映射到邻近向量。MemoryStore 优先试嵌入，模型不可用时用 Jaccard 兜底。

原语五：信任，但验证

Agent 产出流畅、自信、但错误的输出。没有验证，静默丢了一座城市的报告会发给用户，回归直到有人碰巧读到输出才被发现。但并非所有检查成本相同，所以把它们排成层级：几乎免费的确定性结构检查，加上花真金白银的 LLM 评判主观质量。规则是永远先跑便宜档，只有幸存者才升级。

def verify_report(report, goal, required_cities, provider) -> Verdict:
    det = deterministic_check_report(report, required_cities)
    if not det.passed:
        return det                # Cheap tier caught it — don't bother the LLM
    # Deterministic passed → escalate to LLM judge for subjective quality
    return llm_judge_report(report, goal, provider)

喂一份刻意不完整的报告（只写了巴黎，要求三座城市），它会在确定性层失败，reason="Missing cities: ['Tokyo', 'New York']"——评判一个 token 都没花，reason 字符串可操作到重规划步骤（或人类）能精确看到哪里错了。这背后是大多数生产 eval 管线的稳健模式：便宜过滤器在前，昂贵评判只给幸存者。层级之外同等重要的是职责分离：Worker 生产、Critic 评估，生成器永远不给自己打分。

原语六：Planner、Worker、Critic 三人组

一个同时规划、执行、摘要、自我批评的提示词容易混淆目标（规划约束渗进写作风格），也无法隔离"规划部分"，测试和替换都难。

拆成窄角色，各自短系统提示词和单一契约。Planner 接收目标加工具 schema，返回 DAG JSON，运行前验证。Worker 接收 DAG 直接执行。Critic 接收目标和成品报告，返回裁决。Planner 的系统提示词直接拼接实时工具目录，只能引用真实存在的工具：

PLANNER_SYSTEM = """You are a Planner agent. Given a GOAL, produce a dependency graph
of tool calls that will satisfy it.
 
Output ONLY JSON in this shape (no prose, no markdown):
 
{"nodes": [{"id": "...", "tool": "...", "args": {...}, "deps": [...]}, ...]}
 
Nodes may run in parallel if their `deps` are empty or already satisfied.
The final aggregate_report node must depend on all upstream fetch/summary nodes.
Prefer id "aggregate" for that capstone node (any unique id is acceptable).
Available tools (name and schema):
 
<<TOOLS_JSON>>
"""

三个角色走同一个 LLMProvider.complete(..., role=...) 接口，换规划器模型或 mock 评论家都是一行改动。

原语七：油量表与故障模式

基本 Harness 只有一个 max_steps 计数器，掩盖了真实约束：可能还有步骤但没 token 了；token 预算内但工具调用被限流；挂起的网络调用消耗墙钟时间却不增加任何计数器。BudgetMulti 同时跟踪 token、工具调用、墙钟时间和估算美元，任一维度耗尽即停止。最有用的输出是一个标量：

def pressure(self) -> float:
    return max(
        self.tokens_used / self.max_tokens,
        self.tool_calls_used / self.max_tool_calls,
        self.elapsed() / self.max_wall_seconds,
        self.cost_usd / self.max_cost_usd,
    )

Pressure 是所有维度的最大利用率——你被最先耗尽的资源限制，和真实计费一模一样。这个数驱动优雅降级：低于 0.7 全管线运行（含 LLM 评判）；高于 0.9 编排器跳过昂贵的评论家，只回退到确定性检查；到 1.0 运行停止并保留部分结果。生产 Agent 用同类信号切换到更便宜模型、降低检索深度或请求用户确认。

运维理智的另一半是认识到不是每个错误都值得同样对待。限流或超时是瞬态的，指数退避（加抖动，避免 Agent 集群同步重试）后重试。校验错误是工具误用，把结构化错误喂回 LLM 让它自己纠正参数。未知实体是信息缺失，重试反而有害——重试幻觉的城市名每次都会以同样方式失败，正确做法是去掉它重新规划。策略违规是致命的，立即停止。一个小 classify_error() 函数把错误字符串映射到这四个类别，恢复策略跟着类别走，而不是盲目重试。

飞行记录仪

Agent 失败时，需要一个结构化事件的追加写日志，能回答：以什么顺序发生了什么、每步花了多久、哪个角色消耗了 token、出问题前预算压力是否在上升。Tracer 里每个事件捕获身份（step ID 和把 Worker 步骤链回父计划的 parent ID）、语义（角色和动作）、经济学（延迟、token、成本、写入时的预算压力快照），Critic 事件还有裁决。schema 扁平而无聊：可 JSON 序列化的字典列表，能 dump 成文件、发给 OpenTelemetry 或 LangSmith、或直接 matplotlib 绘图。不需要专有格式就能获得真实可观测性，只需要一个够用的 schema。

按角色着色的每步延迟立刻显示 summarize_city() 和 aggregate_report() 主导墙钟时间而查询平坦——秒级消耗都在 LLM 调用上，这正是并发跑它们重要的原因。预算压力随时间单调上升，聚合器处陡增；如果它越过 0.9 降级阈值，trace 本身就解释了为什么 LLM 评判被跳过。

全部串起来

原语就位后，Orchestrator 变得几乎无聊：从记忆构建上下文 → 让 Planner 出 DAG → 把 DAG 交给 Worker（每次工具调用记录 trace 事件并扣预算）→ 检查失败 → 用压力感知降级验证结果 → 把结果存进情景记忆 → 返回打包报告、裁决、DAG、trace 和预算的 RunResult。

它新增的唯一行为是重规划循环。执行失败且错误归类为信息缺失时，把失败上下文送回 Planner 而不是盲目重试，最多 max_replans 次：

for attempt in range(self._max_replans + 1):
    dag = self.planner.plan(goal) if attempt == 0 else self.planner.replan(goal, dag)
    await self.worker.execute(dag, on_step=on_step)
 
    if dag.any_failed():
        failed = (n for n in dag.nodes.values() if n.status == NodeStatus.FAILED)
        if any(classify_error(n.error or "") == ErrorClass.MISSING_INFO for n in failed) \
                and attempt < self._max_replans and budget.has_room():
            continue          # informed re-plan, not a blind retry
        return RunResult(..., status="failed_execute")
    break                     # success
 
# Pressure-aware degradation: skip the LLM judge if budget is tight
if budget.pressure() > 0.9:
    verdict = deterministic_check_report(report, required_cities)
else:
    verdict = self.critic.judge(report, goal, required_cities)

作者踩过一个很真实的坑：mock 规划器总把终结点命名为 aggregate，早期版本编排器就按这个 id 查找。换成真实模型后这个假设悄悄崩了——真实规划器常镜像工具名 aggregate_report 或自造 id，产出结构有效但 mock 没教过我们期待的计划。修复有两半：DAG 现在按工具而不是 id 解析终结点（dag.aggregate_node() 返回唯一的 aggregate_report 节点，拒绝含多个的规划），规划器提示词也加上了显式终结点指令。永远别轻信 LLM，连节点名都不行。

还缺什么

从基本 Harness 到高级 Harness，覆盖了构建成功自定义 Harness 需要的大部分概念：~35 行循环升级为七个生产形态原语——一个 schema 同时给校验和 LLM 内省的类型化工具、不改任何工具就买到并行的 DAG 执行器、硬预算下组装相关上下文的分层记忆、只在通过便宜检查的输出上花钱的验证层级、可独立测试和替换的窄角色、优雅降级而非崩溃的多维预算、把运行变成可比实验而非轶事的追踪器。

这些部件彼此不依赖但能组合：给注册表加工具，规划器自动看到它的 schema；收紧 verify_report()，未来每次运行都面对新标准；换整个 LLM 后端，Harness 原样重跑。可组合性是 PoC 和可扩展 Harness 的区别，也是编排器保持薄的原因。

注意事项：这里的记忆是进程内的，生产系统会把嵌入持久化到 Chroma、Weaviate 或 pgvector；工具输出被当作指令信任，而生产系统必须把它们当数据沙箱化以防提示注入；不可逆操作应要求人工批准；token 记账用字符数估算，真实系统从 SDK 读 usage 元数据。每一个都是再往上一层组合的事。

还有一个刻意的省略：一次成功演示只证明 Harness 能工作，不能证明它在大多数情况下真的工作——那是 eval harness 的工作，作者计划在后续文章中展开。配套代码在 GitHub - DataForScience/LLMs。