什么是 Fusion Gateway？

想象你有三位顶尖同事激烈争论你的问题，然后一位资深主编裁定谁的论点更有说服力——这套软件做的就是这件事。

一句话讲清楚

你每天都在用AI 编程助手。它们很强——但任何单一 AI 都有盲区：容易过度自信，或者只从一个角度看问题。

Fusion Gateway 是一个代理服务器，它拦截你向 AI 发出的请求，暗中同时咨询多个 AI 模型，取各家之长，返回一个综合后的最佳答案。

💡

一句话电梯游说

你向 Claude Code 提出一个难题。不是一个 AI 独自思考，而是三个 AI 从不同角度同时思考——一个专注架构设计，一个专注落地实现，一个专门找漏洞。第四个 AI 读完三份答案后做出最终裁决。你拿到的是那个经过综合、打磨过的最终答案。

四个"虚拟"模型

从外部看，Fusion Gateway 就像一个普通的OpenAI 兼容 API。你只需指定四个特殊模型名，每个名字触发不同的处理策略：

fusion-single 单个 AI。速度快、成本低。适合快速提问和简单任务。

fusion-fast 三个工作节点 + 一个裁决者。标准多模型工作流。工作节点超时时间 30 秒。

fusion-critical 同样是三个工作节点 + 一个裁决者，但超时时间延长至 60 秒——用于高风险决策。

fusion-auto 智能模式：读取你的消息，自动决定使用哪种策略。

跟着一次请求从头走到尾

假设你输入："设计一个 Yocto 系统的 OTA 更新方案"。你用的是 fusion-auto。幕后发生了什么？

🖥️

Claude Code
/ 你

🔀

路由器

📋

模型
注册表

🎯

角色
分配器

⚙️

工作节点
(×3)

⚖️

裁决者

点击"下一步"开始

理解这些能给你带来什么？

作为一个使用 AI 编程工具的人，理解这套架构会给你三项超能力：

🎯

更精准地掌控它

你可以强制指定路由：快速修改用 fusion-single，架构决策用 fusion-critical。当你比系统更清楚时，别让自动路由替你做决定。

🔍

调试问题时有章可循

每次请求都会在 runs/ 下保存调试文件。答案有问题时，可以看每个工作节点说了什么、裁决者保留了哪些内容。

💰

管理 API 用量成本

fusion-fast 每个问题消耗 4 次 API 调用（3 个工作节点 + 1 个裁决者）。预算追踪器在接近每小时/每日限额时自动降级为 single 模式。

四种角色

系统如何决定让哪个 AI 扮演哪个角色——以及为什么"多元视角"才是核心所在。

四种思维，四种职责

Fusion 将每个 AI 模型分配到特定的认知角色。可以把它想象成一个同行评审小组：每位评审人都有不同的职责范围。

🏛️

架构师

专注于系统设计、模块边界和长期演化方向。硬性规则：不允许写实现代码。只能提供接口定义、伪代码和设计权衡。

🔧

实现者

专注于把想法落地：真实代码、文件结构、错误处理、测试骨架。硬性规则：不讨论架构。默认设计方案已经确定。

🔍

评审者

发现故障模式、隐性成本、安全风险和边界情况。硬性规则：不允许提供解决方案。只能说"这里会出问题，原因是……"

⚖️

裁决者

读取三个工作节点的全部输出，识别共识与分歧，作出裁定，并综合出最终答案。唯一对外发声的角色。

同一个问题，各角色怎么回答

你问："嵌入式设备上的 OTA 更新失败了该怎么处理？" 下面是每个角色的回应——它们完全独立作答，裁决者在看到这些输出之前不会干预：

🧠

为什么要分角色？

如果你让一个 AI 同时"设计 + 实现 + 找风险"，它往往会一直待在同一种思维模式里。通过系统提示词强制划定角色边界，才能产生真正的视角多样性——实现者无法讨论架构，评审者无法提供解决方案，这是硬性约束。

模型如何被分配到角色

面对 130+ 个可用模型，系统需要判断谁最擅长什么。它使用 role_assigner.py 中的启发式评分系统。

          role_assigner.py
