使用多个模型构建 AI Agent：实用架构指南

大多数 AI Agent 依赖单一模型来处理执行的每个阶段。规划步骤、工具调用、数据提取、摘要生成以及错误恢复都通过同一个 LLM 运行。虽然这种方法对于初始原型来说很直接，但在生产环境中却会导致显著的效率低下。

需要深度推理的规划步骤并不需要与基础 JSON 提取步骤相同的模型。代码生成任务的要求与分类任务的要求也截然不同。使用像 Claude Fable 5 或 Claude Opus 4.8 这样的一流推理模型来格式化日期字符串，是一种昂贵的资源错配。

构建多模型 AI Agent 允许你将工作流的每个步骤路由到最适合该特定任务的模型。本指南探讨了如何设计、实现和管理这些多模型架构。

如果你是在 API 层而不是 Agent 编排层工作，请参阅 Agent-First API 设计 和 团队为何从直接模型 API 转向统一 AI API。当底层 API 表面足够稳定，无需重写编排代码即可切换模型时，多模型 Agent 的运行最为可靠。

:::info

关键要点

1. 模型匹配任务复杂度：使用小型、快速的模型进行路由、提取和格式化，将大型推理模型留给规划和复杂分析。
2. 标准化 Schema：在每次交接时实施严格的输出验证（如 Pydantic），以防止在不同模型提供商之间切换时出现契约偏移（contract drift）。
3. 为回退（Fallback）设计：构建自动回退路径，以处理速率限制、提供商中断或延迟峰值，而不中断 Agent 工作流。
4. 集中化遥测：跟踪延迟、输入/输出 token 数量以及每一步的成本，以持续优化路由逻辑。 :::

---

多模型 Agent 架构

多模型 Agent 架构根据复杂度、成本和延迟要求将任务分配给专门的模型。

用户请求
    │
    ▼
┌─────────────┐
│   路由器     │  ← 对任务复杂度进行分类
│  (快速模型)  │
└──────┬──────┘
       │
   ┌───┴───┐
   ▼       ▼
┌──────┐ ┌──────┐
│简单模型│ │复杂模型│
└──┬───┘ └──┬───┘
   │        │
   ▼        ▼
┌─────────────┐
│  聚合器     │  ← 合并结果
│  (快速模型)  │
└─────────────┘

核心架构由五个主要组件组成：

1. 路由器（Router）：一种快速、低成本的模型，根据复杂度和意图对传入的任务进行分类。
2. 模型池（Model Pool）：与不同任务类型（如推理、提取或代码生成）相匹配的模型集合。
3. 聚合器（Aggregator）：一种快速模型，将并行步骤的结果合并为最终响应。
4. 回退策略（Fallback Policy）：当首选模型失败、超时或遇到速率限制时，决定使用哪个模型的规则。
5. 遥测层（Telemetry Layer）：一种记录模型选择、延迟和每一步精确 token 成本的日志系统。

如果没有回退策略和遥测，多模型 Agent 可能会变得难以调试，且具有不可预测的延迟和成本概况。

---

使用 OpenAI SDK 实现

使用统一的 API 网关允许你通过单个 SDK 和 API 密钥访问来自不同提供商的模型。这简化了模型切换和路由。

以下示例演示了基本的路由实现。模型可用性和定价应在 TokenLab 模型目录 上进行验证。

from openai import OpenAI

client = OpenAI(
    api_key="sk-tokenlab-xxx",
    base_url="https://api.tokenlab.sh/v1"
)

# 包含成本和能力分级的模型池
MODELS = {
    "router": "deepseek-v4-flash",        # 快速分类
    "simple": "deepseek-v4-flash",        # 提取、格式化
    "reasoning": "claude-sonnet-5",       # 规划、分析
    "complex": "gpt-5.5",                 # 代码生成、复杂逻辑
    "budget": "deepseek-v4-flash",        # 批量处理
}

def route_task(task: str) -> str:
    """使用低成本模型对任务复杂度进行分类。"""
    response = client.chat.completions.create(
        model=MODELS["router"],
        messages=[
            {"role": "system", "content": """将此任务归类为以下类别之一：
- simple: 数据提取、格式化、翻译
- reasoning: 分析、规划、比较
- complex: 代码生成、多步问题解决
- budget: 批量处理、非关键任务
仅回复小写的类别名称。"""},
            {"role": "user", "content": task}
        ],
        max_tokens=10
    )
    category = response.choices[0].message.content.strip().lower()
    return MODELS.get(category, MODELS["simple"])

