nJcx's Blog

十年生死两茫茫,写程序,到天亮。相顾无言,惟有泪千行

大模型网关里面的智能路由技术实践


介绍

大模型网关的智能路由(Intelligent Routing)是近年来随着大模型应用爆发而兴起的一项关键技术。它的核心目标是:在多个可用的模型(或模型服务)之间,根据请求的特征、实时状态和业务目标,自动选择最合适的模型来处理请求。你在很多 IDE 里看到的 “Auto” 选项,就是智能路由的一种典型应用——它让用户无需手动选择模型,系统会自动判断当前任务(比如写代码、解释代码、聊天)适合用哪个模型,从而在效果、速度和成本之间取得平衡。

一、智能路由要解决什么问题?

假设你有一个大模型网关,后面接入了多个模型:

Kimi-k3:能力强,但贵、慢

DeepSeek-v4-flash:便宜、快,但复杂任务可能不够好

Qwen3.7-MAX :能力可以,价格中等

本地小模型(如 Qwen 3.5 9B):几乎免费、极快,但只能处理简单任务

用户的请求是多样的:有的只是问“今天天气怎么样”,有的是“帮我重构这段 500 行的 Python 代码”。如果所有请求都发给最贵的模型,成本会爆炸;如果都发给小模型,复杂任务质量很差。

智能路由的目标就是动态决策:把每个请求送到“性价比”最高的模型上,同时满足用户对质量和延迟的期望。

二、智能路由的常见策略

智能路由可以从简单到复杂,大致分为以下几类:

算法类型 原理 优势 局限
基于规则 正则匹配、关键词、前缀树 毫秒级响应 只能处理固定格式
基于 Embedding 将请求和模型描述转为向量,计算余弦相似度 能理解同义表达 长尾意图覆盖不足
基于 LLM 用大模型分析输入,输出路由标签 灵活度高,能处理模糊输入 延迟高(秒级),有幻觉风险
基于 ML 分类器、BERT模型 训练轻量分类模型(逻辑回归、小参数LLM等) 兼顾准确率与低延迟 依赖标注数据
三、智能路由系统的典型架构

一个完整的智能路由系统通常包含以下组件:

用户请求 → [请求分析器] → [特征提取] → [路由策略] → [模型调用] → [响应]
                ↑                              ↓
                └────────── 反馈日志 ←──────────┘

请求分析器:解析请求内容,提取结构化信息(任务类型、语言、长度、敏感词等)

特征提取:将请求转换为路由策略可用的特征向量(如嵌入、关键词、元数据)

路由策略:核心决策模块,根据特征和当前系统状态选择模型

模型调用:实际向选定模型发送请求并获取响应

反馈日志:记录每次请求的决策、结果质量(如用户反馈、自动评估指标)、延迟、成本等

离线/在线学习:利用反馈数据更新路由策略(如重新训练分类器、更新MAB参数)

成熟的路由系统不只看"任务难不难",还会综合考虑:

成本优化:根据实时 Token 单价选择最低成本模型,或在预算约束下平衡成本与质量

性能与可用性:基于实时延迟、吞吐量、错误率动态选路;支持主备切换、故障自动降级

难度感知:区分简单提示与复杂推理任务,仅对高难度任务调用前沿模型

复合策略:结合历史 Trace 数据与质量反馈,通过强化学习实现自适应调优

基于 Embedding 的分类器实践

基于 Embedding 的分类器实践核心流程为:构建标注样本库 → 向量化存储 → 实时请求 Embedding → 余弦相似度计算 → 阈值判定 → 模型分发。 其本质是将路由问题转化为“语义最近邻搜索”问题,通过比较用户请求与预置样本的向量距离来自动选择模型。

用户请求 (model="auto")
    ↓
① Embedding 模型将请求文本 → 转为向量 (如 1536 维)
    ↓
② 与样本库中预标注的向量做余弦相似度计算
    ↓
③ 取 TopK 最相似样本,看它们属于哪个标签
    ↓