# Scores for "implementer" role
_CODING = {
    "coder": 35,
    "deepseek": 24,
    "qwen": 18,
    "claude": 14,
}
# Cheap models get bonus for workers
_COST_CHEAP = ["mini", "flash", "haiku", "nano"]
# Expensive models get bonus for judge
_COST_EXPENSIVE = ["opus", "reasoner"]
        

白话解释

🏷️ 用一个字典把名称关键词映射到分值。如果模型名里含有"coder"，它在实现者角色上就得 +35 分。

💸 便宜的模型（mini、flash、haiku）在工作节点角色上获得 +20 加分——探索阶段用便宜的就够了。

👑 昂贵的旗舰模型（opus、reasoner）在裁决者角色上获得 +20 加分——综合输出的质量在这里最关键。

⚠️ 这不是 AI 在做决策，就是简单的算术：把关键词匹配分加起来，取最高分的那个。

多样性规则

关键洞察在这里：三个 Claude 模型和一个没什么区别。你需要的是真正的视角多样性，而不是三个来自同一训练血统的 AI。

分配器会施加多样性惩罚：如果某个模型的系列已被另一个工作节点使用，它会扣 −90 分；同公司但不同系列则扣 −45 分。

🎲

真实分配示例

architect → claude-opus（Anthropic）· implementer → sky/deepseek-v4-pro（DeepSeek）· critic → hewei/MiniMax-M2.7（MiniMax）· judge → claude-opus-4-6（Anthropic）。三个工作节点来自四家不同公司，裁决者选用最适合综合推理的模型。

          role_assigner.py
def diversified_score(model, role, selected):
    score = score_model_for_role(model, role)
    identity = model_identity(model.id)
    for sel in selected:
        sel_id = model_identity(sel.id)
        if identity.family == sel_id.family:
            score -= 90
        elif identity.provider == sel_id.provider:
            score -= 45
    return score
        

白话解释

📊 先算出这个模型在当前角色上的基础分。

🔍 逐一检查已经分配到工作节点的每个模型。

👨‍👩‍👧 如果新模型和已有工作节点属于同一系列——扣 90 分，重罚。

🏢 如果是同一家公司但不同系列——扣 45 分，中等惩罚。

🏆 最终得分最高的模型获得该角色。多样性被直接写进数学公式里。

检验一下你的理解

你在查看工作节点的输出，其中一条写道："如果写入途中断电，两个分区都会损坏，设备就变砖了。" 这是哪个角色写的？

你用三个模型配置了 Fusion：claude-opus、claude-sonnet 和 claude-haiku，全都来自 Anthropic。问题出在哪里？

Fusion 处理流程

从你的消息到达那一刻，到你看到答案的那一刻——代码所做的每一个决策，用大白话讲清楚。

第一步：路由器决定使用哪种模式

当你使用 fusion-auto 时，路由器会读取你的消息并决定采用哪种策略。它按层次工作——就像一位先检查最严重症状的分诊护士：

关键升级检查

消息中是否包含"系统架构"、"architecture"、"production"、"security audit"等词？如果是 → fusion-critical。这些词意味着高风险决策，值得投入更多思考时间。

复杂度否决

是否包含"实现"、"重构"、"implement"、"refactor"、"debug"？如果是 → fusion-fast。这些是实质性任务，多视角能带来额外价值。

单模型触发词

是否匹配"什么是"、"explain"、"what is"、"rename"、"translate"等简单模式？如果是 → fusion-single。查个定义不需要召开专家小组。

长度兜底

消息是否短于 50 个字符？如果是 → fusion-single。短消息通常是简单问题。否则 → fusion-fast。

🔒

系统提示词不会干扰路由

Claude Code 会发送一段带有"实现此功能"等指令的长系统提示词。路由器只读取 role: "user" 的消息，因此系统提示词中的词语不会意外触发错误路由。

路由器的代码实现

router.py 中的路由逻辑返回一个包含所有决策细节的数据类，便于调试时记录日志：

          router.py