def execute_task(task: str, context: str = "") -> str:
    """将任务路由到选定模型并执行。"""
    model = route_task(task)
    messages = []
    if context:
        messages.append({"role": "system", "content": context})
    messages.append({"role": "user", "content": task})

    response = client.chat.completions.create(
        model=model,
        messages=messages
    )
    return response.choices[0].message.content

---

真实世界 Agent：代码审查流水线

为了了解构建多模型 AI Agent 的实际影响，请考虑一个旨在审查 Pull Request 的流水线。该工作流将审查分解为专门的步骤，而不是将整个代码差异（diff）发送给单个昂贵的模型。

def review_pr(diff: str) -> dict:
    """多模型 PR 审查流水线。"""

    # 第 1 步：使用快速、低成本模型对变更进行分类
    classification = client.chat.completions.create(
        model="deepseek-v4-flash",
        messages=[{
            "role": "user",
            "content": f"对这些代码变更进行分类：{diff[:2000]}\n"
                       "类别：bugfix, feature, refactor, docs, test"
        }],
        max_tokens=20
    ).choices[0].message.content

    # 第 2 步：使用强大的推理模型进行安全扫描
    security = client.chat.completions.create(
        model="claude-sonnet-5",
        messages=[{
            "role": "system",
            "content": "你是一名安全审查员。检查："
                       "SQL 注入、XSS、身份验证绕过、代码中的密钥、"
                       "不安全的反序列化。请指明具体的行号。"
        }, {
            "role": "user",
            "content": f"审查此 diff 的安全问题：\n{diff}"
        }]
    ).choices[0].message.content

    # 第 3 步：使用通用模型分析代码质量
    quality = client.chat.completions.create(
        model="gpt-5.5",
        messages=[{
            "role": "user",
            "content": f"审查代码质量：命名、结构、"
                       f"错误处理、测试覆盖率。\n{diff}"
        }]
    ).choices[0].message.content

    # 第 4 步：使用快速、低成本模型生成摘要
    summary = client.chat.completions.create(
        model="deepseek-v4-flash",
        messages=[{
            "role": "user",
            "content": f"用 3 个要点总结此 PR 审查：\n"
                       f"类型：{classification}\n"
                       f"安全：{security[:500]}\n"
                       f"质量：{quality[:500]}"
        }]
    ).choices[0].message.content

    return {
        "classification": classification,
        "security": security,
        "quality": quality,
        "summary": summary
    }

成本与效率优化

下表概述了此流水线的模型分配。确切定价因提供商和用量而异；请在 TokenLab 模型目录 上验证当前费率。

步骤	模型	输入 Token	角色 / 专业领域
1. 分类	DeepSeek V4 Flash	~2,100	快速分类，低成本路由
2. 安全	Claude Sonnet 5	~2,500	深度推理，安全分析
3. 质量	GPT-5.5	~2,500	高级代码质量和结构审查
4. 摘要	DeepSeek V4 Flash	~1,200	快速、低成本文本聚合

如果将所有四个步骤都通过像 Claude Sonnet 5 或 GPT-5.5 这样的旗舰推理模型运行，成本将显著增加。通过将简单任务路由到像 DeepSeek V4 Flash 这样的低成本模型，多模型流水线在减少总体 token 开支的同时，保留了关键安全分析步骤所需的深度推理能力。

---

按能力路由，而不仅仅是按价格

虽然降低成本是一个常见目标，但路由决策还应考虑特定的模型能力。稳健的路由策略会从四个关键维度评估模型：

• 推理深度：复杂逻辑、规划和多步推导。
• 上下文窗口：任务所需的背景信息或代码量。
• 工具使用可靠性：函数调用和结构化输出生成的准确性。
• 延迟敏感度：面向用户的应用程序的速度要求。

这些维度有助于建立明确的路由规则：

• 分解和规划任务路由到推理密集型模型。
• 数据提取和格式化任务路由到快速、低成本模型。
• 代码生成和语法分析路由到针对编码任务优化的模型。
• 全库分析任务路由到具有大上下文窗口的模型。

要使你的路由器符合这些要求，请查阅 编码模型比较 和 定价比较，将你的工作流步骤与当前模型基准相匹配。

---

LangChain 集成

