多模型适配架构
多模型适配架构
学习目标
读完本章后,你将能够:
- 设计模型提供商的统一抽象层
- 实现多种认证策略(API Key、OAuth、JWT、AWS SigV4)
- 配置传输层重试和故障转移策略
- 评估本地模型(Ollama/LM Studio)与云端模型的集成方案
前置知识
- HTTP/HTTPS 基础
- OAuth 2.0 和 JWT 基本概念
- 异步 HTTP 客户端
核心概念
1. 为什么需要多模型适配
生产级 Agent 不应锁定单一模型提供商:
- 成本优化:不同任务使用不同价位的模型(简单任务用便宜模型,复杂任务用强模型)
- 可用性:某个提供商故障时自动切换到备用提供商
- 隐私合规:敏感数据使用本地模型,避免上传云端
- 能力互补:不同模型在不同领域各有所长
2. ModelProvider Trait 抽象
统一抽象层是多模型适配的核心:
trait ModelProvider { fn info(&self) -> &ModelProviderInfo; fn capabilities(&self) -> ProviderCapabilities; async fn auth(&self) -> Option<CodexAuth>; async fn api_provider(&self) -> Result<Provider>; async fn runtime_base_url(&self) -> Result<Option<String>>; async fn api_auth(&self) -> Result<SharedAuthProvider>;}关键设计:
ModelProviderInfo:提供商的元数据(base_url、env_key、auth 模式)ProviderCapabilities:能力声明(支持的模型、功能)api_provider():返回实际可用的 API 客户端- 工厂函数
create_model_provider()根据配置名称路由到具体实现
3. 支持的提供商类型
| 提供商 | 协议 | 认证方式 | 典型场景 |
|---|---|---|---|
| OpenAI | Responses API | OAuth / API Key | 通用任务 |
| Amazon Bedrock | Responses API | AWS SigV4 | 企业 AWS 生态 |
| Ollama | OpenAI 兼容 | 无需认证 | 本地隐私敏感任务 |
| LM Studio | OpenAI 兼容 | 无需认证 | 本地开发测试 |
WireApi 协议收敛:所有提供商统一走 OpenAI Responses API 协议,通过适配层处理差异。
4. 认证策略
API Key
最简单的方式:静态 Authorization: Bearer <token>。
- 优势:实现简单、无状态
- 劣势:Token 泄露风险高、无法自动续期
OAuth 2.0 (PKCE)
适用于面向消费者的应用(如 ChatGPT 登录):
1. 生成 PKCE code_verifier + code_challenge2. 打开浏览器,用户授权3. 接收 authorization_code4. 交换 access_token + refresh_token5. Token 过期时,用 refresh_token 自动续期关键机制:
refresh_token自动续期:在 token 过期前主动刷新expected_account_id:防止账户切换导致的安全问题- 本地回调服务器:临时 HTTP 服务器接收授权码
JWT (Ed25519)
适用于服务间调用:
- 使用 Ed25519 非对称密钥对生成 JWT
kid(Key ID)标识使用的密钥- 无需与授权服务器交互,完全本地验证
AWS SigV4
适用于 AWS 生态:
- 对 HTTP 请求进行签名(Header + 时间戳 + HMAC)
- 签名前剥离 Bedrock Mantle 不保留的头部,避免签名失败
- 自动处理凭证轮换
设计模式详解
传输层重试策略
网络请求可能因临时故障失败,需要智能重试:
指数退避 + 抖动:
delay = base_delay * 2^(attempt - 1) * random(0.9, 1.1)- base_delay:初始退避时间(如 1 秒)
- 指数增长:每次重试等待时间翻倍
- 随机抖动:0.9-1.1x 随机因子,避免惊群效应
可重试状态码:
429:速率限制5xx:服务端错误408:请求超时
不可重试状态码:
400:请求格式错误401/403:认证错误(触发认证恢复而非简单重试)
认证恢复状态机
遇到 401/403 时,不是简单重试,而是进入恢复流程:
UnauthorizedRecovery { step: CheckRefreshToken / RefreshToken / Reauth / Fail, expected_account_id: Option<String>, mode: Automatic / Interactive,}恢复步骤:
- 检查是否有
refresh_token - 尝试用
refresh_token获取新的access_token - 如果刷新失败,进入交互式重新认证
- 验证
expected_account_id,防止账户切换
传输层抽象
统一不同传输方式的接口:
trait HttpTransport { async fn execute(&self, req: Request) -> Result<Response, TransportError>; async fn stream(&self, req: Request) -> Result<StreamResponse, TransportError>;}实现方式:
- HTTP/SSE:基于
reqwest的标准 HTTP 客户端 - WebSocket:
tokio-tungstenite,支持自定义 CA - WebRTC:SDP offer/answer 交换,P2P 媒体通道
通用处理:
- 请求/响应日志(OpenTelemetry 追踪)
- 自定义 CA 证书注入
- 长连接保活(流式响应不超时)
问题与规避
认证过期与自动刷新
问题:access_token 过期后,请求失败,用户体验中断。
对策:
- 预刷新:在 token 过期前(如提前 5 分钟)主动刷新
- 后刷新:请求失败后自动刷新并重试
- 持久化:刷新后的 token 立即写回磁盘,避免进程重启后失效
不同提供商的 API 差异
问题:虽然都声称兼容 OpenAI API,但实际存在细微差异(如参数名、错误格式、流式事件)。
对策:
- 适配层:在
ModelProvider实现中处理差异 - 最小公共子集:只使用最通用的 API 功能
- 测试矩阵:对每个提供商运行相同的测试用例
流式响应的超时处理
问题:流式响应(SSE/WebSocket)可能因网络问题无限挂起。
对策:
- 连接超时:建立连接的最大等待时间
- 读取超时:两次数据包之间的最大间隔
- 总超时:整个请求的最大持续时间(流式请求通常不设总超时)
- 心跳机制:定期发送 keep-alive 探测
Auth Profile 旋转与 Cooldown
当同一提供商配置了多个认证凭据时,需要智能的认证 profile 管理机制:
Profile 类型:
- API Key:
{ provider, key },简单但有泄露风险 - OAuth:
{ provider, access, refresh, expires, email? },支持自动续期
Profile ID 命名:
- 默认:
provider:default - OAuth 有邮箱:
provider:<email>(如google:user@gmail.com)
旋转顺序:
- 显式配置:
auth.order[provider](如果设置) - 配置中的 profiles:
auth.profiles按提供商过滤 - 存储的 profiles:
auth-profiles.json中的条目
排序规则(无显式顺序时):
- 主键:profile 类型(OAuth 优先于 API Key)
- 次键:
usageStats.lastUsed(同类型内按最久未使用优先) - 冷却/禁用 profile:移到末尾,按最近过期时间排序
Session Stickiness:
- 每个 session 固定一个选中的 profile,保持提供商缓存温暖
- 固定 profile 在以下情况前不会旋转:
- session 重置(
/new//reset) - 压缩完成(compaction count 增加)
- profile 进入冷却/禁用状态
- session 重置(
Cooldown(冷却期):
- 触发条件:auth/rate-limit 错误、看起来像 rate limiting 的超时
- 冷却期内的 profile 不会被选中
- 支持 model-scoped cooldown:同一提供商的不同模型可独立冷却
- 冷却原因识别:不仅限于
429,还包括提供商特定的限流消息(如Too many concurrent requests、ThrottlingException等)
故障转移两阶段:
Fallback Override 的精细回滚:
- 仅覆盖 session 的模型选择字段(
providerOverride、modelOverride、authProfileOverride) - 防止失败的 fallback 覆盖用户手动
/model更改或其他 session 变更 - fallback 失败后,仅当这些字段仍匹配失败候选时才回滚
设计取舍
统一协议 vs 适配层转换
| 方案 | 优势 | 代价 |
|---|---|---|
| 统一协议 | 代码简洁、所有提供商行为一致 | 受限于最小公共子集、部分高级功能无法使用 |
| 适配层转换 | 充分利用各提供商特性 | 代码复杂、测试覆盖难度大 |
推荐策略:核心流程使用统一协议,高级功能通过适配层选择性启用。
本地模型 vs 云端模型
| 维度 | 本地(Ollama/LM Studio) | 云端(OpenAI/Anthropic) |
|---|---|---|
| 隐私 | 数据不上传 | 数据离开本地 |
| 成本 | 硬件成本 | 按 token 计费 |
| 质量 | 依赖模型选择 | 通常更高 |
| 延迟 | 低(本地) | 高(网络) |
| 可靠性 | 依赖本地环境 | 高可用 |
推荐策略:敏感/离线任务使用本地模型;通用/高质量任务使用云端模型。支持动态切换。
强类型 trait vs 动态分发
| 方案 | 优势 | 代价 |
|---|---|---|
| 强类型 trait | 编译时检查、性能高 | 新增提供商需修改代码 |
| 动态分发 | 运行时扩展、插件化 | 运行时错误、性能稍低 |
推荐策略:核心提供商使用强类型 trait;第三方扩展使用动态分发(如插件系统)。
Bundled Provider 懒加载与动态发现
在运行时加载多模型 Provider 时,一个常见挑战是避免启动时加载所有 Provider 依赖。理想的做法是懒加载:仅当需要某个 Provider 时才 import 其包。
懒加载模式:
BUNDLED_PROVIDERS: Record<string, () => Promise<FactoryFn>> = { "@ai-sdk/openai": () => import("@ai-sdk/openai").then(m => m.createOpenAI), "@ai-sdk/anthropic": () => import("@ai-sdk/anthropic").then(m => m.createAnthropic), // ... 20+ providers}每个 Provider 工厂函数仅在首次需要时才通过动态 import() 加载。这避免了启动时加载数百兆的依赖包。
动态 Provider 发现:
对于未在 Bundled Providers 中注册的第三方 Provider(如社区开发的 ai-sdk-provider-xxx),系统可以在运行时通过 npm install 加载:
- 配置文件中指定
npm: "ai-sdk-provider-xxx" - 运行时检查该包是否已安装
- 未安装时自动
npm install(或bun add) - 通过
import()动态加载,查找以create开头的导出函数 - 调用工厂函数初始化 SDK
关键设计点:
localPath条件:如果npm字段以file://开头,则作为本地入口路径加载- 导出函数约定:自动查找模块中以
create开头的导出(如createProvider),无需手动注册 - 包安装路径隔离:通过 npm 缓存目录加载,不污染项目
node_modules
模型排序与智能推荐
当有多个可用模型时,系统需要决定默认使用哪个模型。一个实用的排序策略是:
priority = ["gpt-5", "claude-sonnet-4", "gemini-3-pro"]sort(models): 1. 按 priority 中匹配的优先级降序 2. 排除含 "latest" 的模型 ID(不稳定版本) 3. 按模型 ID 字典序降序(新模型优先)最近使用模型追踪:将用户最近使用的模型记录到 model.json 中,下次启动时优先使用最近使用的模型。这比硬编码的优先级更尊重用户偏好。
小模型路由(Small Model)
对于上下文压缩、摘要等不需要强推理能力的任务,系统自动路由到小模型以节省成本:
getSmallModel(providerID): priority = ["claude-haiku-4-5", "gemini-3-flash", "gpt-5-nano"] 返回第一个匹配 provider.models 的模型不同 Provider 的 priority 不同(如 GitHub Copilot 优先 gpt-5-mini,某项目自建 API 优先 gpt-5-nano)。
模型变体系统
同一模型可以生成多个变体(Variant),每个变体有不同的配置(如温度、系统提示、工具集)。变体允许用户在不切换模型的情况下快速尝试不同的行为模式。
变体通过配置合并实现:
model.variants = { "creative": { temperature: 0.9, ... }, "precise": { temperature: 0.1, ... }, "code": { system: "You are a coding assistant...", ... },}用户可以通过 /variant creative 或类似方式切换。变体合并时遵循:基础配置 → 用户覆盖 → 禁用过滤(disabled 标记的变体被排除)。
认证分级检测
Provider 的认证状态按以下优先级检测:
| 优先级 | 来源 | 示例 |
|---|---|---|
| 1 | 环境变量 | OPENAI_API_KEY, ANTHROPIC_API_KEY |
| 2 | 认证存储 | <agent> auth login openai 保存的 OAuth Token |
| 3 | 配置文件 | agent.json 中 provider.openai.options.apiKey |
认证分级确保用户有多种方式配置 Provider,无需全部通过配置文件。
链式策略模型路由
除了基于 Provider 的模型选择外,生产系统还需要根据任务特征动态选择同一 Provider 内的具体模型。一个实用的模式是链式策略路由(CompositeStrategy Chain):
策略链的执行规则:
- 顺序执行:策略按优先级依次评估
- 短路机制:任何策略做出决策后立即返回,不再评估后续策略
- 元数据记录:记录选中策略名称、延迟、推理理由
关键策略说明:
| 策略 | 触发条件 | 决策依据 |
|---|---|---|
| Fallback | 当前模型不可用 | 预定义的 fallback 模型列表 |
| Override | 用户通过 -m 等参数显式指定 | 用户输入 |
| ApprovalMode | yolo/verbose 等审批模式 | 宽松模式选快速模型,严格模式选强模型 |
| 本地分类器 | 启用了轻量分类器(如 Gemma) | 在本地运行,判断任务复杂度 |
| 通用分类器 | 基于任务特征 | 任务类型、代码规模、工具需求等 |
| 数值分类器 | 基于 token 阈值 | 预估 token 数超过阈值则选强模型 |
| Default | 以上均未匹配 | 配置的默认模型 |
模型粘滞性(Sequence Stickiness):
一旦选定模型,后续 turn 保持使用同一模型(称为 sequenceModel),直到用户显式切换或会话重置。这避免了每个 turn 重新路由导致的模型抖动和缓存浪费。
对于需要区域配置的 Provider(如 AWS Bedrock、Azure),模型 ID 可能需要添加区域前缀:
us-east-1 + "claude-sonnet-4-20250514" → "us.claude-sonnet-4-20250514"eu-west-1 + "claude-3-5-sonnet" → "eu.claude-3-5-sonnet"区域前缀规则因 Provider 和区域而异。一个实用的策略是维护区域 → 前缀的映射表,并在模型初始化时自动添加。
Python Protocol 作为模型抽象
在 Python 生态中,typing.Protocol + @runtime_checkable 可以实现类似 Rust trait 的接口抽象,而无需创建基类:
from typing import Protocol, runtime_checkable, TypeVar
T = TypeVar('T', bound=BaseModel)
@runtime_checkableclass BaseChatModel(Protocol): model: str
@property def provider(self) -> str: ...
async def ainvoke( self, messages: list[BaseMessage], output_format: type[T] | None = None, **kwargs: Any ) -> ChatInvokeCompletion[T] | ChatInvokeCompletion[str]: ...与基类方式相比的优势:
- 不需要继承关系,任何实现 Protocol 的类都可以作为模型使用
- 支持结构化输出(
output_format: type[T])在协议层声明 @runtime_checkable允许isinstance(model, BaseChatModel)运行时检查
代价:Protocol 不做运行时验证,只检查属性存在性,不检查方法签名正确性。
模型特定的自动优化
与其让所有模型使用相同的参数,不如根据模型特性自动调整:
| 优化维度 | 触发条件 | 调整内容 |
|---|---|---|
| 超时 | Gemini 模型 | 75s(3-pro 为 90s) |
| Groq 模型 | 30s(推理快但易限流) | |
| Claude/O3/DeepSeek | 90s(思考模式需要更长时间) | |
| 截图尺寸 | Claude Sonnet | 1400×850(模型最优视觉分辨率) |
| 视觉开关 | DeepSeek / Grok-3 | 自动禁用 use_vision(不支持图片输入) |
| 坐标点击 | Claude Sonnet/Opus, Gemini Pro | 启用像素坐标点击 |
| Flash 模式 | Browser Use 自研模型 | 自动启用,精简输出 schema |
实现策略:在 Agent 初始化时检查 llm.model 名称,应用预定义的优化规则。无需用户手动配置。
设计权衡:
- 优点:开箱即用,无需调参
- 代价:硬编码的模型名称匹配,新模型发布时需要更新规则
- 替代方案:让每个 Provider 声明自己的能力(
capabilities),Agent 根据能力而非名称决策
简单故障转移 LLM 机制
在复杂的认证旋转和 profile 管理之外,一种更简单的故障转移策略适用于多数场景:
主 LLM 请求 → 捕获 ModelRateLimitError / ModelProviderError ↓检查状态码:401, 402, 429, 500, 502, 503, 504 → 可重试 ↓检查是否有 fallback_llm 配置 → 有 ↓切换:self.llm = self._fallback_llm(永久切换,非单次重试) ↓使用 fallback 继续执行关键设计决策:
- 一旦切换到 fallback,整个运行剩余部分都使用 fallback(不切回)
- 不支持 fallback 的 fallback(只有两级)
- 401(认证失败)被视为可重试——因为 fallback 可能是不同 provider,有自己的密钥
为什么简单方案有效:对于大多数自动化场景,“主 provider 出问题 → 切到备用 provider” 的两级模型已经足够。复杂的 profile 旋转和 model-scoped cooldown 更适合多用户、多密钥的 SaaS 场景。
Token 成本优化策略
除了模型路由和故障转移外,实际生产中的成本优化还需要在多个维度同时入手:
| 优化维度 | 策略 | 成本影响 |
|---|---|---|
| 默认模型降级 | Sonnet 替代 Opus 作为日常编码 | ~60% 降低 |
| 子代理模型降级 | Haiku 用于文件读取/探索/测试运行 | ~80% 降低 |
| 思考 Token 控制 | MAX_THINKING_TOKENS=10000 替代默认 31999 | ~70% 降低隐藏成本 |
| 战略压缩 | 在逻辑断点手动 /compact 而非自动压缩 | 减少上下文 token 浪费 |
| 上下文监控 | 在上下文耗尽/范围蔓延时发出警告 | 避免无意义的 token 消耗 |
关键洞察:成本优化不是”选择最便宜的模型”,而是在每个环节选择够用即可的模型。高频低复杂度的操作(文件读取、简单查询)用最低成本模型,低频高复杂度的操作(架构规划、复杂调试)保留高成本模型。
参考来源
补充:Hermes Agent 的传输层插件架构
来源:Hermes Agent(NousResearch/hermes-agent)agent/transports/、providers/、agent/auxiliary_client.py,commit 2517917
传输层插件:数据路径抽象
Hermes 将多模型支持拆分为两个独立的关注点:
-
Transport 插件(
agent/transports/):只负责数据路径转换ProviderTransport抽象基类:convert_messages→convert_tools→build_kwargs→normalize_response- 每个传输插件仅处理格式转换和响应标准化
- 客户端生命周期、流式管理、凭证刷新由外部管理
- 自动发现:
_discover_transports()按需 import 传输模块
-
Model Provider 插件(
plugins/model-providers/):负责元数据和认证- 每个后端(openrouter、anthropic、gmi、deepseek、nvidia 等)一个插件目录
register_provider(ProviderProfile(...))在模块加载时注册- 懒发现:首次
get_provider_profile()时才扫描目录 - 用户插件可覆盖内置插件(last-writer-wins)
api_mode 调度
Hermes 通过 api_mode 字段决定传输路径:
| api_mode | 传输方式 | 适用场景 |
|---|---|---|
chat_completions | OpenAI 兼容 REST | 大多数提供商(~16 个) |
anthropic_messages | Anthropic Messages API | Claude 原生 API |
bedrock | AWS Bedrock | 企业 AWS 生态 |
codex_responses | OpenAI Codex Responses | Codex 运行时 |
codex_app_server | Codex 子进程 | 终端/文件操作在 Codex 内运行 |
chat_completions 传输层内置了对特殊提供商的处理(如 Gemini 的 thinkingConfig、Moonshot/Kimi 的严格 JSON Schema 要求),通过 ProviderProfile 模式替代了旧的 flag-based kwargs 组装。
辅助模型自动路由
agent/auxiliary_client.py 为辅助任务分配独立的小模型,避免消耗主模型配额:
| 辅助任务 | 默认模型 | 可配置 |
|---|---|---|
| Curator 技能审查 | 小模型 | 是 |
| Vision 分析 | 视觉模型 | 是 |
| Embedding | 嵌入模型 | 是 |
| 标题生成 | 小模型 | 是 |
| 会话搜索摘要 | 小模型 | 是 |
分辨率顺序:显式配置 → 运行时主模型推断 → 配置默认值。这确保辅助任务不会意外使用昂贵的主模型。
Nous Portal 统一工具网关
Hermes 的 Nous Portal 订阅覆盖多个工具后端:
- 搜索:Firecrawl
- 图像生成:FAL
- TTS:OpenAI
- 浏览器:Browser Use
一个命令 hermes setup --portal 通过 OAuth 登录并自动配置所有工具后端,无需单独收集 API key。
补充:LlamaIndex 的插件式 LLM 注册与惰性加载
来源:LlamaIndex llama-index-core v0.14.22,commit f027669
8 个抽象方法的统一接口
LlamaIndex 的 BaseLLM 定义了 8 个抽象方法,覆盖所有调用模式:
| 同步 | 异步 |
|---|---|
chat() / complete() | achat() / acomplete() |
stream_chat() / stream_complete() | astream_chat() / astream_complete() |
每个集成包(如 llama-index-llms-openai、llama-index-llms-anthropic)实现这 8 个方法,核心包不感知具体实现。
resolve_llm() 字符串解析
LlamaIndex 支持通过字符串自动解析和实例化 LLM:
from llama_index.core.llms import resolve_llm
llm = resolve_llm("openai") # 自动创建 OpenAI() 默认配置llm = resolve_llm("ollama/llama3") # 自动创建 Ollama(model="llama3")实现机制:RECOGNIZED_LLMS 字典维护已注册的 LLM 类映射。核心包通过 try/except ImportError 条件导入——只在安装了相应包时注册。核心包只预装了 MockLLM、OpenAI、AzureOpenAI,其余 300+ 模型通过独立集成包在各自的 loading.py 中注册。
Settings 惰性加载
Settings 是一个全局 dataclass 单例,所有字段延迟初始化:
Settings.llm # 首次访问时才调用 resolve_llm("default")陷阱:Settings 是全局单例,多租户场景下需要手动管理独立的 Settings 实例,否则不同请求的 LLM 配置会互相干扰。
补充:双层 LLM 路由策略
来源:TradingAgents(TauricResearch/TradingAgents)tradingagents/graph/trading_graph.py、tradingagents/default_config.py、tradingagents/llm_clients/factory.py、tradingagents/llm_clients/model_catalog.py,commit 61522e1
在生产级多 Agent 系统中,为不同角色分配不同强度的模型是控制成本同时保证质量的关键策略。TradingAgents 引入了两层路由设计:
第一层:按任务复杂度分层
| 层级 | 默认模型(OpenAI) | 承担角色 | 理由 |
|---|---|---|---|
| Quick Think | gpt-5.4-mini | 分析师、交易员、辩论者、反思器 | 这些角色的输入受限(特定数据源或固定立场),不需要深度推理 |
| Deep Think | gpt-5.4 | 研究经理、组合经理 | 需要综合多份报告做出最终决策,推理复杂度高 |
优势:相比全部使用强模型,双层路由可节省 40-60% 的 Token 成本,因为分析师阶段的 Token 量通常占整个流水线的 60% 以上。
陷阱:如果 quick_think_llm 太弱,分析师报告质量不足,deep_think_llm 再强也无法弥补——“garbage in, garbage out”。
第二层:厂商特定的推理强度配置
在选定 LLM 层级后,不同厂商提供了不同的”推理强度”控制参数:
| 厂商 | 配置参数 | 可选值 | 影响 |
|---|---|---|---|
| OpenAI | reasoning_effort | low / medium / high | 控制 o 系列模型的思考深度 |
thinking_level | minimal / low / medium / high | 控制 Gemini 的思考模式 | |
| Anthropic | effort | low / medium / high | 控制 Claude 的 extended thinking token 用量 |
TradingAgents 在 TradingAgentsGraph._get_provider_kwargs() 中根据当前 provider 动态注入对应参数:
def _get_provider_kwargs(self): kwargs = {} provider = self.config.get("llm_provider", "").lower() if provider == "google": kwargs["thinking_level"] = self.config.get("google_thinking_level") elif provider == "openai": kwargs["reasoning_effort"] = self.config.get("openai_reasoning_effort") elif provider == "anthropic": kwargs["effort"] = self.config.get("anthropic_effort") return kwargs设计亮点:
- 参数通过
create_llm_client()的**kwargs传递到具体客户端,无需每个客户端单独处理 - 当参数为
None时不注入,使用厂商默认值 - CLI 中用户选择推理强度后,写入 config 并统一线程
与链式策略路由的对比:
本文档前面介绍的”链式策略模型路由”关注单次请求选择哪个模型。双层 LLM 路由关注系统架构层面预设不同强度的模型槽位。两者互补:双层路由定义了”系统有几个模型槽位”,链式策略定义了”每个槽位具体用哪个模型”。
环境变量统一配置模式
TradingAgents 还引入了 TRADINGAGENTS_* 前缀的环境变量覆盖机制,允许在不修改代码的情况下调整 LLM 配置:
TRADINGAGENTS_LLM_PROVIDER=anthropicTRADINGAGENTS_DEEP_THINK_LLM=claude-opus-4-7TRADINGAGENTS_QUICK_THINK_LLM=claude-sonnet-4-6TRADINGAGENTS_MAX_DEBATE_ROUNDS=2类型推断基于 DEFAULT_CONFIG 中默认值的类型(bool / int / float / string),用户只需写字符串。这使得 Docker 部署、CI 环境、多租户场景下的配置管理变得简单。
与 LlamaIndex Settings 惰性加载的对比:
LlamaIndex 使用全局 Settings 单例 + resolve_llm() 字符串解析。TradingAgents 使用显式配置字典 + 环境变量覆盖。前者更”魔法”但方便原型开发,后者更明确但需要手动传递 config。生产系统中,显式配置字典更适合多实例部署(每个实例独立 config,不受全局单例影响)。
Category 级意图路由模型选择
在多模型 Agent 系统中,一种高效的路由模式是Category 级意图路由——通过任务意图分类(而非直接指定模型名)来选择模型。
核心思想:用户或主编排器声明「需要什么类型的工作」,系统自动映射到最优模型。
内置 Category 映射示例:
| Category | 默认模型 | 适用场景 |
|---|---|---|
visual-engineering | gemini-3.1-pro (high) | 前端、UI/UX、设计 |
ultrabrain | gpt-5.5 (xhigh) | 深度逻辑推理、架构决策 |
deep | gpt-5.5 (medium) | 自主研究和端到端执行 |
quick | gpt-5.4-mini | 单文件修改、错字修复 |
writing | k2p5 | 文档、技术写作 |
unspecified-high | claude-opus-4-7 (max) | 未分类的高难度任务 |
unspecified-low | claude-sonnet-4-6 | 未分类的低难度任务 |
Category 与 Skill 的解耦:
- Category:决定「用哪个模型」(model、variant、temperature)
- Skill:决定「用什么工具和知识」(MCP 服务器、系统指令、工作流)
两者组合通过 task(category, skill) 调用生成最优 Agent。
双重回退系统:
- model-fallback:在调用前根据 provider 健康状态主动选择模型链
- runtime-fallback:在运行时失败后,按配置的
fallback_models数组依次重试 - 支持对象级 fallback:每个 fallback 可指定不同的 variant、thinking 配置
优势:
- 用户无需了解模型细节,只需声明任务类型
- 新模型出现时只需更新 Category 映射,无需修改所有调用点
- 不同模型家族的能力差异对调用者透明
补充:Agno 的错误分类驱动精细化故障转移
来源:Agno
libs/agno/agno/models/fallback.py,2026-05-30
按错误类型分别路由
Agno 的 FallbackConfig 将故障转移分为三类,每类独立配置:
# 伪代码:FallbackConfigFallbackConfig( on_error=[Claude(id="claude-sonnet-4-20250514")], # 通用错误(5xx/网络) on_rate_limit=[OpenAIResponses(id="gpt-4o-mini")], # 限流错误(429/529) on_context_overflow=[Claude(id="claude-sonnet-4-20250514")], # 上下文溢出 callback=lambda primary, fallback, error: log_activation(...),)错误分类优先级
回退模型的查找优先级:
关键设计:不可重试的客户端错误(401/403/400)不使用 on_error 回退。这些是配置 bug,需要开发者直接看到真实错误信息,而不是被回退模型掩盖。
ModelProviderError.classify() 自动分类
如果错误初始没有被明确分类,ModelProviderError.classify() 会尝试自动分类:
- 检查响应状态码(429 → RateLimit,特定消息 → ContextOverflow)
- 检查错误消息中的关键词
- 如果分类失败,保持为通用错误,走 on_error 路径
Deepcopy 防止共享状态
FallbackConfig.resolve_models() 对每个回退模型执行 deepcopy,防止配置在多个 Agent 之间共享时被意外修改。这是多 Agent 系统中容易被忽略的陷阱。
补充:Oh-My-ClaudeCode 的三级模型路由与成本优化
来源:Oh-My-ClaudeCode(Yeachan-Heo/oh-my-claudecode)docs/ARCHITECTURE.md、skills/omc-reference/SKILL.md、agents/definitions.ts,commit ed7800dd,版本 v4.14.4
三级模型路由
OMC 使用三个模型层级,根据任务类型自动选择最优模型:
| 层级 | 模型 | 特点 | 成本 | 典型任务 |
|---|---|---|---|---|
| LOW | Haiku | 快速、便宜 | 低 | 代码查找、文档读取、简单信息提取 |
| MEDIUM | Sonnet | 性能与成本平衡 | 中 | 代码实现、调试、测试、标准 review |
| HIGH | Opus | 最高推理质量 | 高 | 架构设计、战略规划、深度分析、共识规划 |
默认 Agent-模型映射
| Agent | 默认模型 | 理由 |
|---|---|---|
explore | Haiku | 只需要读取和映射代码,不需要深度推理 |
writer | Haiku | 文档生成属于低复杂度任务 |
executor | Sonnet | 代码实现需要中等推理能力 |
debugger | Sonnet | 根因分析需要理解代码逻辑 |
verifier | Sonnet | 验证需要执行和断言检查 |
analyst | Opus | 需求分析需要发现隐藏约束 |
planner | Opus | 任务排序需要全局视角 |
architect | Opus | 系统设计需要深度推理 |
critic | Opus | 计划审查需要多角度分析 |
code-reviewer | Opus | 全面审查需要高质量判断 |
成本优化效果
通过三级路由,OMC 声称可节省 30-50% 的 Token 成本:
- 简单任务(如代码搜索)使用 Haiku 而非 Sonnet/Opus,成本降低 10×+
- 标准任务(如代码实现)使用 Sonnet 而非 Opus,成本降低 3-5×
- 复杂任务(如架构设计)才使用 Opus,确保质量
动态模型覆盖
OMC 允许通过配置覆盖默认模型映射:
{ "modelRouting": { "executor": { "default": "opus" }, // 提升 executor 到 Opus "explore": { "default": "sonnet" } // 提升 explore 到 Sonnet }}使用场景:
- 复杂项目中,简单的代码搜索也可能涉及复杂的模式匹配——提升 explore 到 Sonnet
- 关键模块的实现需要最高质量——提升 executor 到 Opus
陷阱
- 默认映射不是万能的:某些 Haiku 任务可能需要 Sonnet 的推理能力(如复杂的正则表达式分析)
- 成本估算偏差:Haiku 的快速响应可能产生错误结果,需要重新执行,反而增加总成本
- 模型降级策略缺失:当 Opus 不可用时(如速率限制),没有自动降级到 Sonnet 的机制
- 环境模型继承:通过
resolveInheritedModelFromEnv支持模型继承,但配置错误时可能静默使用错误模型
补充:DeepSeek 推理模型在多轮工具调用中的 reasoning_content 回传适配
来源:JeecgBoot v3.9.2,AIChatHandler.java(injectThinkingPlaceholderIfNeeded、mergeParams),commit 3f826c4
问题背景
DeepSeek 推理模型(如 deepseek-reasoner、deepseek-v4-flash)在多轮工具调用场景中有一个特殊要求:每次请求必须回传上一次响应的 reasoning_content 字段。如果历史消息中没有该字段,API 返回错误:
“The reasoning_content in the thinking mode must be passed back to the API.”
根因分析
langchain4j 的 sendThinking 参数仅在 AiMessage.thinking() 非空时才会注入 reasoning_content 字段。但历史消息持久化层(如 Redis 中的 MessageHistory)通常只保存文本内容,不保存 reasoning_content。从历史重建 AiMessage 时 thinking 为 null,导致校验失败。
解决方案:占位 thinking 注入
// 伪代码:为历史 AI 消息注入占位 thinkingprivate static List<ChatMessage> injectThinkingPlaceholderIfNeeded(messages, modelName) { if (!isDeepSeekThinkingModel(modelName)) return messages;
for msg in messages: if msg is AiMessage and msg.thinking() is empty: rebuilt = AiMessage.builder() .text(msg.text()) .thinking("...") // 占位:让 langchain4j 带上 reasoning_content 字段 .toolExecutionRequests(msg.toolExecutionRequests()) .build() result.add(rebuilt) else: result.add(msg) return result}为什么占位有效:"..." 占位让 langchain4j 的 isNullOrEmpty 校验通过,框架会在请求中注入 reasoning_content: "..." 字段,满足 DeepSeek API 的字段存在性要求。
模型识别策略
isDeepSeekThinkingModel(modelName): 1. 精确匹配已知推理模型列表 2. 关键字包含匹配:reasoner / v4-flash / v4-pro 3. 兼容带版本后缀的变体(如 deepseek-v4-flash-0428)设计取舍:精确匹配 + 关键字包含的双重策略,既保证已知模型的准确识别,又兼容未来新变体。
参数自动开启
对 DeepSeek 推理模型,mergeParams 自动开启:
returnThinking = true:解析响应中的 reasoning_content 到 AiMessage.thinkingsendThinking = true:将 AiMessage.thinking 以 reasoning_content 字段回传到下次请求noThinking = false:禁止关闭思考过程
通用模式提炼
任何推理模型(不仅 DeepSeek)在多轮对话中需要回传 thinking 的场景,都可以应用此模式:
- 识别哪些模型需要回传 thinking
- 为历史消息中的 AI 回复注入占位 thinking(若为空)
- 自动开启 returnThinking + sendThinking 参数