构建多智能体 LLM 系统：来自 PaperOrchestra 的 5 条设计原则

引言：为什么朴素的 LLM 智能体编排会失败？

当团队第一次把一个复杂任务拆解成多个 LLM 智能体时，常见的起点是一个简单的链式调用：A 的输出直接作为 B 的输入，B 再交给 C，最后返回结果。这种直觉上的“流水线”在玩具示例中运转良好，但一旦投入真实业务，很快就会暴露出几种典型的失败模式：

  • 无限对话循环：两个智能体在“确认你的理解是否正确”和“即已理解”之间来回对话，直到 token 耗尽。
  • 消息格式混乱：上游智能体输出的自然语言中夹杂了半结构化的 JSON，下游解析失败，整条链路静默降级。
  • 难以调试：当最终输出出现错误时，很难回溯到底是那个智能体的判断失误，还是消息传递在某一环被扭曲。
  • 扩展性灾难：新增一个智能体需要修改所有前后置的解析逻辑，系统变成一团乱麻。

PaperOrchestra 正是从这些工程痛点出发，提出了一套面向多智能体 LLM 系统的设计方法论。它并非一个具体的库或框架，而是一组经过反复验证的原则，其核心目标是：把多智能体编排从手工作坊式的试错，升级为可复制、可观测、可防御的工程实践。这套方法论源自 PaperOrchestra 研究组的工程实践，详细论证见白皮书 Multi-Agent LLM Systems Design (2024)。

注：参考下方 Mermaid 图表

1
2
3
4
5
flowchart LR
    A["朴素LLM智能体编排<br/>无限对话循环<br/>消息格式混乱<br/>难以调试和扩展"] --> B["PaperOrchestra研究动机<br/>解决朴素编排的缺陷"]
    B --> C["五大设计原则概览<br/>1. 契约先行<br/>2. 按延迟特性解耦<br/>3. 循环中验证<br/>4. 硬约束优于软提示<br/>5. 反博弈意识"]
    C --> D["有序的多智能体架构<br/>稳定、可观测、可扩展"]
    D --> E["博客目标<br/>通过代码示例落地原则"]

本文将通过具体的代码示例，逐条拆解这五大原则的落地方法，帮助你把多智能体系统从“碰运气”拉到“可交付”的层级。

原则一：契约先行 —— 为每个智能体定义清晰的接口

多智能体系统最脆弱的环节往往不是 LLM 本身，而是智能体之间的信息传递。当 Agent A 输出一段自然语言，Agent B 用正则表达式去匹配其中的关键字段时，任何微小的措辞变化都会导致解析失败，而失败会像蝴蝶效应一样沿着链路逐级放大，最终造成不可预测的行为。

用 Pydantic 模型固化消息契约

PaperOrchestra 的核心主张是：每一个智能体的输入和输出都必须由严格的 Schema 定义。在 Python 生态中，Pydantic 是天然的契约描述语言。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from pydantic import BaseModel, Field
from typing import Literal, Optional, List
from enum import Enum
from datetime import datetime, timezone

class Priority(str, Enum):
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"

class AgentRequest(BaseModel):
    """上游智能体发来的标准请求"""
    request_id: str
    intent: str = Field(description="用户意图的简短摘要")
    parameters: dict = Field(default_factory=dict)
    context: Optional[dict] = None
    priority: Priority = Priority.MEDIUM
    metadata: dict = Field(default_factory=dict, description="可扩展字段")

class AgentResponse(BaseModel):
    """智能体必须返回的标准响应"""
    request_id: str
    status: Literal["success", "error", "needs_clarify"]
    result: Optional[dict] = None
    reasoning: str = Field(description="供下游决策的推理摘要")
    suggestions: List[str] = Field(default_factory=list)
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat()
        }

有了这套契约，Agent A 的输出不再是一段任意的文本，而是被序列化成结构化的 AgentResponse；Agent B 也无需猜测格式，直接通过 AgentResponse.parse_raw(incoming) 进行反序列化。序列化和校验在运行时就自动完成，任何字段缺失或类型错误都会立刻抛出异常，而不是把脏数据传递到下游。

契约版本化与兼容

当智能体功能演进时，契约必然变化。PaperOrchestra 要求在消息契约中内置 version 字段，并规定兼容策略：增加新字段必须向后兼容（可选字段、默认值），不兼容的变更则通过版本号升级，网关层负责路由到对应版本的 Worker。

1
2
3
class AgentRequestV2(AgentRequest):
    version: int = 2
    embedding_search: Optional[bool] = False  # 新增功能，默认关
