成本优化秘籍：如何通过模型路由（Model Routing）降低 Agent 推理成本？
成本优化秘籍：如何通过模型路由（Model Router）降低 Agent 推理成本？
副标题：从理论到实践，构建智能路由系统，让你的 AI Agent 既省钱又高效
第一部分：引言与基础
1. 摘要/引言
问题陈述
在当今的 AI 应用开发中，大语言模型（LLMs）已成为构建智能 Agent 的核心组件。然而，随着模型能力的增强，其推理成本也呈指数级增长。想象一下：如果你的 Agent 每天处理数百万个查询，全部使用 GPT-4 或 Claude 3 Opus，那账单数字可能会让你的财务部门心惊肉跳。但另一方面，全部使用廉价小模型又可能导致回答质量下降，影响用户体验。这就是我们面临的核心困境：如何在成本和质量之间找到最佳平衡点？
核心方案
答案就是模型路由（Model Routing）。简单来说，模型路由就是一个智能决策系统，它能根据查询的复杂度、类型、用户需求等因素，自动将不同的请求分发到最合适的模型上。简单查询用小模型，复杂查询用大模型，专业查询用微调模型。通过这种方式，我们可以在保持甚至提升整体服务质量的同时，显著降低推理成本。
主要成果/价值
读完这篇文章，你将：
- 深入理解模型路由的核心概念、架构和工作原理
- 掌握设计和实现模型路由系统的关键技术
- 了解不同路由策略的优缺点及适用场景
- 获得一个可直接运行的 Python 实现示例
- 学会如何评估和优化你的路由系统
- 掌握成本分析和 ROI 计算的方法
文章导览
我们将从问题背景出发，逐步深入模型路由的核心概念，然后详细介绍不同的路由策略，接着通过一个完整的项目示例展示如何实现一个实用的模型路由系统，最后讨论性能优化、最佳实践和未来发展趋势。
2. 目标读者与前置知识
目标读者
- AI/ML 工程师：希望优化 LLM 应用成本的开发者
- 后端/系统架构师：负责设计和构建可扩展 AI 系统的专业人士
- 产品经理/技术负责人：关注 AI 应用成本效益的决策者
- 对 LLM 应用开发感兴趣的开发者：希望了解企业级 LLM 应用最佳实践的学习者
前置知识
- 基本的 Python 编程能力
- 对大语言模型（LLMs）及其 API 使用有一定了解
- 基本的系统设计概念
- （可选）对机器学习和自然语言处理有基础认识
3. 文章目录
-
引言与基础
- 摘要/引言
- 目标读者与前置知识
- 文章目录
-
问题背景与动机
- LLM 推理成本的构成与现状
- 现有成本优化方法的局限性
- 为什么模型路由是关键解决方案
-
核心概念与理论基础
- 模型路由的定义与核心要素
- 路由策略分类与对比
- 质量-成本权衡理论
- 相关概念：模型分级、提示工程、缓存等
-
环境准备
- 所需工具与库
- 安装与配置指南
- 项目结构设计
-
分步实现：基础路由系统
- 简单规则路由实现
- 基于语义相似度的路由
- 基于学习的路由系统
- 多模型集成与调用封装
-
关键代码解析与深度剖析
- 路由决策算法详解
- 模型适配器设计模式
- 错误处理与降级策略
- 监控与日志系统设计
-
结果展示与验证
- 成本对比测试
- 质量评估方法
- A/B 测试结果分析
-
性能优化与最佳实践
- 路由缓存策略
- 批量处理优化
- 模型性能监控
- 成本分配与标签化
-
常见问题与解决方案
- 路由错误处理
- 模型延迟波动应对
- 成本超支预警
- 质量下降检测与修正
-
未来展望与扩展方向
- 自适应路由系统
- 多模态模型路由
- 联邦学习与隐私保护
- 与 MLOps 平台集成
-
总结
-
参考资料
-
附录
第二部分：核心内容
4. 问题背景与动机
4.1 LLM 推理成本的构成与现状
在深入探讨解决方案之前，让我们先彻底理解问题的本质。LLM 推理成本主要由以下几个因素构成：
-
Token 计费模型：大多数 LLM 服务提供商（如 OpenAI、Anthropic、Google）都采用基于 Token 的计费方式。输入 Token 和输出 Token 通常分开计费，且输出 Token 价格更高。
-
模型能力与价格正相关：一般来说，模型能力越强，价格越高。例如，GPT-4 的价格是 GPT-3.5-turbo 的 10-30 倍。
-
延迟成本：除了直接的金钱成本，我们还需要考虑延迟成本。不同模型的推理速度差异很大，这会直接影响用户体验。
让我们来看一些具体的数字（截至 2024 年中期）：
| 模型 | 输入 Token 价格 (每百万) | 输出 Token 价格 (每百万) | 相对价格 (与 GPT-3.5 比) |
|---|---|---|---|
| GPT-3.5-turbo | $0.50 | $1.50 | 1x |
| GPT-4 (8K) | $30.00 | $60.00 | 30x |
| GPT-4 Turbo (128K) | $10.00 | $30.00 | 15x |
| Claude 3 Haiku | $0.25 | $1.25 | 0.6x |
| Claude 3 Sonnet | $3.00 | $15.00 | 5x |
| Claude 3 Opus | $15.00 | $75.00 | 35x |
| Llama 3 8B (自托管) | ~$0.05 | ~$0.15 | 0.08x |
| Llama 3 70B (自托管) | ~$0.50 | ~$1.50 | 1x |
现在，让我们做一个简单的计算。假设你的 Agent 每天处理 100,000 个查询，平均每个查询包含 500 个输入 Token 和 500 个输出 Token：
- 全部使用 GPT-3.5-turbo：每天成本 = (100,000 * 500 / 1,000,000 * $0.50) + (100,000 * 500 / 1,000,000 * $1.50) = $25 + $75 = $100/天 = $3,000/月
- 全部使用 GPT-4：每天成本 = (100,000 * 500 / 1,000,000 * $30) + (100,000 * 500 / 1,000,000 * $60) = $1,500 + $3,000 = $4,500/天 = $135,000/月
这是 45 倍的成本差异！而且这还没有考虑到 GPT-4 的推理延迟通常比 GPT-3.5 高 2-3 倍。
但问题是，我们真的需要在所有查询上都使用 GPT-4 吗？根据我的经验，在大多数实际应用中，只有 10-20% 的查询真正需要 GPT-4 级别的能力，剩下的 80-90% 完全可以用更小、更便宜的模型处理。这就是模型路由能为我们节省大量成本的理论基础。
4.2 现有成本优化方法的局限性
在模型路由出现之前，人们已经尝试了多种方法来降低 LLM 推理成本。让我们来逐一分析这些方法及其局限性：
-
提示工程优化
- 方法：通过优化提示词，减少不必要的 Token 消耗，提高模型效率
- 局限性：收益有限，通常只能节省 10-20% 的成本，且过度优化可能影响模型性能
-
输出约束
- 方法：通过参数（如
max_tokens、temperature）或提示词限制输出长度 - 局限性：可能影响回答的完整性和质量，且对输入 Token 成本没有影响
- 方法：通过参数（如
-
自托管开源模型
- 方法：部署 Llama、Mistral 等开源模型，按服务器成本而非 API 调用付费
- 局限性：需要大量的工程投入和维护成本，且模型能力可能不如商业模型
-
缓存机制
- 方法：缓存常见查询的响应，避免重复计算
- 局限性：只对重复查询有效，对新的、独特的查询无能为力
-
批量处理
- 方法：将多个查询合并为一个批次处理
- 局限性：不适用于实时交互场景，增加了系统复杂度
这些方法各有其价值，但都没有解决根本问题：根据查询的实际需求选择合适的模型。这就是模型路由的用武之地。
4.3 为什么模型路由是关键解决方案
模型路由之所以成为降低 Agent 推理成本的关键解决方案，是因为它直接针对成本问题的核心：能力过剩。
在大多数 LLM 应用中，我们面临的查询复杂度分布通常呈现长尾特征：
- 头部（10-20%）：复杂、高价值的查询，需要最强模型的能力
- 中部（30-40%）：中等复杂度的查询，中型模型即可处理
- 尾部（40-60%）：简单查询，小型模型甚至规则引擎就能搞定
模型路由的核心思想就是匹配复杂度与模型能力，让每个查询都由“刚刚好”的模型来处理，既不浪费能力，也不牺牲质量。
模型路由的优势包括：
- 显著的成本降低：根据实际场景，通常可以降低 50-80% 的推理成本
- 质量保持或提升：通过将复杂查询专门路由给大模型，反而可能提升整体质量
- 灵活性：可以轻松添加、移除或替换模型，适应技术发展和业务变化
- 渐进式优化：可以从简单的策略开始，逐步迭代优化
- 综合优化：可以与其他优化方法（如缓存、提示工程）结合使用，发挥最大效果
在接下来的章节中，我们将深入探讨如何设计和实现一个有效的模型路由系统。
5. 核心概念与理论基础
在开始实现之前，让我们先建立一些核心概念的共同理解。这将帮助我们更好地设计系统和做出技术决策。
5.1 模型路由的定义与核心要素
模型路由（Model Routing） 是一种智能决策机制，它根据查询的特征和上下文，自动将请求分发到最合适的模型或处理单元，以优化成本、性能和质量的平衡。
一个完整的模型路由系统通常包含以下核心要素：
- 查询分析器（Query Analyzer）：负责解析和提取查询的特征，如复杂度、主题、情感、语言等
- 路由策略引擎（Routing Strategy Engine）：根据查询特征和预定义策略，决定将查询路由到哪个模型
- 模型池（Model Pool）：包含多个可供选择的模型，每个模型都有其特定的能力、成本和性能特征
- 执行引擎（Execution Engine）：负责调用选定的模型，处理响应，并可能进行后处理
- 反馈与学习系统（Feedback & Learning System）：收集路由结果和用户反馈，用于优化路由策略
- 监控与日志系统（Monitoring & Logging）：记录所有路由决策、成本、性能和质量指标，用于分析和优化
这些要素之间的关系可以用下面的架构图来表示：
5.2 路由策略分类与对比
路由策略是模型路由系统的核心。根据决策方式的不同，我们可以将路由策略分为以下几类：
5.2.1 基于规则的路由（Rule-based Routing）
这是最简单直接的路由策略，基于预定义的硬编码规则进行决策。
常见规则包括：
- 关键词匹配：如果查询包含特定关键词（如"代码"、“分析”），路由到大模型
- 查询长度：如果查询超过一定长度，路由到大模型
- 用户类型：VIP 用户或付费用户使用大模型
- 时间/负载：高峰期使用小模型，低峰期使用大模型
- 语言/地区：根据用户语言或地区选择专门优化的模型
优点：
- 实现简单，易于理解和调试
- 完全可控，行为可预测
- 无需训练数据
- 执行速度快
缺点：
- 规则可能无法覆盖所有情况
- 难以处理模糊或复杂的边界情况
- 维护成本高，规则可能变得复杂且难以管理
- 无法自动适应变化
适用场景：
- 查询模式相对固定的应用
- 初期快速原型开发
- 需要严格控制路由行为的合规场景
5.2.2 基于语义相似度的路由（Semantic Similarity Routing）
这种策略首先将查询转换为向量嵌入，然后计算其与预定义样本查询的相似度，根据相似度进行路由决策。
工作流程：
- 准备一组样本查询，每个样本都标注了应该使用的模型
- 将所有样本查询转换为向量嵌入，存储在向量数据库中
- 对于新查询，将其转换为向量嵌入
- 在向量数据库中查找最相似的样本查询
- 根据最相似样本的标注决定路由
优点：
- 可以理解查询的语义，而不仅仅是关键词
- 相对容易扩展，只需添加新的样本查询
- 可以处理同义词和语义变体
缺点：
- 需要准备和维护样本查询库
- 嵌入模型本身也有成本和延迟
- 对于完全新类型的查询可能效果不佳
- 相似度阈值的设置可能需要反复调优
适用场景：
- 查询类型相对有限，但表达方式多样的应用
- 有一定历史数据可以利用的场景
5.2.3 基于学习的路由（Learning-based Routing）
这种策略使用机器学习模型来预测哪个模型最适合处理给定的查询。
常见方法：
- 分类模型：将路由决策视为多分类问题，训练模型预测最佳模型
- 强化学习：通过试错和奖励机制优化路由策略
- 元学习：学习如何学习，快速适应新的模型和查询类型
优点：
- 可以自动发现复杂的模式和规律
- 可以持续优化和适应变化
- 通常能获得最佳的整体性能
缺点：
- 实现复杂度高
- 需要大量高质量的标注数据
- 训练和维护成本高
- 决策过程可能不透明，难以调试
适用场景：
- 大规模、高流量的应用
- 查询模式复杂且多变的场景
- 有资源投入到模型训练和维护的团队
5.2.4 混合策略（Hybrid Strategy）
大多数实际系统不会只使用单一策略，而是将多种策略结合起来，形成混合策略。
常见组合方式：
- 分层路由：先用简单规则处理明显的情况，再用更复杂的策略处理剩余情况
- 投票机制：多个策略并行工作，通过投票决定最终路由
- 级联降级：主策略失败时，使用备用策略
为了更清晰地对比这些策略，我们可以看下面的表格：
| 策略类型 | 实现复杂度 | 维护成本 | 决策质量 | 可解释性 | 数据需求 | 最佳适用场景 |
|---|---|---|---|---|---|---|
| 基于规则 | 低 | 中高 | 中 | 高 | 无 | 简单场景、原型开发 |
| 基于语义 | 中 | 中 | 中高 | 中 | 中 | 语义变化多、类型有限 |
| 基于学习 | 高 | 高 | 高 | 低 | 高 | 大规模、复杂场景 |
| 混合策略 | 中高 | 中高 | 高 | 中 | 中高 | 大多数生产环境 |
5.3 质量-成本权衡理论
模型路由的核心是在质量和成本之间找到最佳平衡点。让我们用一些数学概念来形式化这个问题。
首先，让我们定义一些变量：
- 设 QQQ 为质量指标，通常可以通过人工评估、自动评估指标（如 BLEU、ROUGE、METEOR）或用户满意度来衡量
- 设 CCC 为成本指标，通常以 Token 消耗、API 调用费用或计算资源使用量来衡量
- 设 M={m1,m2,...,mn}M = \{m_1, m_2, ..., m_n\}M={m1,m2,...,mn} 为可用模型集合
- 设 q(mi)q(m_i)q(mi) 为模型 mim_imi 的质量得分
- 设 c(mi)c(m_i)c(mi) 为模型 mim_imi 的成本
- 设 XXX 为查询空间，x∈Xx \in Xx∈X 为一个具体的查询
- 设 r:X→Mr: X \rightarrow Mr:X→M 为路由函数，将查询映射到模型
我们的目标是找到一个路由函数 rrr，使得在满足一定质量约束的前提下，总成本最小，或者在一定成本预算下，总质量最高。
形式化地说，我们可以将其表示为一个约束优化问题：
minr∑x∈Xc(r(x))⋅p(x) \min_{r} \sum_{x \in X} c(r(x)) \cdot p(x) rminx∈X∑c(r(x))⋅p(x)
subject to: ∑x∈Xq(r(x),x)⋅p(x)≥Qmin \text{subject to: } \sum_{x \in X} q(r(x), x) \cdot p(x) \geq Q_{\text{min}} subject to: x∈X∑q(r(x),x)⋅p(x)≥Qmin
其中 p(x)p(x)p(x) 是查询 xxx 的概率分布，QminQ_{\text{min}}Qmin 是可接受的最低质量水平。
或者，我们也可以将其表示为一个带权重的单目标优化问题：
maxr∑x∈X[α⋅q(r(x),x)−(1−α)⋅c(r(x))]⋅p(x) \max_{r} \sum_{x \in X} \left[ \alpha \cdot q(r(x), x) - (1-\alpha) \cdot c(r(x)) \right] \cdot p(x) rmaxx∈X∑[α⋅q(r(x),x)−(1−α)⋅c(r(x))]⋅p(x)
其中 α∈[0,1]\alpha \in [0, 1]α∈[0,1] 是质量和成本之间的权衡参数。α\alphaα 越大，我们越重视质量；α\alphaα 越小，我们越重视成本。
在实际应用中，我们通常不会直接求解这个优化问题，而是通过一些启发式方法来近似最优解。模型路由的各种策略本质上都是在尝试找到这个优化问题的良好近似解。
5.4 相关概念：模型分级、提示工程、缓存等
模型路由不是孤立存在的，它通常与其他优化技术结合使用，以达到最佳效果。让我们简要了解一些相关概念：
5.4.1 模型分级（Model Tiering）
模型分级是模型路由的前提，它将可用模型按照能力、成本和性能分为不同的等级。
典型的分级方式：
- Tier 1（顶级）：最强模型，最高成本，用于最复杂、最高价值的查询
- Tier 2（中级）：平衡模型，中等成本，用于一般复杂度的查询
- Tier 3（基础）：小模型，最低成本，用于简单查询
- Tier 0（规则）：规则引擎或模板，零边际成本，用于完全可预测的查询
5.4.2 提示工程（Prompt Engineering）
虽然提示工程本身不能替代模型路由，但它可以与路由结合使用，进一步优化成本和质量。
与路由结合的提示工程策略：
- 为不同等级的模型设计专门优化的提示词
- 小模型使用更详细、更具体的提示词，以弥补能力不足
- 大模型使用更简洁的提示词，以节省 Token
- 使用思维链（Chain-of-Thought）提示词提升小模型在特定任务上的表现
5.4.3 缓存机制（Caching）
缓存是另一种重要的成本优化技术，它可以与模型路由协同工作。
与路由结合的缓存策略：
- 语义缓存：缓存相似查询的响应，无论使用哪个模型生成
- 分层缓存：小模型的响应缓存时间较短，大模型的响应缓存时间较长
- 路由决策缓存：对于相似查询，直接使用之前的路由决策，而不重新计算
5.4.4 降级策略（Fallback Strategy）
降级策略是系统可靠性的重要保障，它定义了当首选模型不可用时应该怎么做。
典型的降级策略：
- 模型级降级：如果大模型不可用，降级到中等模型，再降级到小模型
- 质量级降级：如果无法保证质量，提供简化版回答或错误提示
- 功能级降级：如果高级功能不可用，只提供基本功能
这些概念相互关联，共同构成了一个完整的 LLM 成本优化工具箱。在接下来的章节中，我们将看到如何将它们整合到一个实际的模型路由系统中。
6. 环境准备
在开始实现我们的模型路由系统之前，让我们先准备好开发环境。我们将使用 Python 作为主要编程语言，因为它在 AI/ML 领域有最丰富的生态系统。
6.1 所需工具与库
我们将使用以下工具和库：
| 工具/库 | 版本要求 | 用途 |
|---|---|---|
| Python | 3.9+ | 主要编程语言 |
| OpenAI SDK | 最新 | 调用 OpenAI 模型 |
| Anthropic SDK | 最新 | 调用 Anthropic 模型 |
| LangChain | 最新 | LLM 应用开发框架 |
| ChromaDB | 最新 | 向量数据库 |
| Sentence-Transformers | 最新 | 生成文本嵌入 |
| Pydantic | 最新 | 数据验证和设置管理 |
| FastAPI | 最新 | API 服务框架 |
| Uvicorn | 最新 | ASGI 服务器 |
| python-dotenv | 最新 | 环境变量管理 |
| pytest | 最新 | 单元测试 |
| matplotlib | 最新 | 数据可视化 |
| pandas | 最新 | 数据处理 |
6.2 安装与配置指南
让我们一步步设置我们的开发环境：
6.2.1 基础环境设置
首先，确保你安装了 Python 3.9 或更高版本：
python --version
我们推荐使用虚拟环境来隔离项目依赖：
# 创建虚拟环境
python -m venv model-router-env
# 激活虚拟环境（Windows）
model-router-env\Scripts\activate
# 激活虚拟环境（macOS/Linux）
source model-router-env/bin/activate
6.2.2 安装依赖
创建一个 requirements.txt 文件，包含所有需要的依赖：
openai>=1.0.0
anthropic>=0.20.0
langchain>=0.1.0
langchain-openai>=0.0.5
langchain-anthropic>=0.1.0
chromadb>=0.4.0
sentence-transformers>=2.2.0
pydantic>=2.0.0
fastapi>=0.100.0
uvicorn>=0.23.0
python-dotenv>=1.0.0
pytest>=7.4.0
matplotlib>=3.7.0
pandas>=2.0.0
然后安装这些依赖：
pip install -r requirements.txt
6.2.3 API 密钥配置
我们需要配置一些 API 密钥来访问各种 LLM 服务。创建一个 .env 文件：
# OpenAI API
OPENAI_API_KEY=your_openai_api_key_here
# Anthropic API
ANTHROPIC_API_KEY=your_anthropic_api_key_here
# 应用配置
APP_NAME=ModelRouter
APP_VERSION=1.0.0
DEBUG=True
请确保将 your_openai_api_key_here 和 your_anthropic_api_key_here 替换为你的实际 API 密钥。
6.3 项目结构设计
一个好的项目结构对于代码的可维护性和可扩展性至关重要。让我们设计以下项目结构：
model-router/
├── .env # 环境变量
├── .gitignore # Git 忽略文件
├── requirements.txt # 依赖列表
├── README.md # 项目说明
├── app/ # 应用代码
│ ├── __init__.py
│ ├── main.py # FastAPI 应用入口
│ ├── config.py # 配置管理
│ ├── models/ # 数据模型
│ │ ├── __init__.py
│ │ ├── request.py # 请求模型
│ │ ├── response.py # 响应模型
│ │ └── routing.py # 路由相关模型
│ ├── router/ # 路由系统核心
│ │ ├── __init__.py
│ │ ├── base.py # 基础路由器类
│ │ ├── rule_based.py # 基于规则的路由器
│ │ ├── semantic.py # 基于语义的路由器
│ │ ├── learning.py # 基于学习的路由器
│ │ └── hybrid.py # 混合路由器
│ ├── llm/ # LLM 集成
│ │ ├── __init__.py
│ │ ├── base.py # 基础 LLM 适配器
│ │ ├── openai_adapter.py # OpenAI 适配器
│ │ ├── anthropic_adapter.py # Anthropic 适配器
│ │ └── model_pool.py # 模型池管理
│ ├── analysis/ # 分析和监控
│ │ ├── __init__.py
│ │ ├── cost_tracker.py # 成本追踪
│ │ ├── quality_evaluator.py # 质量评估
│ │ └── reporting.py # 报告生成
│ └── utils/ # 工具函数
│ ├── __init__.py
│ ├── cache.py # 缓存工具
│ ├── embedding.py # 嵌入生成
│ └── token_counter.py # Token 计数
├── data/ # 数据目录
│ ├── sample_queries/ # 样本查询
│ └── results/ # 结果存储
├── tests/ # 测试代码
│ ├── __init__.py
│ ├── test_router.py # 路由器测试
│ ├── test_llm_adapters.py # LLM 适配器测试
│ └── test_integration.py # 集成测试
└── notebooks/ # Jupyter 笔记本
├── 01_exploration.ipynb # 数据探索
├── 02_strategy_comparison.ipynb # 策略对比
└── 03_optimization.ipynb # 优化分析
这个结构将我们的代码组织成逻辑清晰的模块，便于开发和维护。在接下来的章节中，我们将逐步实现这些模块。
7. 分步实现：基础路由系统
现在，让我们开始实现我们的模型路由系统。我们将从最简单的基于规则的路由开始，然后逐步添加更复杂的功能。
7.1 基础设置与配置管理
首先，让我们实现配置管理模块，这将帮助我们集中管理应用的各种配置。
在 app/config.py 中：
from pydantic_settings import BaseSettings
from typing import List, Dict, Any
import os
class Settings(BaseSettings):
"""应用配置管理"""
# 基础配置
app_name: str = "ModelRouter"
app_version: str = "1.0.0"
debug: bool = True
# API 密钥
openai_api_key: str
anthropic_api_key: str
# 模型配置
model_tiers: Dict[str, Dict[str, Any]] = {
"tier_1": {
"models": ["gpt-4", "claude-3-opus-20240229"],
"cost_multiplier": 1.0,
"quality_threshold": 0.9
},
"tier_2": {
"models": ["gpt-3.5-turbo", "claude-3-sonnet-20240229"],
"cost_multiplier": 0.3,
"quality_threshold": 0.7
},
"tier_3": {
"models": ["claude-3-haiku-20240307"],
"cost_multiplier": 0.1,
"quality_threshold": 0.5
}
}
# 路由配置
default_tier: str = "tier_2"
routing_strategy: str = "hybrid" # rule_based, semantic, learning, hybrid
# 缓存配置
cache_enabled: bool = True
cache_ttl: int = 3600 # 秒
# 监控配置
monitoring_enabled: bool = True
log_level: str = "INFO"
class Config:
env_file = ".env"
case_sensitive = False
# 创建全局配置实例
settings = Settings()
接下来，让我们定义一些基础的数据模型。在 app/models/request.py 中：
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
class ChatMessage(BaseModel):
"""聊天消息模型"""
role: str = Field(..., description="消息角色，如 'user', 'assistant', 'system'")
content: str = Field(..., description="消息内容")
class InferenceRequest(BaseModel):
"""推理请求模型"""
messages: List[ChatMessage] = Field(..., description="聊天消息列表")
model: Optional[str] = Field(None, description="指定模型，不指定则由路由器决定")
max_tokens: Optional[int] = Field(None, description="最大生成 Token 数")
temperature: Optional[float] = Field(None, description="温度参数，控制随机性")
user_id: Optional[str] = Field(None, description="用户 ID，用于个性化路由")
metadata: Optional[Dict[str, Any]] = Field(None, description="附加元数据")
在 app/models/response.py 中：
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
from datetime import datetime
class InferenceResponse(BaseModel):
"""推理响应模型"""
id: str = Field(..., description="响应 ID")
content: str = Field(..., description="生成的内容")
model: str = Field(..., description="使用的模型")
created: datetime = Field(..., description="创建时间")
usage: Dict[str, int] = Field(..., description="Token 使用情况")
routing_decision: Dict[str, Any] = Field(..., description="路由决策信息")
latency_ms: float = Field(..., description="响应延迟（毫秒）")
在 app/models/routing.py 中：
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
from enum import Enum
class RoutingStrategy(str, Enum):
"""路由策略枚举"""
RULE_BASED = "rule_based"
SEMANTIC = "semantic"
LEARNING = "learning"
HYBRID = "hybrid"
class RoutingDecision(BaseModel):
"""路由决策模型"""
strategy: RoutingStrategy = Field(..., description="使用的路由策略")
selected_model: str = Field(..., description="选择的模型")
selected_tier: str = Field(..., description="选择的模型层级")
confidence: float = Field(..., description="决策置信度，0-1")
reasoning: Optional[str] = Field(None, description="决策理由")
factors: Dict[str, Any] = Field(default_factory=dict, description="影响决策的因素")
alternatives: List[str] = Field(default_factory=list, description="备选模型")
有了这些基础设置，我们就可以开始实现核心的路由逻辑了。
7.2 基于规则的路由实现
让我们首先实现最简单但也最实用的基于规则的路由器。
在 app/router/base.py 中，我们定义一个基础路由器类：
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
from app.models.request import InferenceRequest
from app.models.routing import RoutingDecision, RoutingStrategy
from app.config import settings
class BaseRouter(ABC):
"""路由器基类"""
def __init__(self):
self.name = "BaseRouter"
self.strategy = RoutingStrategy.RULE_BASED # 默认策略
self.model_tiers = settings.model_tiers
self.default_tier = settings.default_tier
@abstractmethod
def route(self, request: InferenceRequest) -> RoutingDecision:
"""
核心路由方法，子类必须实现
Args:
request: 推理请求
Returns:
路由决策
"""
pass
def _get_available_models(self, tier: Optional[str] = None) -> List[str]:
"""
获取可用模型列表
Args:
tier: 可选，指定层级
Returns:
模型列表
"""
if tier:
return self.model_tiers.get(tier, {}).get("models", [])
all_models = []
for tier_config in self.model_tiers.values():
all_models.extend(tier_config.get("models", []))
return all_models
def _select_model_from_tier(self, tier: str) -> str:
"""
从指定层级选择一个模型（这里简单选择第一个，实际可实现负载均衡）
Args:
tier: 模型层级
Returns:
选中的模型
"""
models = self._get_available_models(tier)
return models[0] if models else self._select_model_from_tier(self.default_tier)
现在，让我们在 app/router/rule_based.py 中实现基于规则的路由器：
from typing import Any, Dict, List, Optional
import re
from app.models.request import InferenceRequest, ChatMessage
from app.models.routing import RoutingDecision, RoutingStrategy
from app.router.base import BaseRouter
class RuleBasedRouter(BaseRouter):
"""基于规则的路由器"""
def __init__(self):
super().__init__()
self.name = "RuleBasedRouter"
self.strategy = RoutingStrategy.RULE_BASED
# 预定义规则
self.complexity_keywords = {
"high": [
"分析", "代码", "编写", "设计", "优化", "复杂", "高级",
"推理", "计算", "研究", "论文", "报告", "策略", "规划"
],
"medium": [
"解释", "比较", "区别", "如何", "怎样", "为什么", "建议",
"步骤", "方法", "过程", "示例", "说明"
],
"low": [
"什么是", "定义", "意思", "简单", "基本", "常见",
"问候", "你好", "谢谢", "再见", "日期", "时间"
]
}
# 编译正则表达式以提高性能
self.complexity_patterns = {
level: re.compile("|".join(keywords), re.IGNORECASE)
for level, keywords in self.complexity_keywords.items()
}
# 查询长度阈值
self.length_thresholds = {
"high": 500,
"medium": 200
}
def route(self, request: InferenceRequest) -> RoutingDecision:
"""
基于规则的路由决策
Args:
request: 推理请求
Returns:
路由决策
"""
# 如果用户明确指定了模型，直接使用
if request.model:
tier = self._get_tier_for_model(request.model)
return RoutingDecision(
strategy=self.strategy,
selected_model=request.model,
selected_tier=tier,
confidence=1.0,
reasoning="用户明确指定模型",
factors={"user_specified": True}
)
# 提取最后一条用户消息
last_user_message = self._get_last_user_message(request.messages)
if not last_user_message:
# 如果没有用户消息，使用默认层级
model = self._select_model_from_tier(self.default_tier)
return RoutingDecision(
strategy=self.strategy,
selected_model=model,
selected_tier=self.default_tier,
confidence=0.5,
reasoning="无有效用户消息，使用默认模型",
factors={"fallback": True}
)
# 应用规则进行决策
decision_factors = {}
complexity_level = self._assess_complexity(last_user_message.content, decision_factors)
selected_tier = self._complexity_to_tier(complexity_level)
selected_model = self._select_model_from_tier(selected_tier)
# 计算置信度
confidence = self._calculate_confidence(decision_factors)
return RoutingDecision(
strategy=self.strategy,
selected_model=selected_model,
selected_tier=selected_tier,
confidence=confidence,
reasoning=f"基于规则评估，复杂度级别: {complexity_level}",
factors=decision_factors
)
def _get_last_user_message(self, messages: List[ChatMessage]) -> Optional[ChatMessage]:
"""获取最后一条用户消息"""
for message in reversed(messages):
if message.role == "user":
return message
return None
def _assess_complexity(self, text: str, factors: Dict[str, Any]) -> str:
"""评估查询复杂度"""
# 检查关键词
keyword_level = "low"
for level, pattern in self.complexity_patterns.items():
if pattern.search(text):
keyword_level = level
# 高级别关键词覆盖低级别
if level == "high":
break
factors["keyword_level"] = keyword_level
# 检查长度
text_length = len(text)
factors["text_length"] = text_length
length_level = "low"
if text_length >= self.length_thresholds["high"]:
length_level = "high"
elif text_length >= self.length_thresholds["medium"]:
length_level = "medium"
factors["length_level"] = length_level
# 综合评估
# 如果任一条件为 high，整体为 high
if keyword_level == "high" or length_level == "high":
return "high"
# 如果任一条件为 medium，整体为 medium
elif keyword_level == "medium" or length_level == "medium":
return "medium"
# 否则为 low
else:
return "low"
def _complexity_to_tier(self, complexity_level: str) -> str:
"""将复杂度级别映射到模型层级"""
mapping = {
"high": "tier_1",
"medium": "tier_2",
"low": "tier_3"
}
return mapping.get(complexity_level, self.default_tier)
def _get_tier_for_model(self, model: str) -> str:
"""获取模型所属的层级"""
for tier, config in self.model_tiers.items():
if model in config.get("models", []):
return tier
return self.default_tier
def _calculate_confidence(self, factors: Dict[str, Any]) -> float:
"""计算决策置信度"""
confidence = 0.5 # 基础置信度
# 如果关键词和长度评估一致，增加置信度
if factors.get("keyword_level") == factors.get("length_level"):
confidence += 0.3
# 文本越长，我们对长度评估越有信心
text_length = factors.get("text_length", 0)
if text_length > 100:
confidence += min(0.2, text_length / 1000)
return min(1.0, confidence)
这个基于规则的路由器已经具备了基本的功能，它可以根据关键词和查询长度来评估复杂度，并选择合适的模型层级。但我们还可以做得更好，让我们继续实现基于语义的路由器。
7.3 基于语义相似度的路由
基于规则的路由虽然简单直观，但它只能理解表面特征，无法真正理解查询的语义。接下来，让我们实现一个基于语义相似度的路由器。
首先，在 app/utils/embedding.py 中实现嵌入生成工具：
from typing import List, Optional
import numpy as np
from sentence_transformers import SentenceTransformer
class EmbeddingGenerator:
"""文本嵌入生成器"""
def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
"""
初始化嵌入生成器
Args:
model_name: Sentence-Transformers 模型名称
"""
self.model_name = model_name
self.model = None
self._load_model()
def _load_model(self):
"""加载模型"""
try:
self.model = SentenceTransformer(self.model_name)
print(f"成功加载嵌入模型: {self.model_name}")
except Exception as e:
print(f"加载嵌入模型失败: {e}")
raise
def generate_embedding(self, text: str) -> np.ndarray:
"""
生成单条文本的嵌入
Args:
text: 输入文本
Returns:
文本嵌入向量
"""
if not self.model:
raise RuntimeError("嵌入模型未加载")
return self.model.encode(text, convert_to_numpy=True)
def generate_embeddings(self, texts: List[str]) -> np.ndarray:
"""
批量生成文本嵌入
Args:
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
8
0- 0
已为社区贡献140条内容
所有评论(0)