def route_virtual_model(model, messages):
    if model in {"fusion-single", "fusion-fast", ...}:
        return RouteDecision(route=model, ...)
    text = message_text(messages)  # user-only
    lower = text.lower()
    critical_hit = next((kw for kw in _CRITICAL_KEYWORDS
                          if kw in lower), None)
    if critical_hit:
        return RouteDecision(route="fusion-critical", ...)
    # ... 后续是否决、触发词、长度检查
        

大白话解释

🎯 如果你明确指定了 fusion-single、fusion-fast 或 fusion-critical，就直接用那个，无需路由判断。

📝 只提取用户消息（跳过系统提示词），并全部转为小写以便匹配。

🔍 扫描关键词。next() 遇到第一个匹配就停止——快速且低开销。

⬆️ 如果命中关键词，返回一个 RouteDecision 对象，记录命中详情（便于调试日志）。

第二步：三个工作节点同时运行

路由决定后，三个工作节点通过并行执行立刻启动。每个工作节点获得不同的系统提示词，将其锁定在对应角色：

          worker_runner.py
temperatures = {"architect": 0.4,
               "implementer": 0.3,
               "critic": 0.5}
tasks = [
    run_worker(client, role, model, messages,
               temperatures.get(role, 0.3), timeout=worker_timeout)
    for role, model in role_models.items()
]
results = await asyncio.gather(*tasks)
        

大白话解释

🌡️ 每个角色使用不同的温度值。评审者取 0.5（最随机）——你希望它能想到不寻常的失败场景。实现者取 0.3（最精确）——代码需要具体明确。

📋 构建工作节点任务列表——每个角色一个，每个任务对应一次不同 AI 模型的调用。

⚡ asyncio.gather() 同时启动所有任务并等待全部完成。总等待时间 = 最慢的工作节点耗时，而非所有工作节点耗时之和。

🔄

工作节点失败了怎么办？

工作节点在遇到网络错误时最多重试 2 次（退避时间为 1 秒、2 秒）。其他错误则立即失败。如果成功的工作节点少于 2 个，系统会降级为单模型响应，而不尝试裁决者综合。

第三步：裁决者综合输出

在把任何内容发给裁决者之前，系统会先检查各工作节点的答案是否一致。它使用一种巧妙的字符 n-gram 相似度来比较输出——无需昂贵的嵌入模型：

          judge_runner.py
def _char_ngram_similarity(a, b, n=3):
    def ngrams(text):
        t = re.sub(r"\s+", "", text.lower())
        return {t[i:i+n] for i in range(len(t)-n+1)}
    sa, sb = ngrams(a), ngrams(b)
    return len(sa & sb) / len(sa | sb)
        

大白话解释

✂️ 将两段文本切成重叠的 3 字符窗口。"hello" → {"hel", "ell", "llo"}。

🔤 先去掉空格并转为小写——"代理层"和"代理层"应该视为相同。

🧮 & = 共有字符序列，| = 所有字符序列合并。相除得到相似度分数（0 到 1）。

🇨🇳 对中文效果极好，因为"代理"这样的中文语素会出现在"中间层"和"代理层"等近义词中——它们共享"层"字，即便是同义词，相似度也不会降为零。

若工作节点趋于一致（相似度 ≥ 0.12），裁决者使用较短的"综合"提示词；若有分歧，则使用更完整的"仲裁"提示词，要求其明确解决冲突。两种情况下，裁决者的输出都包含两个部分：

🔒

## 内部分析

内部推理：共识、分歧、裁决结论、独特见解、盲点。保存到磁盘用于调试——不会发给你。

📤

## 对外答案

公开答案：最终方案、验证清单和下一步行动。通过 API 提取并返回给你。

第四步：只提取你需要的内容

网关不会直接返回裁决者的完整输出，而是使用正则表达式解析裁决者的 Markdown，只提取配置中指定的公开部分：

          judge_runner.py
def extract_public_output(full_output, section_names):
    public_match = re.search(
        r"(?ims)^##\s*对外答案\s*$([\s\S]*)",
        full_output
    )
    search_text = public_match.group(1) if public_match else full_output
    # 然后从 search_text 中提取各命名章节
        

大白话解释

🔍 在裁决者的完整输出中搜索标题"## 对外答案"。

