路由的四种实现方式：完整代码示例

补充内容，基于《智能体设计模式》第2章扩展
日期：2026-03-31

方式1：基于 LLM 的路由（原书已有，这里精简版）

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 让 LLM 分析意图，只输出类别标识符
router_prompt = ChatPromptTemplate.from_messages([
    ("system", """分析用户请求，只输出一个词：
- 涉及订单、物流、退款 → 'order'
- 涉及价格、产品功能、规格 → 'product'
- 涉及故障、报错、使用方法 → 'support'
- 其他 → 'general'"""),
    ("user", "{request}")
])

router = router_prompt | llm | StrOutputParser()

# 运行
category = router.invoke({"request": "我的快递三天没更新了"})
# 输出: "order"

原理：LLM 理解语义，输出类别标识符，程序根据标识符路由。

方式2：基于嵌入的路由

原理：将用户输入和每个路由目标都转为向量，计算余弦相似度，选最相似的。

from sentence_transformers import SentenceTransformer
import numpy as np

# 加载嵌入模型（本地运行，免费）
model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

# 定义路由目标及其描述
routes = {
    "order": "处理订单查询、物流跟踪、退款退货请求",
    "product": "提供产品信息、价格查询、功能规格介绍",
    "support": "解决技术故障、使用指导、报错排查",
    "general": "一般性问答、闲聊、其他未分类问题"
}

# 预计算每个路由目标的向量（只需算一次）
route_names = list(routes.keys())
route_descriptions = list(routes.values())
route_embeddings = model.encode(route_descriptions)

def embed_route(user_input: str) -> str:
    """将用户输入路由到最相似的目标"""
    # 将用户输入转为向量
    input_embedding = model.encode([user_input])

    # 计算与每个路由目标的余弦相似度
    similarities = np.dot(route_embeddings, input_embedding.T).flatten()

    # 找最相似的路由
    best_idx = np.argmax(similarities)
    best_route = route_names[best_idx]
    best_score = similarities[best_idx]

    # 打印所有相似度（调试用）
    for name, score in zip(route_names, similarities):
        print(f"  {name}: {score:.4f}")

    print(f"  → 路由到: {best_route} (相似度: {best_score:.4f})")
    return best_route

# --- 测试 ---
print("=== 测试1 ===")
embed_route("我的快递三天没更新了")
# order: 0.72, product: 0.31, support: 0.45, general: 0.28
# → 路由到: order

print("\n=== 测试2 ===")
embed_route("这个手机支持5G吗")
# order: 0.25, product: 0.78, support: 0.40, general: 0.22
# → 路由到: product

print("\n=== 测试3 ===")
embed_route("APP打开就闪退怎么办")
# order: 0.18, product: 0.35, support: 0.81, general: 0.30
# → 路由到: support

print("\n=== 测试4 ===")
embed_route("今天天气怎么样")
# order: 0.15, product: 0.12, support: 0.20, general: 0.68
# → 路由到: general

安装依赖：

pip install sentence-transformers numpy

关键点：

  • SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')：支持中文的多语言模型
  • 路由目标的描述要写得准确，这是决定路由质量的关键
  • 余弦相似度范围 [-1, 1]，越接近 1 越相似
  • 可以设置阈值，低于阈值的输入走默认路由

方式3：基于规则的路由

原理：用关键词、正则表达式、if-else 做匹配，零成本、最快。

import re

# 定义规则：每个路由对应一组关键词/正则
routing_rules = {
    "order": {
        "keywords": ["订单", "快递", "物流", "发货", "退款", "退货", "到货", "签收"],
        "patterns": [r"单号\s*\d+", r" Track"]
    },
    "product": {
        "keywords": ["价格", "多少钱", "规格", "功能", "参数", "尺寸", "重量", "颜色"],
        "patterns": [r"\d+GB", r"\d+英寸", r"支持.*吗"]
    },
    "support": {
        "keywords": ["故障", "报错", "闪退", "无法", "不了", "失败", "卡住", "黑屏"],
        "patterns": [r"error", r"Error", r"\d{3,4}错误码"]
    }
}

DEFAULT_ROUTE = "general"

def rule_route(user_input: str) -> str:
    """基于关键词和正则规则的路由"""
    for route_name, rules in routing_rules.items():
        # 检查关键词匹配
        for keyword in rules["keywords"]:
            if keyword in user_input:
                print(f"  命中关键词: '{keyword}' → {route_name}")
                return route_name

        # 检查正则匹配
        for pattern in rules["patterns"]:
            if re.search(pattern, user_input, re.IGNORECASE):
                print(f"  命中正则: '{pattern}' → {route_name}")
                return route_name

    print(f"  未命中任何规则 → {DEFAULT_ROUTE}")
    return DEFAULT_ROUTE

# --- 测试 ---
print("=== 测试1 ===")
rule_route("我的快递三天没更新了，单号SF1234567890")
# 命中关键词: '快递' → order

print("\n=== 测试2 ===")
rule_route("这个手机支持5G吗，屏幕是6.7英寸的")
# 命中关键词: '支持' → product（注：这里"支持"比较泛，实际要优化规则）

print("\n=== 测试3 ===")
rule_route("APP报error代码404了")
# 命中正则: 'error' → support