④ 最高相似度 ≥ 阈值 → 采纳样本标签;否则 → 退回走规则兜底
    ↓
⑤ 根据标签选择对应模型组,转发请求

第一步:构建样本库(离线阶段) 这是整个系统的“弹药库”。你需要为每个路由类别准备若干代表性样本。

# 样本库结构示意
sample_library = {
    "simple": [
        "你好",
        "今天天气怎么样",
        "帮我翻译这句话",
        "这段代码什么意思",
        "写一个 Hello World",
        # ... 每个类别 5~20 条即可
    ],
    "complex": [
        "帮我设计一个微服务架构方案",
        "分析这段代码的性能瓶颈并给出优化建议",
        "重构这个模块,要求支持插件化扩展",
        "帮我写一个完整的用户认证系统",
        # ...
    ],
    "code_generation": [
        "用 Python 实现一个 LRU 缓存",
        "写一个 React 组件,支持拖拽排序",
        # ...
    ]
}

关键参数:

每个类别 5~20 条样本即可冷启动,覆盖主要口语变体

样本要覆盖不同表达方式(正式/口语/简短/详细),而非堆砌相似句

样本质量 > 数量,一条坏样本可能污染整个类别的匹配

第二步:向量化存储(离线阶段) 用 Embedding 模型将所有样本转为向量,存入向量数据库(或直接用内存中的 NumPy 数组)。

import numpy as np
from openai import OpenAI

client = OpenAI()

class EmbeddingRouter:
    def __init__(self, embedding_model="text-embedding-3-small"):
        self.embedding_model = embedding_model
        self.samples = {}       # {label: [text, ...]}
        self.sample_vectors = {} # {label: np.array}

    def build_index(self, sample_library: dict):
        """离线构建向量索引"""
        for label, texts in sample_library.items():
            # 批量调用 Embedding API
            response = client.embeddings.create(
                model=self.embedding_model,
                input=texts
            )
            vectors = np.array([item.embedding for item in response.data])
            self.samples[label] = texts
            self.sample_vectors[label] = vectors

        print(f"索引构建完成: {len(self.samples)} 个类别, "
              f"共 {sum(len(v) for v in self.samples.values())} 条样本")

第三步:实时路由决策(在线阶段)

请求进来后,实时 Embedding → 计算相似度 → 决策。