✂️ 如果找到，取该标题之后的所有内容——这就是公开区域。如果没找到，使用全部内容（优雅降级）。

📑 然后在公开区域内扫描配置中的各章节名称（最终方案、验证清单、下一步），逐一提取。

📁

磁盘上保存了什么

每个开启调试的请求会将文件写入 runs/YYYY-MM-DD/<task-id>/：worker.architect.md、worker.implementer.md、worker.critic.md、judge.md（完整输出）、final.md（你收到的内容），以及 metadata.json（路由信息、模型分配、延迟、费用）。旧日志 7 天后自动删除。

检验你的理解

你用 fusion-fast 问了 10 个问题。网关总共发起了多少次上游 API 调用？

用户问道："这个配置里的 debug 模式有什么用？"fusion-auto 会选择哪条路由？

巧妙的设计

BudgetTracker、健康检查、重试逻辑与收敛检测——让系统在生产环境中稳定运行的工程模式。

自动成本保护

fusion-fast 每次请求消耗 4 次 API 调用。高频用户很快就会触及速率限制。BudgetTracker 监控你的用量，当接近每小时或每日限制时自动降级为 fusion-single：

          budget.py
def check(self):
    self._reset_windows_if_needed()
    hourly_threshold = int(
        self.config.hourly_call_limit *
        self.config.auto_downgrade_at
    )
    if self._hourly_calls >= hourly_threshold:
        return BudgetDecision(True, "hourly budget...")
    return BudgetDecision(False)
        

白话解释

🕐 如果距离开始计数已过去一小时，将小时计数器归零。每日计数器同理。

📊 计算阈值：若限制为每小时 200 次调用，auto_downgrade_at 为 0.9，则阈值为 180 次。

⚠️ 如果本小时内已发起 180 次以上调用，返回"是，需要降级"的决定，并附带说明。

✅ 否则，返回"否，无需降级"的决定。API 层读取此决定，必要时覆盖你所请求的路由。

💾

预算是进程本地的

计数器存在内存中，重启网关后清零。对于分布式部署，你需要共享存储（Redis、数据库）。但对于单用户本地网关，内存计数器快速简单——零数据库依赖。

后台健康检查

上游服务器暴露了 130 多个模型，部分可能离线或损坏。网关可选择性地运行健康检查——但对每次请求都运行 130 次检查会增加 20 秒以上的延迟。因此改为在后台运行：

          api.py
@app.on_event("startup")
async def startup():
    # Non-blocking background health check
    asyncio.create_task(
        app.state.registry.get_models(health_check=True)
    )
    # Refresh every 5 minutes
    asyncio.create_task(background_health_check())
        

白话解释

🚀 网关启动时，立即在后台启动一个任务来获取并测试所有模型。无需等待其完成——让服务器立刻开始接受请求。

🔄 同时启动第二个后台任务，它会永久循环，每次健康检查之间休眠 5 分钟，保持模型列表的新鲜度。

⏱️ 首次请求时使用上次健康检查的缓存数据。在用户发起第二次请求时，新鲜数据已就绪。用户侧零延迟。

健康检查运行时，向每个模型发送一个简单问题（如"hello"），超时时间为 15 秒。模型响应则标记为 healthy=True，超时或报错则标记为 healthy=False。角色分配器对不健康的模型扣减 −1000 分，实际上将其排除在外。

工作节点重试逻辑

网络请求会失败，在中国大陆通过代理路由时尤为如此。工作节点不会在第一次超时就放弃，而是以指数退避方式重试网络错误：

          worker_runner.py
async def run_worker(..., max_retries=2):
    for attempt in range(max_retries + 1):
        try:
            output = await client.chat_text(...)
            return role, output, None
        except (httpx.TimeoutException, httpx.NetworkError):
            if attempt < max_retries:
                await asyncio.sleep(1 * (attempt + 1))
                continue
            return role, None, "retries exhausted"
        

白话解释

🔁 最多循环 3 次（第 0、1、2 次尝试），调用上游 API。

✅ 若调用成功，立即返回输出——无需重试。