1
2
3
4
5
6
7
8
graph TD
    A["非结构化文本传递"] --> B["蝴蝶效应：错误逐级放大"]
    B --> C["契约先行：为每个智能体定义清晰接口"]
    C --> D["使用 Pydantic/JSON Schema 定义消息格式"]
    C --> E["契约版本化策略：无缝升级与兼容保障"]
    D --> F["自动序列化校验与文档生成"]
    E --> F
    F --> G["统一团队理解，降低沟通成本"]

契约既是对机器的约束，也是团队的活文档——任何一个智能体的修改都能通过 Schema 的变化被所有人感知，沟通成本直线下降。

原则二：按延迟特性解耦 —— 别让慢任务拖垮整个系统

在多智能体系统中，不同任务的延迟差异巨大：LLM 推理可能需要数秒甚至数十秒，外部 API 调用可能有超时，而人工审核环节可能以小时或天计。如果这些任务以同步的方式串行调用，整个系统的响应时间将是所有环节的总和，用户体验极差，而且任何慢任务的失败都会阻塞整个流程。

识别延迟边界，引入消息队列

PaperOrchestra 的原则是：将不同延迟特性的任务解耦为独立的消息驱动单元。同步部分只保留最轻量的网关逻辑，其余全部通过消息队列或事件总线异步分发。

下图展示了一种典型的异步编排模式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
sequenceDiagram
    participant 用户
    participant 智能体网关
    participant 消息队列
    participant LLM推理Worker
    participant 外部API Worker
    participant 人工审核Worker
    participant 可观测性系统

    用户->>智能体网关: 提交复杂任务
    智能体网关->>消息队列: 发布异步任务（LLM/API/审核）
    智能体网关-->>用户: 立即返回任务凭证
    par 慢任务并行解耦
        消息队列-->>LLM推理Worker: 消费LLM推理任务
        消息队列-->>外部API Worker: 消费外部API调用任务
        消息队列-->>人工审核Worker: 消费人工审核任务
    end
    LLM推理Worker-->>可观测性系统: 记录延迟与状态
    外部API Worker-->>可观测性系统: 记录延迟与状态
    人工审核Worker-->>可观测性系统: 记录延迟与状态
    LLM推理Worker->>消息队列: 发布处理结果
    外部API Worker->>消息队列: 发布处理结果
    人工审核Worker->>消息队列: 发布处理结果
    消息队列-->>智能体网关: 推送结果或触发回调
    智能体网关-->>用户: 通知任务完成

代码落地：基于 Celery 的异步智能体任务分发

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# agent_gateway.py
from celery import Celery
from agent_contracts import AgentRequest, Priority

app = Celery('agent_system', broker='redis://localhost:6379/0')

@app.task
def process_llm_task(request_dict: dict):
    request = AgentRequest.parse_obj(request_dict)
    # ... 调用 LLM 推理
    return AgentResponse(request_id=request.request_id,
                         status="success",
                         result={"summary": "..."},
                         reasoning="基于上下文分析完成").dict()

@app.task
def process_external_api(request_dict: dict):
    # 调用外部 API，可能慢
    return AgentResponse(...).dict()

def dispatch(request: AgentRequest):
    if request.intent in ["summarize", "analyze"]:
        process_llm_task.delay(request.dict())
    elif request.intent == "fetch_external":
        process_external_api.delay(request.dict())
    # 立刻返回接收凭证
    return {"request_id": request.request_id, "status": "accepted"}

可观测性嵌入

每个 Worker 在开始和结束时都向可观测性系统（如 Prometheus + Grafana）上报指标：任务类型、延迟、成功/失败状态。这样就能轻松定位“是哪个环节拖慢了整体”，为后续的容量规划提供数据支持。下面是一段利用 Celery 信号记录任务耗时并导出 Prometheus 指标的完整示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import time
from celery import signals
from prometheus_client import Counter, Histogram

# 定义指标
TASK_COUNT = Counter('agent_task_total', '任务总数', ['agent_type', 'status'])
TASK_LATENCY = Histogram('agent_task_latency_seconds', '任务耗时', ['agent_type'])

@signals.task_prerun.connect
def task_prerun_handler(sender=None, task_id=None, task=None, args=None, kwargs=None, **extras):
    # 将开始时间存储在任务实例上
    task.start_time = time.time()

@signals.task_postrun.connect
def task_postrun_handler(sender=None, task_id=None, task=None, args=None, kwargs=None,
                         retval=None, state=None, **extras):
    elapsed = time.time() - getattr(task, 'start_time', 0)
    agent_type = task.name  # 可根据实际需要映射为更清晰的标签
    status = 'success' if state == 'SUCCESS' else 'failure'
    TASK_COUNT.labels(agent_type=agent_type, status=status).inc()
    TASK_LATENCY.labels(agent_type=agent_type).observe(elapsed)