def cosine_similarity(a, b):
    """计算两个向量的余弦相似度"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def route(self, user_query: str, threshold=0.72, top_k=5):
    """
    核心路由逻辑
    - threshold: 相似度阈值,超过才采纳样本匹配结果
    - top_k: 取最相似的 K 个样本参与投票
    """
    # 1. 将用户请求向量化
    response = client.embeddings.create(
        model=self.embedding_model,
        input=[user_query]
    )
    query_vector = np.array(response.data[0].embedding)

    # 2. 与所有样本计算相似度,找 TopK
    all_scores = []  # [(score, label), ...]

    for label, vectors in self.sample_vectors.items():
        for vec in vectors:
            score = cosine_similarity(query_vector, vec)
            all_scores.append((score, label))

    # 按相似度降序排列,取 TopK
    all_scores.sort(key=lambda x: x[0], reverse=True)
    top_k_scores = all_scores[:top_k]

    # 3. 阈值判定
    best_score, best_label = top_k_scores[0]

    if best_score >= threshold:
        # 命中样本库,采用多数投票
        label_votes = {}
        for score, label in top_k_scores:
            if score >= threshold:
                label_votes[label] = label_votes.get(label, 0) + 1

        final_label = max(label_votes, key=label_votes.get)
        return final_label, best_score, "sample_match"
    else:
        # 未命中,走规则兜底
        return self._fallback_rules(user_query), best_score, "rule_fallback"

第四步:规则兜底(当 Embedding 不确定时)

当所有样本相似度都低于阈值时,用轻量规则兜底,避免"强行匹配"导致误路由。

def _fallback_rules(self, query: str) -> str:
    """规则兜底:基于长度、关键词等浅层特征"""

    # 规则1: 超长输入大概率是复杂任务
    if len(query) > 500:
        return "complex"

    # 规则2: 包含特定关键词
    complex_keywords = ["架构", "重构", "优化", "性能", "设计模式", "微服务"]
    if any(kw in query for kw in complex_keywords):
        return "complex"

    # 规则3: 包含代码块
    if "```" in query or "def " in query or "function " in query:
        return "code_generation"

    # 默认走简单模型
    return "simple"

关键参数调优指南

参数 作用 调优建议
threshold(阈值) 相似度超过此值才采纳样本匹配 通常 0.70~0.80;太高会导致大量走兜底,太低会误匹配
top_k 参与投票的最相似样本数 通常 3~5;太小容易单样本误判,太大引入噪声
Embedding 模型选择 决定语义理解能力 轻量场景用 text-embedding-3-small(1536维);中文场景可用 Qwen3-Embedding-0.6B 等
样本数量 每类样本的覆盖度 冷启动 5~10 条,生产环境建议 20~50 条,覆盖主要变体

生产环境要注意的坑

Embedding 延迟:每次路由都要调一次 Embedding API,通常 50~200ms。如果对延迟敏感,可以用本地 ONNX 模型(如 OpenSquilla 的做法,用本地 ONNX embedder 完全在设备端完成分类)

样本漂移:用户使用模式会随时间变化,需要定期回标并更新样本库

缓存亲和:同一对话中频繁切换模型会导致 prompt 缓存失效,路由决策要考虑缓存切换成本

阈值不是万能的:边界模糊的请求(如"代码审查"既涉及编码也涉及安全)容易在类别间摇摆,建议设置信度阈值,低于阈值的走默认安全模型
基于BERT模型 的分类器实践

在 LLM 路由场景中,基于 BERT 的分类器属于“基于 ML 分类器的路由”路线,其核心思路是:用 BERT 作为编码器提取语义特征,加一个分类头,直接输出路由决策。BERT(Bidirectional Encoder Representations from Transformers)是 Google 于 2018 年发布的预训练语言模型,其核心突破在于“双向理解”——能同时参考上下文来理解一个词的含义,彻底改变了此前语言模型只能“单向阅读”的局限。它特别擅长理解文本含义,但不擅长生成文本。在大模型网关里,我们正是利用它"理解"的能力来做路由分类。

为什么 BERT 在路由场景中特别合适?

速度快:1.1 亿参数,推理只需几毫秒,远快于大模型
输出确定:直接输出类别概率,不会"幻觉"
可微调:用几百条标注数据就能适配你的业务场景
部署轻:CPU 就能跑,不需要 GPU 集群

现代升级版:ModernBERT 传统 BERT 的上下文长度只有 512 tokens,而现代 LLM 请求动辄几千 tokens。因此业界已经转向 ModernBERT——BERT 的现代化升级版:

上下文长度:8192 tokens(传统 BERT 的 16 倍)
参数量:Base 版 1.39 亿,Large 版 3.95 亿
速度:比传统编码器快 2~4 倍
训练数据:2 万亿 tokens(网页、代码、科学文章),比只基于维基百科训练的传统 BERT 更稳健

已有实际案例使用 ModernBERT 作为统一基座,在其上训练多个 LoRA 分类头,分别处理意图分类、安全拦截、复杂度判断等路由任务。

以下是基于 Hugging Face 的 BERT 路由分类器完整实践流程:

  1. 准备训练数据
from datasets import load_dataset

# 方案A:使用现成的路由数据集
# DevQuasar/llm_router_dataset-synth 包含约 15000 条用户提示
# 标签:1 = large_llm(复杂任务),0 = small_llm(简单任务)
dataset = load_dataset("DevQuasar/llm_router_dataset-synth")

# 方案B:自建数据集(更贴合实际业务)
# 格式:{"prompt": "帮我重构这个模块", "label": 1}
#        {"prompt": "今天天气怎么样", "label": 0}
  1. 加载模型与分词器
from transformers import AutoTokenizer, AutoModelForSequenceClassification

# 推荐用 ModernBERT,传统 BERT 也行
model_id = "answerdotai/ModernBERT-base"
# model_id = "google-bert/bert-base-uncased"  # 传统 BERT

tokenizer = AutoTokenizer.from_pretrained(model_id)
tokenizer.model_max_length = 512

# 加载预训练模型 + 分类头(2 类:简单/复杂)
model = AutoModelForSequenceClassification.from_pretrained(
    model_id,
    num_labels=2,
    label2id={"small_llm": 0, "large_llm": 1},
    id2label={0: "small_llm", 1: "large_llm"},
)
  1. 微调训练
from transformers import TrainingArguments, Trainer

def tokenize(batch):
    return tokenizer(batch['prompt'], padding='max_length', 
                     truncation=True, max_length=512)

tokenized_dataset = dataset.map(tokenize, batched=True)

training_args = TrainingArguments(
    output_dir="./bert-router",
    num_train_epochs=3,
    per_device_train_batch_size=32,
    learning_rate=2e-5,
    evaluation_strategy="epoch",
    save_strategy="epoch",
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=tokenized_dataset["train"],
    eval_dataset=tokenized_dataset["test"],
)

trainer.train()
  1. 推理部署(集成到网关)
import torch

class BERTRouter:
    def __init__(self, model_path="./bert-router"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_path)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_path)
        self.model.eval()

    def route(self, user_query: str, threshold=0.7) -> dict:
        """
        返回路由决策
        - threshold: 置信度阈值,低于此值走兜底策略
        """
        inputs = self.tokenizer(user_query, return_tensors="pt", 
                                truncation=True, max_length=512)

        with torch.no_grad():
            outputs = self.model(**inputs)
            probs = torch.softmax(outputs.logits, dim=-1)[0]

        predicted_label = torch.argmax(probs).item()
        confidence = probs[predicted_label].item()

        if confidence < threshold:
            # 不确定时走默认安全模型
            return {"model": "default_safe_model", "reason": "low_confidence"}

        model_map = {0: "gpt-3.5-turbo", 1: "gpt-4o"}
        return {
            "model": model_map[predicted_label],
            "reason": "bert_classifier",
            "confidence": confidence
        }

# 使用示例
router = BERTRouter()
result = router.route("帮我设计一个微服务架构方案")
# → {"model": "gpt-4o", "reason": "bert_classifier", "confidence": 0.94}

result = router.route("你好")
# → {"model": "gpt-3.5-turbo", "reason": "bert_classifier", "confidence": 0.97}

BERT 分类器 vs Embedding 分类器

维度 BERT 分类器 Embedding 分类器
原理 端到端训练,直接输出类别 计算与样本库的余弦相似度
训练方式 需要标注数据微调 无需训练,只需构建样本库
准确率 通常更高(学到了判别边界) 依赖样本质量和阈值设置
冷启动 需要标注数据 手写几条样本即可
扩展性 新增类别需重新训练 新增样本即可
延迟 5~20ms 50~200ms(需调 Embedding API)

实际生产系统中,常见做法是分层混合使用:

第一层(BERT 分类器):快速判断粗粒度类别(简单/复杂/代码/安全拦截)
第二层(Embedding 或规则):在粗分类内部做细粒度路由
兜底层(规则)或者 LLM (比如用一个Qwen3.5 9B):处理 BERT 不确定的边界情况
这种架构兼顾了速度和准确性,是目前大模型网关智能路由的主流工程实践。
六、总结

智能路由的本质是在多个模型之间做权衡决策,它结合了传统软件工程中的负载均衡、成本优化和机器学习中的预测与在线学习。从简单的规则到复杂的强化学习,你可以根据实际需求选择合适的复杂度。IDE 中的 “Auto” 选项正是智能路由的一个轻量级应用,它的背后通常也是基于任务类型和成本/延迟的简单判断,但深入下去就是一套完整的系统工程。