你也可以在 LangChain 等编排框架内实现多模型路由。以下示例使用统一的 API 基础 URL 配置了不同的模型：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 使用不同的配置初始化模型
fast_model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key="sk-tokenlab-xxx",
    base_url="https://api.tokenlab.sh/v1"
)

reasoning_model = ChatOpenAI(
    model="claude-sonnet-5",
    api_key="sk-tokenlab-xxx",
    base_url="https://api.tokenlab.sh/v1"
)

# 定义专门的链
classify_chain = ChatPromptTemplate.from_template(
    "对该请求的意图进行分类：{input}"
) | fast_model

analyze_chain = ChatPromptTemplate.from_template(
    "对此问题进行详细分析：{input}"
) | reasoning_model

---

何时使用多模型 Agent

引入多个模型增加了架构复杂性。这种方法通常在以下情况下最有益：

• 任务需求多样：Agent 处理混合了简单任务（如分类或格式化）和复杂任务（如战略规划或代码生成）的工作。
• 高容量和高成本：每月的 API 支出很高，以至于优化能带来显著的节省。
• 专门的模型优势：工作流受益于特定提供商的优势，例如 Gemini 的上下文窗口、Claude 的编码能力或 GPT 的工具使用速度。
• 非对称延迟需求：工作流的某些部分必须立即返回结果，而其他后台步骤可以花费更长时间。

对于单一用途的 Agent 或简单的聊天界面，单一模型通常更容易维护。如果每个请求都需要相同级别的能力，那么路由的操作开销可能是不合理的。

---

常见失败模式

多模型架构引入了需要缓解的特定失败模式：

1. 过度设计的路由器

如果路由器提示词变得过于复杂，分类步骤本身可能会变得缓慢且昂贵。保持路由提示词简洁，分类类别宽泛。

2. 输出契约偏移

即使被指示返回 JSON，不同的模型格式化输出的方式也可能不同。一个模型可能返回原始 JSON，而另一个可能将其包装在 markdown 代码块中。为了防止下游解析器失败，请在每一步交接时使用 Pydantic 等验证库强制执行严格的 Schema。

3. 静默质量下降

如果回退策略在主要提供商中断期间将请求路由到低级模型，Agent 可能会在不报错的情况下返回质量较低的答案。实施明确的 速率限制策略 和警报系统有助于跟踪回退何时处于活动状态。

4. 碎片化的遥测

当模型使用分散在多个直接提供商 API 中时，汇总成本和性能指标变得困难。通过单个网关集中请求可以简化日志记录和成本跟踪。

---

最小评估循环

为了维护多模型 Agent，建立一个基本的评估循环来跟踪性能。你可以将每次运行的以下指标记录到数据库表中：

• 任务类别：路由器分配的分类。
• 选定模型：每一步选择的模型。
• 步骤延迟：完成每一步所花费的时间。
• Token 使用量：精确的输入和输出 token 数量。
• 回退状态：是否触发了回退模型。
• 用户反馈：最终输出是否成功的二进制指标。

分析这些数据有助于确定路由器是否选择了正确的模型，哪些步骤占用了大部分成本，以及回退模型是否保持了可接受的质量。

---

常见问题解答

如何处理不同模型之间的不同提示词格式？

不同的模型对不同的提示词结构响应最好。例如，一些模型在系统提示词下表现更好，而另一些则更喜欢嵌入在用户提示词中的指令。为了处理这个问题，将你的提示词抽象为根据目标模型进行调整的模板，而不是向池中的每个模型发送相同的原始字符串。

路由是否会给面向用户的应用程序增加太多延迟？

路由确实会为分类步骤引入少量延迟。你可以通过以下方式最小化这一点：为路由器使用高度优化、低延迟的模型，保持最大 token 限制较低（低于 10 个 token），或者当分类可以从用户的应用程序状态或入口点推断出来时并行化步骤。

切换模型时如何防止 JSON 解析错误？

为了防止解析错误，请使用模型提供商支持的结构化输出功能（如 JSON 模式或工具调用）。此外，在将所有模型输出传递给流水线的下一步之前，使用 Pydantic 或类似库在验证层中包装所有模型输出，以解析、验证和修复有效负载。

---

通过一个 API 访问每个模型：开始使用 TokenLab，通过单个 API 密钥访问超过 300 个模型。构建多模型 Agent，无需管理多个提供商账户，也无需为不同的 API 重写路由逻辑。