print("\n=== 测试4 ===")
rule_route("今天杭州天气怎么样")
# 未命中任何规则 → general

关键点：

  • 速度最快，不依赖任何外部服务，零成本
  • 关键词和正则的顺序决定了优先级（先匹配先生效）
  • 关键词可能有歧义（如"支持"既可以是产品问题也可以是其他），需要优化
  • 适合规则明确、类别有限、关键词区分度高的场景

方式4：基于机器学习模型的路由

原理：用标注数据训练一个文本分类器，推理时直接分类，不需要调 LLM。

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report

# --- 1. 准备训练数据（实际项目中需要更多数据，这里用示例）---
training_data = [
    # order 类
    ("我的订单什么时候发货", "order"),
    ("快递到哪了", "order"),
    ("我要退款", "order"),
    ("物流三天没更新", "order"),
    ("查一下我的退货进度", "order"),
    ("帮我修改收货地址", "order"),
    ("发票什么时候开", "order"),

    # product 类
    ("这个手机多少钱", "product"),
    ("支持5G网络吗", "product"),
    ("屏幕尺寸是多大", "product"),
    ("电池续航怎么样", "product"),
    ("有白色款吗", "product"),
    ("和上一代有什么区别", "product"),
    ("处理器是什么型号", "product"),

    # support 类
    ("APP打开就闪退", "support"),
    ("支付失败了怎么办", "support"),
    ("密码忘记怎么重置", "support"),
    ("连接不上蓝牙", "support"),
    ("页面显示404错误", "support"),
    ("下载速度很慢", "support"),
    ("账号被锁定了", "support"),

    # general 类
    ("你好", "general"),
    ("谢谢", "general"),
    ("再见", "general"),
    ("你们公司在哪里", "general"),
    ("今天天气怎么样", "general"),
    ("讲个笑话", "general"),
    ("你们工作时间是什么", "general"),
]

# 分离文本和标签
texts = [item[0] for item in training_data]
labels = [item[1] for item in training_data]

# --- 2. 训练分类器 ---
pipeline = Pipeline([
    ("tfidf", TfidfVectorizer()),           # 文本转向量
    ("clf", LogisticRegression())           # 逻辑回归分类器
])

# 拆分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(
    texts, labels, test_size=0.2, random_state=42
)

# 训练
pipeline.fit(X_train, y_train)

# 评估
predictions = pipeline.predict(X_test)
print("=== 模型评估 ===")
print(classification_report(y_test, predictions))

# --- 3. 使用分类器做路由 ---
def ml_route(user_input: str) -> str:
    """用训练好的分类器做路由"""
    prediction = pipeline.predict([user_input])[0]
    # 获取每个类别的概率
    probabilities = pipeline.predict_proba([user_input])[0]
    class_names = pipeline.classes_

    # 打印每个类别的概率
    for name, prob in zip(class_names, probabilities):
        print(f"  {name}: {prob:.4f}")

    print(f"  → 路由到: {prediction}")
    return prediction

# --- 测试 ---
print("\n=== 测试1 ===")
ml_route("我的包裹到了没")
# general: 0.15, order: 0.72, product: 0.08, support: 0.05
# → 路由到: order

print("\n=== 测试2 ===")
ml_route("这个笔记本内存可以升级吗")
# general: 0.10, order: 0.12, product: 0.65, support: 0.13
# → 路由到: product

print("\n=== 测试3 ===")
ml_route("登录的时候一直转圈")
# general: 0.08, order: 0.05, product: 0.15, support: 0.72
# → 路由到: support

安装依赖：

pip install scikit-learn

关键点：

  • TF-IDF + 逻辑回归是最经典的文本分类方案，简单有效
  • 推理速度极快（毫秒级），不需要 GPU
  • 可以用 LLM 生成大量合成训练数据，降低人工标注成本
  • 实际项目中可以用 BERT 等预训练模型替换 TF-IDF，效果更好
  • predict_proba() 输出概率分布，可以设置置信度阈值

四种方式对比总结

对比维度 LLM 路由 嵌入路由 规则路由 ML 分类路由
速度 慢（秒级） 中（百毫秒） 快（微秒） 快（毫秒）
成本 高（每次调 API） 中（需模型） 零 零（推理时）
灵活性 ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐ ⭐⭐ ⭐⭐⭐
语义理解 ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐ ⭐ ⭐⭐⭐
开发成本 低 中 低 中（需标注数据）
维护成本 低 低 高（规则膨胀） 低
适用场景 复杂语义判断 语义相似度匹配 简单关键词分流 高频、稳定场景

实际建议：

  • 快速原型：用 LLM 路由，开发最快
  • 生产环境高频调用：训练 ML 分类器，速度快成本低
  • 需要语义理解但不想每次调 LLM：用嵌入路由
  • 规则明确的简单场景：用规则路由，最快最稳
  • 最佳实践：规则路由做第一层快速拦截 + LLM/ML 路由做第二层精细分类
# 学习
Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

Claude API + Python实战：结构化输出后，如何拦住“看起来正确”的JSON

avatar AI编程社区
cover

2026年8月开发者实战指南：ChatGPT Plus / Pro + Codex 如何协同完成需求分析、编码、测试、调试与代码审查

avatar AI编程社区
cover

投资一台B300，部署Claude Opus 5多久才能回本？

avatar AI编程社区