目录1. 概述2. 价值3. 关键技术4. 代码实现5. 代码说明核心组件1. 路由类型枚举RouteType2. 路由决策结果RoutingDecision3. 路由器基类Router4. LLM路由器LLMRouter5. 规则路由器RuleBasedRouter6. 混合路由器HybridRouter7. 路由执行器RouteExecutor主要特点1. 多路由策略支持2. 灵活的扩展性3. 完整的执行流程4. 置信度评分机制5. 历史管理和分析路由策略说明1. LLM路由的优势2. 规则路由的优势3. 混合路由的优势使用建议1. 概述智能体系统通常需要根据环境状态、用户输入或前序操作结果等因素在多个潜在动作之间进行仲裁这种动态决策能力即根据特定条件将控制流导向不同的专用函数、工具或子流程是通过路由机制实现的。路由为智能体的操作框架引入了条件逻辑使其从固定执行路径转变为动态评估特定标准、从一组可能的后续动作中进行选择的模式从而实现更灵活、具备上下文感知的系统行为。图1路由设计模式2. 价值路由模式是构建动态、响应式智能体系统的关键通过路由智能体能超越简单线性流程智能决策如何处理信息、响应用户输入、调用工具或子智能体。分析输入并有条件地引导工作流是应对真实任务多样性的基础。3. 关键技术路由模式的核心组件是执行评估并引导流程的机制其实现方式包括基于LLM的路由、基于嵌入的路由、基于规则的路由、基于机器学习模型的路由等。4. 代码实现以下是一个完整的路由Routing模式的 Python程序代码展示了多种路由机制的实现import json import numpy as np from typing import Dict, Any, List, Tuple, Optional, Callable from enum import Enum import hashlib # 模拟的模型客户端根据您的实际导入方式调整 from model_factory import get_model_client client get_model_client() model gpt-4o class RouteType(Enum): 路由类型枚举 QA 问答 CALCULATION 计算 CODE_GENERATION 代码生成 DATA_ANALYSIS 数据分析 TRANSLATION 翻译 SUMMARIZATION 摘要 SENTIMENT 情感分析 CREATIVE 创意写作 UNKNOWN 未知 class RoutingDecision: 路由决策结果 def __init__(self, route_type: RouteType, confidence: float, reason: str, metadata: Dict[str, Any] None): self.route_type route_type self.confidence confidence self.reason reason self.metadata metadata or {} def to_dict(self) - Dict[str, Any]: return { route_type: self.route_type.value, route_code: self.route_type.name, confidence: self.confidence, reason: self.reason, metadata: self.metadata } class Router: 路由器基类 def __init__(self, name: str): self.name name def route(self, query: str, context: Dict[str, Any] None) - RoutingDecision: 路由决策方法子类需实现 raise NotImplementedError class LLMRouter(Router): 基于LLM的路由器 def __init__(self): super().__init__(LLM路由器) self.route_descriptions { RouteType.QA: 回答事实性问题、解释概念、提供信息, RouteType.CALCULATION: 执行数学计算、单位转换、公式求解, RouteType.CODE_GENERATION: 生成、解释或调试代码, RouteType.DATA_ANALYSIS: 分析数据、统计、可视化建议, RouteType.TRANSLATION: 语言翻译、本地化处理, RouteType.SUMMARIZATION: 文本摘要、要点提取, RouteType.SENTIMENT: 情感分析、观点提取, RouteType.CREATIVE: 创意写作、故事生成、诗歌创作 } def call_model(self, system_prompt: str, user_prompt: str) - str: 调用大模型 messages [ {role: system, content: system_prompt}, {role: user, content: user_prompt}, ] try: response client.chat.completions.create( modelmodel, messagesmessages, streamFalse ) return response.choices[0].message.content except Exception as e: print(fLLM调用失败: {e}) return {} def route(self, query: str, context: Dict[str, Any] None) - RoutingDecision: 使用LLM进行路由决策 system_prompt 你是一个智能路由决策系统。请分析用户的查询决定应该路由到哪个处理模块。 可用路由选项 1. QA - 问答回答事实性问题、解释概念、提供信息 2. CALCULATION - 计算执行数学计算、单位转换、公式求解 3. CODE_GENERATION - 代码生成生成、解释或调试代码 4. DATA_ANALYSIS - 数据分析分析数据、统计、可视化建议 5. TRANSLATION - 翻译语言翻译、本地化处理 6. SUMMARIZATION - 摘要文本摘要、要点提取 7. SENTIMENT - 情感分析情感分析、观点提取 8. CREATIVE - 创意写作创意写作、故事生成、诗歌创作 输出必须是严格的JSON格式包含以下字段 - route_type: 路由类型使用上面的代码QA、CALCULATION等 - confidence: 置信度0.0到1.0 - reason: 做出此路由决策的原因 - suggested_tools: 建议使用的工具列表 - complexity: 任务复杂度low/medium/high context_str f上下文信息{json.dumps(context, ensure_asciiFalse)} if context else 无额外上下文 user_prompt f用户查询{query} {context_str} 请分析这个查询并做出路由决策 result self.call_model(system_prompt, user_prompt) try: # 解析JSON输出 decision_data json.loads(result) route_type RouteType[decision_data.get(route_type, UNKNOWN)] confidence float(decision_data.get(confidence, 0.5)) reason decision_data.get(reason, 无具体原因) # 提取元数据 metadata { suggested_tools: decision_data.get(suggested_tools, []), complexity: decision_data.get(complexity, medium), router: self.name } return RoutingDecision(route_type, confidence, reason, metadata) except (json.JSONDecodeError, KeyError) as e: print(fLLM路由解析失败: {e}) return RoutingDecision(RouteType.UNKNOWN, 0.0, f路由解析失败: {str(e)}) class RuleBasedRouter(Router): 基于规则的路由器 def __init__(self): super().__init__(规则路由器) # 定义关键词规则 self.rules { RouteType.CALCULATION: [ 计算, 等于多少, 加减乘除, 公式, 方程式, 面积, 体积, 百分比, 平均数, , -, *, / ], RouteType.CODE_GENERATION: [ 代码, 编程, 函数, 算法, Python, Java, C, 实现, def , class , import , 调试, bug, 错误 ], RouteType.TRANSLATION: [ 翻译, 英文, 中文, 日语, 法语, 德语, translate, language, convert to ], RouteType.SUMMARIZATION: [ 总结, 摘要, 概括, 要点, 主要内容, summarize, brief, overview ], RouteType.SENTIMENT: [ 情感, 情绪, 观点, 态度, 喜欢, 讨厌, sentiment, opinion, feeling, attitude ], RouteType.QA: [ 什么是, 为什么, 怎么样, 如何, 谁, 哪里, 什么时候, 解释, 说明, 定义 ] } def route(self, query: str, context: Dict[str, Any] None) - RoutingDecision: 基于规则的路由决策 query_lower query.lower() # 计算每个路由类型的匹配分数 scores {} for route_type, keywords in self.rules.items(): score 0 matched_keywords [] for keyword in keywords: if keyword.lower() in query_lower: score 1 matched_keywords.append(keyword) if score 0: scores[route_type] { score: score, keywords: matched_keywords } if not scores: return RoutingDecision( RouteType.UNKNOWN, 0.3, 未匹配到任何规则关键词, {router: self.name} ) # 选择分数最高的路由 best_route max(scores.items(), keylambda x: x[1][score]) route_type best_route[0] confidence min(best_route[1][score] / 5.0, 1.0) # 归一化置信度 reason f匹配到关键词{, .join(best_route[1][keywords][:3])} return RoutingDecision( route_type, confidence, reason, { matched_keywords: best_route[1][keywords], score: best_route[1][score], router: self.name } ) class HybridRouter(Router): 混合路由器组合多种路由策略 def __init__(self): super().__init__(混合路由器) self.routers { llm: LLMRouter(), rule: RuleBasedRouter() } self.weights { llm: 0.7, # LLM路由器权重 rule: 0.3 # 规则路由器权重 } def route(self, query: str, context: Dict[str, Any] None) - RoutingDecision: 混合路由决策 decisions {} # 收集各个路由器的决策 for router_name, router in self.routers.items(): try: decision router.route(query, context) decisions[router_name] decision except Exception as e: print(f{router_name}路由器失败: {e}) if not decisions: return RoutingDecision( RouteType.UNKNOWN, 0.0, 所有路由器都失败了, {router: self.name} ) # 加权投票决策 route_votes {} for router_name, decision in decisions.items(): route_type decision.route_type weight self.weights.get(router_name, 0.5) confidence decision.confidence * weight if route_type not in route_votes: route_votes[route_type] { total_confidence: 0.0, contributors: [], reasons: [] } route_votes[route_type][total_confidence] confidence route_votes[route_type][contributors].append(router_name) route_votes[route_type][reasons].append(decision.reason) # 选择置信度最高的路由 best_route_type max( route_votes.items(), keylambda x: x[1][total_confidence] )[0] best_vote route_votes[best_route_type] return RoutingDecision( best_route_type, min(best_vote[total_confidence], 1.0), f混合决策{, .join(best_vote[contributors])}, { contributors: best_vote[contributors], individual_reasons: best_vote[reasons], total_confidence: best_vote[total_confidence], router: self.name } ) class RouteExecutor: 路由执行器 def __init__(self, router: Router None): self.router router or HybridRouter() self.execution_history [] def execute_route(self, query: str, context: Dict[str, Any] None) - Dict[str, Any]: 执行完整的路由流程 print(f\n{*60}) print(f处理查询: {query[:50]}...) print(f{*60}) # 步骤1: 路由决策 print(\n步骤1: 路由决策) decision self.router.route(query, context) print(f 路由类型: {decision.route_type.value}) print(f 置信度: {decision.confidence:.2f}) print(f 原因: {decision.reason}) # 步骤2: 执行路由对应的处理 print(f\n步骤2: 执行{decision.route_type.value}处理) result self._execute_handler(decision.route_type, query, context) # 步骤3: 记录执行历史 execution_record { timestamp: self._get_timestamp(), query: query, decision: decision.to_dict(), result: result } self.execution_history.append(execution_record) # 步骤4: 返回完整结果 final_result { success: True, route_decision: decision.to_dict(), execution_result: result, execution_id: hashlib.md5(query.encode()).hexdigest()[:8] } print(f\n执行完成!) print(f 路由类型: {decision.route_type.value}) print(f 执行状态: 成功) return final_result def _execute_handler(self, route_type: RouteType, query: str, context: Dict[str, Any]) - Dict[str, Any]: 执行对应的处理程序 handlers { RouteType.QA: self._handle_qa, RouteType.CALCULATION: self._handle_calculation, RouteType.CODE_GENERATION: self._handle_code_generation, RouteType.DATA_ANALYSIS: self._handle_data_analysis, RouteType.TRANSLATION: self._handle_translation, RouteType.SUMMARIZATION: self._handle_summarization, RouteType.SENTIMENT: self._handle_sentiment, RouteType.CREATIVE: self._handle_creative } handler handlers.get(route_type, self._handle_unknown) return handler(query, context) def _handle_qa(self, query: str, context: Dict[str, Any]) - Dict[str, Any]: 处理问答 system_prompt 你是一个专业的问答助手。请准确、清晰地回答用户的问题。 user_prompt f问题{query} result self._call_model_wrapper(system_prompt, user_prompt) return { handler: 问答处理器, answer: result, format: 文本回答 } def _handle_calculation(self, query: str, context: Dict[str, Any]) - Dict[str, Any]: 处理计算 system_prompt 你是一个数学计算助手。请计算用户提出的数学问题并展示计算过程。 user_prompt f计算{query} result self._call_model_wrapper(system_prompt, user_prompt) return { handler: 计算处理器, calculation: result, format: 数学表达式和结果 } def _handle_code_generation(self, query: str, context: Dict[str, Any]) - Dict[str, Any]: 处理代码生成 system_prompt 你是一个编程助手。请生成、解释或调试代码确保代码正确、高效。 user_prompt f代码相关需求{query} result self._call_model_wrapper(system_prompt, user_prompt) return { handler: 代码处理器, code: result, language: Python if python in query.lower() else 多种语言 } def _handle_data_analysis(self, query: str, context: Dict[str, Any]) - Dict[str, Any]: 处理数据分析 system_prompt 你是一个数据分析师。请分析数据、提供统计见解或可视化建议。 user_prompt f数据分析需求{query} result self._call_model_wrapper(system_prompt, user_prompt) return { handler: 数据分析处理器, analysis: result, recommendations: [统计摘要, 可视化建议, 趋势分析] } def _handle_translation(self, query: str, context: Dict[str, Any]) - Dict[str, Any]: 处理翻译 system_prompt 你是一个专业的翻译助手。请准确翻译文本保持原意和风格。 user_prompt f请翻译{query} result self._call_model_wrapper(system_prompt, user_prompt) return { handler: 翻译处理器, translation: result, detected_language: 自动检测 } def _handle_summarization(self, query: str, context: Dict[str, Any]) - Dict[str, Any]: 处理摘要 system_prompt 你是一个摘要专家。请提取文本的要点生成简洁准确的摘要。 user_prompt f请总结{query} result self._call_model_wrapper(system_prompt, user_prompt) return { handler: 摘要处理器, summary: result, key_points: 3 # 假设提取3个要点 } def _handle_sentiment(self, query: str, context: Dict[str, Any]) - Dict[str, Any]: 处理情感分析 system_prompt 你是一个情感分析专家。请分析文本的情感倾向和情绪。 user_prompt f请分析以下文本的情感{query} result self._call_model_wrapper(system_prompt, user_prompt) return { handler: 情感分析处理器, sentiment: result, metrics: [情感极性, 情绪强度, 主观程度] } def _handle_creative(self, query: str, context: Dict[str, Any]) - Dict[str, Any]: 处理创意写作 system_prompt 你是一个创意写作助手。请生成有创意、有趣的内容。 user_prompt f创意需求{query} result self._call_model_wrapper(system_prompt, user_prompt) return { handler: 创意写作处理器, creative_content: result, style: 根据查询自适应 } def _handle_unknown(self, query: str, context: Dict[str, Any]) - Dict[str, Any]: 处理未知类型 return { handler: 通用处理器, message: 无法确定具体处理类型使用通用处理, response: 我将尝试回答您的问题..., fallback: True } def _call_model_wrapper(self, system_prompt: str, user_prompt: str) - str: 包装模型调用 try: messages [ {role: system, content: system_prompt}, {role: user, content: user_prompt}, ] response client.chat.completions.create( modelmodel, messagesmessages, streamFalse ) return response.choices[0].message.content except Exception as e: return f处理失败: {str(e)} def _get_timestamp(self) - str: 获取时间戳 from datetime import datetime return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def get_history_summary(self) - Dict[str, Any]: 获取执行历史摘要 if not self.execution_history: return {message: 无执行历史} route_counts {} for record in self.execution_history: route_type record[decision][route_type] route_counts[route_type] route_counts.get(route_type, 0) 1 return { total_executions: len(self.execution_history), route_distribution: route_counts, recent_executions: self.execution_history[-5:] # 最近5次 } # 使用示例 if __name__ __main__: # 创建不同的路由器进行测试 print(测试不同的路由器) # 测试查询 test_queries [ 计算圆的面积半径是5厘米, 帮我写一个Python函数计算斐波那契数列, 翻译Hello, how are you?成中文, 总结一下人工智能的主要应用领域, 分析这段话的情感我今天真的很开心终于完成了这个项目 ] # 测试规则路由器 print(\n *60) print(测试规则路由器) print(*60) rule_router RuleBasedRouter() for query in test_queries: decision rule_router.route(query) print(f\n查询: {query}) print(f路由决策: {decision.route_type.value}) print(f置信度: {decision.confidence:.2f}) print(f原因: {decision.reason}) # 测试混合路由器 print(\n *60) print(测试混合路由器) print(*60) hybrid_router HybridRouter() for query in test_queries: decision hybrid_router.route(query) print(f\n查询: {query}) print(f路由决策: {decision.route_type.value}) print(f置信度: {decision.confidence:.2f}) print(f原因: {decision.reason}) # 测试路由执行器 print(\n *60) print(测试路由执行器) print(*60) executor RouteExecutor() for query in test_queries[:2]: # 只测试前两个查询 result executor.execute_route(query) print(f\n执行结果摘要:) print(f 路由类型: {result[route_decision][route_type]}) print(f 执行状态: {成功 if result[success] else 失败}) # 获取历史摘要 print(\n *60) print(执行历史摘要) print(*60) history executor.get_history_summary() print(f总执行次数: {history[total_executions]}) print(f路由分布: {history[route_distribution]})5. 代码说明核心组件1. 路由类型枚举RouteType定义了多种路由类型问答、计算、代码生成、数据分析、翻译、摘要、情感分析、创意写作等每种类型对应不同的处理逻辑2. 路由决策结果RoutingDecision包含路由类型、置信度、决策原因和元数据提供标准化的决策结果格式3. 路由器基类Router定义了路由器的接口规范所有具体路由器都继承此类4. LLM路由器LLMRouter使用大语言模型进行智能路由决策能够理解复杂的语义和上下文输出JSON格式的结构化决策结果5. 规则路由器RuleBasedRouter基于关键词匹配进行路由决策响应速度快确定性高适合简单明确的查询场景6. 混合路由器HybridRouter组合LLM和规则路由器的优势使用加权投票机制进行决策提高决策的准确性和鲁棒性7. 路由执行器RouteExecutor执行完整的路由流程管理路由决策和执行历史提供统一的执行接口主要特点1. 多路由策略支持LLM路由智能语义理解规则路由快速关键词匹配混合路由组合多种策略2. 灵活的扩展性易于添加新的路由类型支持自定义路由策略可调整权重和参数3. 完整的执行流程路由决策 → 处理执行 → 结果记录支持上下文传递提供执行历史分析4. 置信度评分机制每个决策都包含置信度支持多路由器加权投票可设置决策阈值5. 历史管理和分析记录所有执行历史提供路由分布统计支持结果持久化路由策略说明1. LLM路由的优势理解复杂的语义和上下文处理模糊或混合型查询提供解释性的决策原因2. 规则路由的优势响应速度快无需模型调用确定性高可预测性强适合简单明确的查询3. 混合路由的优势结合两者的优点提高决策的准确性和鲁棒性权重可调适应不同场景使用建议1. 根据场景选择路由器简单应用使用规则路由器复杂场景使用LLM或混合路由器2. 优化路由规则根据实际数据调整关键词优化LLM的提示词3. 添加新路由类型扩展RouteType枚举实现对应的处理程序4. 性能优化缓存路由决策结果并行处理多个路由器这个框架提供了完整的路由模式实现您可以根据具体需求进行调整和扩展构建更加智能和灵活的智能体系统。参照书籍《Agentic Design Patterns》的基本概念和观点。