应用程序通常需要动态行为,而不是执行预定义、线性的步骤序列。举例来说,设想一个旨在回答学术问题的系统。关于量子力学的问题,与关于罗马帝国的问题,需要不同的上下文和语气。如果都通过相同的通用提示和模型处理,将产生次优结果。因此,需要一种机制来智能地将输入路由到最适合的处理路径。这正是路由工作流程旨在解决的问题。它为您的工作流程引入条件逻辑,允许应用程序根据输入选择多条可能路径中的一条。路由器不再是 $g(f(\text{Input}))$ 这样的固定序列,而是实现了一种条件结构,类似于编程中的 if-else 语句。大型语言模型 (LLM) 被用作决策引擎来引导执行流程。路由器链的架构路由工作流程由两个主要组成部分构成:路由链:这是一条负责分析初始输入并决定将其发送到哪个目标的链。此链的提示会指示大型语言模型 (LLM) 对输入进行分类,并返回相应目标类别的名称。目标链:这是一组链,每条链都为特定任务设计。对于我们的学术助手示例,我们可能有 physics_chain、history_chain 和 math_chain。路由器根据其分析选择其中一条来执行。整个过程如下进行:用户的输入首先传递给路由链。路由链的大型语言模型 (LLM) 输出一个目标名称。系统随后使用此名称从其集合中查找对应的目标链,并使用原始输入执行它。digraph G { rankdir=TB; splines=ortho; node [shape=box, style="rounded,filled", fontname="Arial", margin="0.2,0.1"]; edge [fontname="Arial"]; bgcolor="transparent"; "Input" [label="输入", fillcolor="#a5d8ff"]; "RouterLLM" [label="路由LLM\n(选择目标)", fillcolor="#ffd8a8"]; "PhysicsChain" [label="物理链", fillcolor="#d0bfff"]; "MathChain" [label="数学链", fillcolor="#b2f2bb"]; "HistoryChain" [label="历史链", fillcolor="#ffc9c9"]; "Output" [label="输出", fillcolor="#e9ecef"]; Input -> RouterLLM; RouterLLM -> PhysicsChain [label=" 'physics' "]; RouterLLM -> MathChain [label=" 'math' "]; RouterLLM -> HistoryChain [label=" 'history' "]; PhysicsChain -> Output; MathChain -> Output; HistoryChain -> Output; }输入首先由路由大型语言模型 (LLM) 进行评估,该模型选择几条专用目标链中的一条来生成最终输出。构建多提示路由器我们来构建一个路由工作流程的实用示例,该流程将问题发送给不同的专家:物理学家、数学家和历史学家。首先,我们为目标链定义提示。每个提示都引导大型语言模型 (LLM) 采用特定的角色。import os from langchain_openai import ChatOpenAI from langchain_core.prompts import PromptTemplate from langchain_core.output_parsers import StrOutputParser # 假设您的环境中已设置 OPENAI_API_KEY # os.environ["OPENAI_API_KEY"] = "your-api-key" llm = ChatOpenAI(temperature=0, model="gpt-4o") physics_template = """你是一位非常聪明的物理学教授。 你擅长以简洁易懂的方式回答物理问题。 当你不知道一个问题的答案时,你会承认不知道。 这是一个问题: {input}""" math_template = """你是一位非常出色的数学家。你擅长回答数学问题。 你之所以如此出色,是因为你能够将难题分解成其组成部分, 回答这些组成部分,然后将它们组合起来回答更广泛的问题。 这是一个问题: {input}""" history_template = """你是一位非常出色的历史学家。你对历史有着丰富的知识和卓越的记忆力。 你尤其擅长通过讲故事的方式回答问题。 这是一个问题: {input}""" prompt_infos = [ { "name": "physics", "description": "适合回答物理问题", "prompt_template": physics_template, }, { "name": "math", "description": "适合回答数学问题", "prompt_template": math_template, }, { "name": "history", "description": "适合回答历史问题", "prompt_template": history_template, }, ]接下来,我们使用管道 | 语法创建目标链。每条链由一个提示、大型语言模型 (LLM) 和一个输出解析器组成。destination_chains = {} for p_info in prompt_infos: name = p_info["name"] prompt_template = p_info["prompt_template"] prompt = PromptTemplate.from_template(template=prompt_template) chain = prompt | llm | StrOutputParser() destination_chains[name] = chain现在是核心路由逻辑。我们定义一个提示,指示大型语言模型 (LLM) 充当分类器。它应该只输出最符合输入的类别名称。destinations = [f"{p['name']}: {p['description']}" for p in prompt_infos] destinations_str = "\n".join(destinations) router_template = """根据以下用户输入,将其分类为关于 {options} 中的一项。 选择描述: {destinations} 返回所选项的名称,仅此而已。 输入: {input} """ router_prompt = PromptTemplate.from_template(router_template) router_prompt = router_prompt.partial(destinations=destinations_str, options=", ".join([p['name'] for p in prompt_infos])) router_chain = router_prompt | llm | StrOutputParser()最后,我们使用 RunnableLambda 组装所有部分。我们定义一个 route 函数,它接收路由器链的输出(类别名称),并返回相应的目标链。from langchain_core.runnables import RunnableLambda, RunnablePassthrough def route(info): # 路由链的输出通过 info["destination"] 传递到此处 destination = info["destination"].strip().lower() # 从字典中选择链 if destination in destination_chains: return destination_chains[destination] else: # 如果目前不明确,则回退到物理链 return destination_chains["physics"] # 我们构建完整的链: # 1. 运行路由器并传递输入 # 2. 使用 route 函数选择下一条链 chain = { "destination": router_chain, "input": RunnablePassthrough() } | RunnableLambda(route)我们用几个不同的输入来测试一下。# 用物理问题进行测试 response_physics = chain.invoke("What is the formula for gravitational potential energy?") print(f"物理响应: {response_physics}") # 用历史问题进行测试 response_history = chain.invoke("When was the Battle of Hastings?") print(f"历史响应: {response_history}")当你运行物理问题时,路由器正确识别出 physics 类别,并由专门的物理提示生成输出:物理响应: 行星表面附近引力势能 (U) 的公式是: U = mgh 其中: - m 是物体的质量。 - g 是重力加速度。 - h 是物体相对于参考点的高度。同样,历史问题也被正确地路由到历史链。这种动态路由使得应用程序能够使用专用提示,显著提升其响应的质量和相关性。处理默认情况一个设计良好的系统必须妥善处理歧义。如果用户提出的问题不完全符合任何预定义类别,例如“数学史是什么?”路由器可能难以做出明确选择。为了解决这个问题,我们可以定义一条默认链,并更新我们的路由逻辑,使其在路由器的输出与已知目标不匹配时回退到该链。以下是如何在我们的示例中添加默认链:# 创建一个通用链作为回退 default_prompt = PromptTemplate.from_template("{input}") default_chain = default_prompt | llm | StrOutputParser() def route(info): destination = info["destination"].strip().lower() # 使用 .get() 方法提供默认回退 return destination_chains.get(destination, default_chain) # 使用更新后的路由逻辑重新初始化完整链 chain = { "destination": router_chain, "input": RunnablePassthrough() } | RunnableLambda(route)加入此功能后,路由器无法确定分类(或输出未知类别)的任何输入都将由 default_chain 处理,从而使应用程序更加一致。这种模式提供了一种有效的方法,通过超越简单的线性序列并整合条件式、逻辑驱动的工作流程,来构建更精巧、更智能的应用程序。