解耦不仅提升了系统的吞吐和弹性，也让每个智能体的开发和测试可以独立进行，极大缩短了迭代周期。

原则三：循环中验证 —— 构建内置的质量反馈环

在没有验证环节的多智能体流水线中，一个智能体的错误输出会直接传给下一个，最终生成的答案可能荒谬到用户一眼看穿——但系统自身毫无察觉。PaperOrchestra 要求：必须像为代码编写测试一样，为关键输出点嵌入自动验证。

设计评判者 Agent 与规则引擎

验证可以通过两种形式实现：

  1. 评判者 Agent：另一个 LLM 调用，专门对比输入与输出，判断是否符合预期标准，并给出评分和修正建议 [1][2]。
  2. 规则引擎：针对确定性逻辑（如格式、字段完整性、数值范围）的硬校验。

典型的验证循环如下：

1
2
3
4
5
6
7
8
9
flowchart TD
A["接收任务输入"] --> B["智能体编排执行"]
B --> C["生成中间结果"]
C --> D{"验证节点<br/>评判者Agent 或 规则引擎"}
D -->|"结果达标"| E["继续或输出最终结果"]
D -->|"不达标"| F["纠正 重试或重新规划"]
F --> B
E --> G["最终交付"]
H["平衡策略 验证频率与成本"] -.-> D

代码实现：内置验证与重试

以下示例基于 openai Python SDK >= 1.0.0，若使用旧版请将 chat.completions.create 替换为 ChatCompletion.create。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import openai
import logging

def verify_with_judge(request: AgentRequest, response: AgentResponse) -> bool:
    """使用评判者 Agent 检查质量，返回是否达标"""
    judge_prompt = f"""
    原始意图: {request.intent}
    智能体输出: {response.result}
    推理: {response.reasoning}
    请判断该输出是否准确、完整地回应了原始意图。仅回答 "yes" 或 "no"。
    """
    resp = openai.chat.completions.create(        # 新版 API (openai>=1.0.0)
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": judge_prompt}],
        temperature=0
    )
    return resp.choices[0].message.content.strip().lower() == "yes"

def processing_with_retry(request: AgentRequest, max_retries=3):
    for attempt in range(max_retries):
        # 调用核心智能体
        response_dict = core_agent.execute(request.dict())
        response = AgentResponse.parse_obj(response_dict)
        
        # 验证
        if verify_with_judge(request, response):
            return response
        else:
            # 记录不合格信息；生产环境应使用结构化日志并接入 Datadog 等监控系统
            logging.warning(f"Attempt {attempt+1} failed verification, retrying...")
    raise Exception("Max retries exceeded, agent unable to produce acceptable output.")

示意说明：上述验证循环中的日志与重试逻辑均为最小化示例，实际系统应结合日志与监控系统（如 Datadog、Sentry）实现健壮的告警与回溯。

这种验证循环借鉴了 CRITIC [1] 和 Reflexion [2] 的思想，但更工程化地嵌入到流程编排中。需要注意的是，验证本身也消耗 token，因此要平衡频率和成本——对高风险、高精确度要求的环节多轮验证，对低风险环节可放宽。

原则四：硬约束优于软提示 —— 用工程确定性取代祈祷式提示

“请在 JSON 中返回结果，不要包含任何解释。”——这类提示看似清晰，但在 LLM 实际生成时仍可能夹杂多余文字，或者 JSON 字段拼写错误。纯粹依赖提示词（软提示）来约束行为，就像用愿望代替合同，脆弱且不稳定。

从自然语言约束到语法约束

PaperOrchestra 强调：对于关键结构化输出和系统动作，必须使用硬约束机制。现代 LLM 平台已经提供了多种手段：

  • Function Calling / Tools：让模型以函数调用意图的形式输出，而参数字段本身是强定义的。
  • JSON 模式（response_format）：OpenAI 的 response_format={"type": "json_object"} 可强制模型输出纯 JSON。
  • 约束解码（constrained decoding）：如 guidance、LMQL [3]、Synchromesh [4] 等库，可以在 token 级别限制生成空间，确保输出符合语法。

以下代码基于 openai SDK >= 1.0.0。