💥 若发生超时或网络错误，且重试次数未耗尽，则休眠（attempt + 1）秒。第一次重试等 1 秒，第二次等 2 秒。

❌ 若所有重试均失败，返回 None 及错误信息。触发降级逻辑——若少于 2 个工作节点成功，切换到单模型模式。

⚡

并非所有错误都会重试

只有 TimeoutException 和 NetworkError 会重试——这些是瞬时性故障（代理抖动、短暂拥塞）。其他错误（认证失败、模型不存在、响应格式错误）会立即失败——重试解决不了这类问题。

处理异常的上游响应

上游服务器有时对非流式请求也返回 SSE 流。网关将其规范化为标准 JSON 响应：

          worker_runner.py
def parse_upstream_response(response, model_id):
    content_type = response.headers.get("content-type", "")
    if "text/event-stream" not in content_type:
        return response.json()
    content_parts = []
    for line in response.text.splitlines():
        if not line.startswith("data:"): continue
        data = line.removeprefix("data:").strip()
        chunk = json.loads(data)
        delta = chunk["choices"][0]["delta"]
        if delta.get("content"):
            content_parts.append(delta["content"])
    return {"choices": [{"message": {"content": "".join(content_parts)}}]}
        

白话解释

🔍 检查响应的 Content-Type 头。如果是普通 JSON，直接解析并返回。

📊 如果是 SSE 流，读取完整响应文本并按行分割，找出以"data:"开头的行。

🧩 每个 data 行都是一个 JSON 块，提取 delta.content 字段（增量文本片段）并追加到列表中。

📦 最后，将所有内容片段拼接成一个字符串，封装成标准 OpenAI 响应格式。至此它看起来就像普通的非流式响应。

自动日志清理

每次 fusion-fast 请求会在磁盘上保存 6 个以上的调试文件。高频使用一周后，目录数量会达到数千个。网关在启动时自动清理 7 天前的日志：

          logging_store.py
def cleanup_before(self, days):
    cutoff = time.time() - days * 86400
    for date_dir in Path(self.config.logging.dir).iterdir():
        try:
            date_ts = datetime.strptime(
                date_dir.name, "%Y-%m-%d"
            ).timestamp()
            if date_ts < cutoff:
                shutil.rmtree(date_dir)
        except: pass
        

白话解释

⏰ 计算截止时间戳：当前时间减去 7 天（7 × 86400 秒）。

📁 遍历 runs/ 下的所有目录，每个目录命名格式如"2026-06-15"。

🗓️ 将目录名解析为日期并转换为时间戳。

🗑️ 若该时间戳早于截止时间，递归删除整个目录树——包括其中所有文件和子目录。

🛡️ 用 try/except 包裹，防止一个异常目录名导致清理流程崩溃。

最终测验

默认配置：hourly_call_limit=200，auto_downgrade_at=0.9。你只使用 fusion-fast。在自动降级触发之前，你能提问多少次？

一个工作节点调用因 API 密钥错误而返回 401 Unauthorized。会发生什么？

为什么健康检查在后台运行，而不是在每次请求时运行？

你现在掌握了什么

你刚刚从内部学习了 Fusion Gateway 的工作原理——不是通过记忆 API 文档，而是通过追踪真实的数据流并阅读实际代码。你现在能够：

🎯

自信地驾驭它

你理解自动路由何时选择各种模式，以及何时手动覆盖。你知道 fusion-fast 消耗 4 次 API 调用，以及 BudgetTracker 存在的原因。

🐛

更聪明地调试

当答案看起来有问题时，你可以读取保存的调试文件，查看每个工作节点的输出，以及裁决者选择保留或丢弃了什么。

🔧

自行扩展它

你已经了解了评分系统、路由层、重试逻辑。现在你可以向路由器添加自定义关键词、调整多样性惩罚，或更改返回给用户的内容章节。

🚀

下一步

尝试在本地运行网关（make start CONFIG=config.local.yaml），将 Claude Code 指向它，然后观察 runs/ 目录被调试日志填满。选一个复杂的问题，读取所有工作节点的输出，看看裁决者如何综合它们。那一刻，一切豁然开朗。