名称： smart-router
描述： >
具备语义领域评分、上下文溢出保护与安全信息脱敏能力的专家感知模型路由系统。通过加权专家评分（基于2026年2月基准测试）自动选择最优AI模型。支持Claude、GPT、Gemini、Grok，提供自动故障转移链、人工介入（HITL）关卡及成本优化功能。
作者： c0nSpIc0uS7uRk3r
版本： 2.1.0
许可证： MIT
元数据：
openclaw:
requires:
bins: ["python3"]
env: ["ANTHROPIC_API_KEY"]
optional_env: ["GOOGLE_API_KEY", "OPENAI_API_KEY", "XAI_API_KEY"]
features:
- 语义领域检测
- 加权专家评分（0-100分）
- 基于风险的强制路由
- 上下文溢出保护（>150K → Gemini）
- 安全凭证脱敏
- 带持久化状态的熔断器
- 低置信度路由的人工介入（HITL）关卡
benchmarks:
source: "2026年2月 MLOC 分析报告"
models:
- "Claude Opus 4.5: SWE-bench 80.9%"
- "GPT-5.2: AIME 100%, 控制流错误率 22 errors/MLOC"
- "Gemini 3 Pro: 并发问题 69 issues/MLOC"
通过分层分类、自动故障转移处理和成本优化，智能地将请求路由至最优的AI模型。
路由系统透明运行——用户正常发送消息，即可获得最适合其任务的最佳模型响应。无需特殊指令。
可选可见性：在任何消息中包含 [show routing] 即可查看路由决策详情。
路由系统采用三层决策流程：
┌─────────────────────────────────────────────────────────────────┐
│ 第一层：意图检测 │
│ 识别请求的主要目的 │
├─────────────────────────────────────────────────────────────────┤
│ 代码 │ 分析 │ 创意 │ 实时 │ 通用 │
│ 编写/调试 │ 研究 │ 写作 │ 新闻/直播 │ 问答/聊天 │
│ 重构 │ 解释 │ 故事 │ X/Twitter │ 翻译 │
│ 审查 │ 比较 │ 头脑风暴 │ 价格 │ 总结 │
└──────┬───────┴──────┬──────┴─────┬──────┴─────┬─────┴─────┬─────┘
│ │ │ │ │
▼ ▼ ▼ ▼ ▼
┌─────────────────────────────────────────────────────────────────┐
│ 第二层：复杂度评估 │
├─────────────────────────────────────────────────────────────────┤
│ 简单（$级） │ 中等（$$级） │ 复杂（$$$级） │
│ • 单步任务 │ • 多步任务 │ • 深度推理 │
│ • 简短回复即可 │ • 有一定复杂性 │ • 大量输出 │
│ • 事实查询 │ • 中等上下文 │ • 关键任务 │
│ → Haiku/Flash │ → Sonnet/Grok/GPT → Opus/GPT-5 │
└──────────────────────────┴─────────────────────┴───────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 第三层：特殊情形覆盖 │
├─────────────────────────────────────────────────────────────────┤
│ 条件 │ 覆盖至 │
│ ───────────────────────────────────┼───────────────────────────│
│ 上下文 >100K tokens │ → Gemini Pro (1M 上下文) │
│ 上下文 >500K tokens │ → Gemini Pro 唯一 │
│ 需要实时数据 │ → Grok（无条件） │
│ 图像/视觉输入 │ → Opus 或 Gemini Pro │
│ 用户显式覆盖 │ → 请求的模型 │
└────────────────────────────────────┴────────────────────────────┘
当请求包含多个明确意图时（例如："编写代码来分析这些数据并进行创意性解释"）：
1. 识别主要意图——核心交付物是什么？
2. 路由至最高能力模型——混合任务需要多功能性
3. 默认设为复杂复杂度——多意图 = 多步骤
示例：
- "编写代码并解释其工作原理" → 代码（主要）+ 分析 → 路由至 Opus
- "总结这个内容并告诉我关于它的最新消息" → 实时意图优先 → Grok
- "基于真实时事创作故事" → 实时 + 创意 → Grok（实时性优先）
非英语请求正常处理——所有支持的模型都具备多语言能力：
| 模型 | 非英语支持能力 |
|---|---|
| Opus/Sonnet/Haiku | 优秀（100+ 种语言） |
| GPT-5 | 优秀（100+ 种语言） |
| Gemini Pro/Flash | 优秀（100+ 种语言） |
| Grok | 良好（主要语言） |
意图检测仍然有效，因为：
- 关键词模式包含常见的非英语等效词
- 代码意图通过文件扩展名、代码块检测（与语言无关）
- 复杂度通过查询长度估算（跨语言有效）
边缘情况：如果因语言导致意图不明确，则默认设为通用意图，复杂度为中等。
| 意图 | 简单 | 中等 | 复杂 |
|---|---|---|---|
| 代码 | Sonnet | Opus | Opus |
| 分析 | Flash | GPT-5 | Opus |
| 创意 | Sonnet | Opus | Opus |
| 实时 | Grok | Grok | Grok-3 |
| 通用 | Flash | Sonnet | Opus |
当模型在会话中变得不可用（令牌配额耗尽、达到速率限制、API错误）时，路由系统会自动切换到下一个最佳可用模型，并通知用户。
当因耗尽而发生模型切换时，用户会收到通知：
┌─────────────────────────────────────────────────────────────────┐
│ ⚠️ 模型切换通知 │
│ │
│ 您的请求无法在 claude-opus-4-5 上完成 │
│ （原因：令牌配额耗尽）。 │
│ │
│ ✅ 请求已使用以下模型完成：anthropic/claude-sonnet-4-5 │
│ │
│ 下方的回复由故障转移模型生成。 │
└─────────────────────────────────────────────────────────────────┘
| 原因 | 描述 |
|---|---|
token quota exhausted |
达到每日/每月令牌限制 |
rate limit exceeded |
每分钟请求过多 |
context window exceeded |
输入对于模型过大 |
API timeout |
模型响应超时 |
API error |
提供商返回错误 |
model unavailable |
模型暂时离线 |
def execute_with_fallback(primary_model: str, fallback_chain: list[str], request: str) -> Response:
"""
执行请求，包含自动故障转移和用户通知。
"""
attempted_models = []
switch_reason = None
# 首先尝试主模型
models_to_try = [primary_model] + fallback_chain
for model in models_to_try:
try:
response = call_model(model, request)
# 如果切换了模型，在回复前添加通知
if attempted_models:
notification = build_switch_notification(
failed_model=attempted_models[0],
reason=switch_reason,
success_model=model
)
return Response(
content=notification + "\n\n---\n\n" + response.content,
model_used=model,
switched=True
)
return Response(content=response.content, model_used=model, switched=False)
except TokenQuotaExhausted:
attempted_models.append(model)
switch_reason = "token quota exhausted"
log_fallback(model, switch_reason)
continue
except RateLimitExceeded:
attempted_models.append(model)
switch_reason = "rate limit exceeded"
log_fallback(model, switch_reason)
continue
except ContextWindowExceeded:
attempted_models.append(model)
switch_reason = "context window exceeded"
log_fallback(model, switch_reason)
continue
except APITimeout:
attempted_models.append(model)
switch_reason = "API timeout"
log_fallback(model, switch_reason)
continue
except APIError as e:
attempted_models.append(model)
switch_reason = f"API error: {e.code}"
log_fallback(model, switch_reason)
continue
# 所有模型均已耗尽
return build_exhaustion_error(attempted_models)
def build_switch_notification(failed_model: str, reason: str, success_model: str) -> str:
"""构建模型切换时的用户通知。"""
return f"""⚠️ **模型切换通知**
您的请求无法在 `{failed_model}` 上完成（原因：{reason}）。
✅ **请求已使用以下模型完成：** `{success_model}`
下方的回复由故障转移模型生成。"""
def build_exhaustion_error(attempted_models: list[str]) -> Response:
"""构建所有模型耗尽时的错误信息。"""
models_tried = ", ".join(attempted_models)
return Response(
content=f"""❌ **请求失败**
无法完成您的请求。所有可用模型均已耗尽。
**尝试过的模型：** {models_tried}
**您可以：**
1. **等待** —— 令牌配额通常每小时或每日重置
2. **简化请求** —— 尝试更短或更简单的请求
3. **检查状态** —— 运行 `/router status` 查看模型可用性
如果问题持续存在，您的人类管理员可能需要检查API配额或添加其他提供商。""",
model_used=None,
switched=False,
failed=True
)
当模型耗尽时，路由系统会为相同任务类型选择下一个最佳模型：
| 原始模型 | 故障转移优先级（相同能力） |
|---|---|
| Opus | Sonnet → GPT-5 → Grok-3 → Gemini Pro |
| Sonnet | GPT-5 → Grok-3 → Opus → Haiku |
| GPT-5 | Sonnet → Opus → Grok-3 → Gemini Pro |
| Gemini Pro | Flash → GPT-5 → Opus → Sonnet |
| Grok-2/3 | （警告：无实时故障转移可用） |
模型切换后，代理应在回复中注明：
1. 原始模型不可用
2. 实际完成请求的模型
3. 回复质量可能与原始模型的典型输出有所不同
这确保了透明度并设定了适当的期望。
使用流式响应时，故障转移处理需要特殊考虑：
async def execute_with_streaming_fallback(primary_model: str, fallback_chain: list[str], request: str):
"""
处理流式响应中的故障转移。
如果模型在流式传输过程中（而非之前）失败，部分响应将丢失。
策略：在成功收到第一个数据块之前，不开始流式传输。
"""
models_to_try = [primary_model] + fallback_chain
for model in models_to_try:
try:
# 首先使用非流式ping测试（可选，会增加延迟）
# await test_model_availability(model)
# 开始流式传输
stream = await call_model_streaming(model, request)
first_chunk = await stream.get_first_chunk(timeout=10_000) # 第一个数据块10秒超时
# 如果执行到这里，说明模型正在响应——继续流式传输
yield first_chunk
async for chunk in stream:
yield chunk
return # 成功
except (FirstChunkTimeout, StreamError) as e:
log_fallback(model, str(e))
continue # 尝试下一个模型
# 所有模型都失败
yield build_exhaustion_error(models_to_try)
关键点：在收到第一个数据块后再确认使用该模型。如果第一个数据块超时，则在向用户显示任何部分响应之前进行故障转移。
RETRY_CONFIG = {
"initial_timeout_ms": 30_000, # 首次尝试30秒超时
"fallback_timeout_ms": 20_000, # 故障转移尝试20秒超时（更快失败）
"max_retries_per_model": 1, # 不重试同一模型
"backoff_multiplier": 1.5, # 未使用（无同一模型重试）
"circuit_breaker_threshold": 3, # 跳过模型前的失败次数阈值
"circuit_breaker_reset_ms": 300_000 # 5分钟后重试已失败的模型
}
熔断器：如果一个模型在5分钟内失败3次，则在接下来的5分钟内完全跳过它。这可以防止反复访问已宕机的服务。
当首选模型失败（速率限制、API宕机、错误）时，按顺序降级到下一个选项：
Opus → Sonnet → GPT-5 → Gemini Pro
Opus → GPT-5 → Gemini Pro → Sonnet
Opus → GPT-5 → Sonnet → Gemini Pro
Grok-2 → Grok-3 → （警告：无实时故障转移）
Flash → Haiku → Sonnet → GPT-5
┌─────────────────────────────────────────────────────────────────┐
│ 长上下文故障转移链 │
├─────────────────────────────────────────────────────────────────┤
│ 令牌数量 │ 故障转移链 │
│ ───────────────────┼───────────────────────────────────────────│
│ 128K - 200K │ Opus (200K) → Sonnet (200K) → Gemini Pro │
│ 200K - 1M │ Gemini Pro → Flash (1M) → ERROR_MESSAGE │
│ > 1M │ ERROR_MESSAGE（无模型支持） │
└─────────────────────┴───────────────────────────────────────────┘
实现：
```python
def handle_long_context(token_count: int, available_models: dict) -> str | ErrorMessage:
"""处理长上下文请求，提供优雅降级。"""
# 第1级：128K - 200K tokens（Opus/Sonnet可以处理）
if token_count <= 200_000:
for model in ["opus", "sonnet", "haiku", "gemini-pro", "flash"]:
if model in available_models and get_context_limit(model) >= token_count:
return model
# 第2级：200K - 1M tokens（仅Gemini）
elif token_count <= 1_000_000:
for model in ["gemini-pro", "flash"]:
if model in available_models:
return model
# 第3级：> 1M tokens（无可用模型）
# 降级到错误处理
# 未找到合适的模型——返回有用的错误信息
return build_context_error(token_count, available_models)
def build_context_error(token_count: int, available_models: dict) -> ErrorMessage:
"""当没有模型能处理输入时，构建有用的错误信息。"""
# 查找最大的可用上下文窗口
max_available = max(
(get_context_limit(m) for m in available_models),
default=0
)
# 确定缺少什么
missing