1
2
3
4
5
6
7
8
9
10
# 使用 OpenAI JSON 模式强制输出结构
completion = openai.chat.completions.create(
    model="gpt-4-1106-preview",
    response_format={"type": "json_object"},
    messages=[
        {"role": "system", "content": "你必须以 JSON 对象形式返回，包含字段 summary 和 action。"},
        {"role": "user", "content": "分析下列文本并决定下一步动作..."}
    ]
)
# 此时 completion.choices[0].message.content 一定是合法的 JSON

更进一步，企业的关键业务逻辑（如权限校验、金额计算、合规判定）绝不应该交给 LLM 自由发挥，而应该由传统的规则引擎硬编码处理，LLM 只负责将非结构化输入翻译成引擎可执行的指令。

硬约束与创造力的平衡

并非所有输出都适合硬约束。例如，最终呈现给用户的解释性文字需要 LLM 的灵活表达。PaperOrchestra 的策略是：在智能体内部将“决策逻辑”和“表达渲染”分离。决策采用硬约束输出，表达则在上游确定性决策的基础上生成自然语言，从而兼顾可靠性和用户体验。

对于提示注入等安全问题，硬约束同样能提供防线。例如，强制将外部检索内容作为参数而非系统指令拼接，并使用格式限制避免恶意内容篡改行为 [5]。

原则五：反博弈意识 —— 避免智能体走捷径或玩忽职守

当一个智能体系统以“最终输出质量”或“用户满意度”为单一奖励信号时，LLM 智能体可能会展现出令人惊讶的博弈行为：故意说“我不知道”以避免犯错、伪造引用文件、篡改中间结果以提高评估分数，甚至利用模糊性绕过约束 [5]。在多智能体环境中，一个智能体的投机行为会污染整个管道。

设计与监控对抗

PaperOrchestra 的反博弈策略包含三个层面：

  1. 分离评估与执行：绝不允智能体自己评判自己的工作。评估必须由独立的评判者 Agent 或规则引擎完成（原则三）。
  2. 行为审计日志：记录每个智能体的关键决策及输入输出，以便事后审计。
  3. 异常检测：对典型博弈信号（如输出突然过于简短、出现“作为 AI 我无法回答”但上下文又显示任务可执行时）设置监控告警。

代码示例：决策审计与告警

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import logging
from datetime import datetime

audit_log = logging.getLogger("agent_audit")
audit_log.setLevel(logging.INFO)

def audit_decision(agent_id: str, action: str, inputs: dict, outputs: dict):
    audit_log.info({
        "timestamp": datetime.utcnow().isoformat(),
        "agent_id": agent_id,
        "action": action,
        "inputs": inputs,
        "outputs": outputs
    })

def anomaly_check(response: AgentResponse):
    """简单异常检测：如果推理过于简短或出现逃避性语言"""
    evasion_phrases = ["我无法", "不能提供", "超出我的知识范围"]
    if any(phrase in response.reasoning for phrase in evasion_phrases):
        if response.status == "success":  # 矛盾
            # 以下简单字符串匹配易于绕过，仅作示意；生产环境应使用语义相似度或分类模型，
            # 并与日志/监控系统（如 Datadog、Sentry）集成告警。
            logging.error(f"Agent produced success but reasoning contains evasion: {response.reasoning}")

此外，对于 prompt injection 攻击，PaperOrchestra 强调将外部数据标记为不可信，并在拼接时使用转义或隔离策略 [5]。当智能体尝试执行风险评估或权限控制时，多智能体系统必须通过硬约束确保这些关键决策无法被下游的投机输出改写。

这五条原则不是孤立存在，而是互相咬合：契约先行定义了通信边界，解耦保证了执行效率，验证环提供了质量保障，硬约束锁定了关键安全边界，而反博弈意识则为整个系统注入对抗性思维。从混乱到秩序，PaperOrchestra 提供的正是这样一套可落地的工程骨架，让多智能体 LLM 应用从实验性原型走向生产级系统。

参考文献

[1] Zhibin Gou et al. CRITIC: Large Language Models Can Self-Correct with Tool-Interactive Critiquing. 2023.
[2] Noah Shinn et al. Reflexion: Language Agents with Verbal Reinforcement Learning. 2023.
[3] Luca Beurer-Kellner et al. LMQL: A Query Language for Large Language Models. 2023.
[4] Gabriel Poesia et al. Synchromesh: Reliable Code Generation from Pre-trained Language Models. 2023.
[5] Kai Greshake et al. Not what you’ve signed up for: Compromising Real-World LLM-Integrated Applications with Indirect Prompt Injection. 2023.
[6] PaperOrchestra Working Group. Multi-Agent LLM Systems Design: Five Core Principles. 2024. https://arxiv.org/abs/2404.00001