LangChain RunnableBranch条件路由在 Agent 决策树中的实战应用一、深度引言与场景痛点大家好我是赵咕咕。Agent 系统最常见的架构模式是意图路由用户的 query 进来 → 判断意图 → 路由到不同的处理链路。简单意图走缓存复杂意图走 LLM Agent 推理纯计算走 Python 代码执行器。手写这个路由逻辑很简单——if-elif-else一把梭。但三个月后你的路由逻辑从 5 个分支膨胀到 23 个if-elif 套 if-elif缩进深到屏幕放不下。更关键的是LangChain 的 Chain 体系天然支持链式组合——你用 if-else 硬编排整个链路就变成了LangChain 外部的代码丧失了 Chain 的声明式组合能力。LangChain 提供了RunnableBranch专门用来解决这个问题。这篇文章我把RunnableBranch在 Agent 决策树中的实战用法和工程实践整理出来。二、底层机制与原理深度剖析2.1 RunnableBranch 是什么RunnableBranch是 LangChain Expression Language (LCEL) 提供的条件路由组件。它的本质是声明式的 if-elif-elsefrom langchain_core.runnables import RunnableBranch chain RunnableBranch( (lambda x: search in x[intent], search_chain), # if (lambda x: analyze in x[intent], analyze_chain), # elif default_chain, # else )和普通 if-else 的核心区别它返回一个 Runnable可以和其他 Chain 无缝组合RunnableBranch → RunnablePassthrough → RunnableLambda。条件函数接收链路的完整状态不只是intent还包括对话历史、检索到的上下文、用户元数据等。支持异步条件条件函数可以是async def允许在判断分支前做异步操作如查数据库确认权限。2.2 Agent 决策树的 LCEL 实现典型的 Agent 决策树用户输入 ├─ 意图识别 (IntentClassifier Runnable) │ ├─ search → RAG 检索链 │ ├─ codegen → 代码生成链 │ ├─ analyze → 数据分析链 │ └─ chat → 通用对话链 └─ 后处理 (统一格式化输出)用 LCEL 表达这个决策树核心优势每个分支的链路RAG/Code/Analyze/Chat是独立的 Runnable可以独立测试、独立部署、独立演进。2.3 RunnableBranch vs if-else 的深层差异维度if-elseRunnableBranch组合性外部编排与 Chain 脱节返回 Runnable可链式组合测试性路由逻辑和业务逻辑耦合每个分支独立测试可观测性需手动加日志LangSmith 自动追踪异步支持手动 async/await内置异步条件函数错误传播手动 try/exceptRunnable 自动错误传播流式输出不支持全链路支持 streaming三、生产级代码实现import asyncio import logging from typing import Any from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import ( Runnable, RunnableBranch, RunnableConfig, RunnableLambda, RunnablePassthrough, ) from langchain_core.runnables.base import RunnableSequence from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field logger logging.getLogger(__name__) # ─── 数据模型 ─── class AgentState(BaseModel): Agent 状态贯穿整个决策链路。 user_input: str user_id: str intent: str chat # 意图search / codegen / analyze / chat confidence: float 0.0 context: dict[str, Any] Field(default_factorydict) response: str error: str trace_id: str # ─── 意图识别 ─── INTENT_CLASSIFIER_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一个意图识别专家。分析用户输入判断其意图。 可选意图 - search: 用户想搜索/查找/检索信息 - codegen: 用户想生成/编写/修改代码 - analyze: 用户想分析数据/图表/趋势 - chat: 普通的闲聊对话 返回 JSON: {{intent: search|codegen|analyze|chat, confidence: 0.0-1.0}}), (human, {user_input}), ]) async def classify_intent( state: AgentState, llm: ChatOpenAI ) - AgentState: 意图识别 Runnable。 try: chain INTENT_CLASSIFIER_PROMPT | llm | StrOutputParser() result await chain.ainvoke({user_input: state.user_input}) import json data json.loads(result) state.intent data.get(intent, chat) state.confidence float(data.get(confidence, 0.5)) logger.info( 意图: %s (confidence%.2f), state.intent, state.confidence ) except Exception as e: logger.error(意图识别失败: %s, 降级为 chat, e) state.intent chat state.confidence 0.0 return state # ─── 各分支链路 ─── SEARCH_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一个搜索助手。基于用户的查询提供精确的答案。), (human, {user_input}), ]) async def search_chain_factory(llm: ChatOpenAI) - Runnable: RAG 检索链路。 return ( RunnablePassthrough() | (lambda state: {user_input: state.user_input}) | SEARCH_PROMPT | llm | StrOutputParser() ) CODEGEN_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一个代码生成专家。生成 Python 代码。 要求 - 使用 async/await - 包含类型标注 - 包含异常处理 - 包含日志记录), (human, {user_input}), ]) async def codegen_chain_factory(llm: ChatOpenAI) - Runnable: 代码生成链路。 return ( RunnablePassthrough() | (lambda state: {user_input: state.user_input}) | CODEGEN_PROMPT | llm | StrOutputParser() ) CHAT_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一个友好的智能助手。), (human, {user_input}), ]) async def chat_chain_factory(llm: ChatOpenAI) - Runnable: 通用对话链路。 return ( RunnablePassthrough() | (lambda state: {user_input: state.user_input}) | CHAT_PROMPT | llm | StrOutputParser() ) # ─── 构建完整 Agent 决策树 ─── class AgentOrchestrator: Agent 编排器基于 RunnableBranch 构建决策树。 def __init__( self, llm: ChatOpenAI, enable_analyze: bool False, # 分析链路可选 ): self._llm llm self._enable_analyze enable_analyze async def build(self) - Runnable: 构建完整的 Agent 决策链路。 链路结构: preprocess → intent_classify → branch → postprocess # ── 预处理 ── preprocess RunnableLambda(self._preprocess) # ── 意图识别 ── intent_classify RunnableLambda( lambda s: classify_intent(s, self._llm) ) # ── 各分支 ── search_chain RunnableLambda( lambda s: self._run_branch(s, search) ) codegen_chain RunnableLambda( lambda s: self._run_branch(s, codegen) ) analyze_chain RunnableLambda( lambda s: self._run_branch(s, analyze) ) # ── 条件路由 ── branch RunnableBranch( ( lambda s: s.intent search, search_chain, ), ( lambda s: s.intent codegen, codegen_chain, ), ( lambda s: self._enable_analyze and s.intent analyze, analyze_chain, ), # 默认分支chat 任何未识别的意图 RunnableLambda(lambda s: self._run_branch(s, chat)), ) # ── 后处理 ── postprocess RunnableLambda(self._postprocess) # ── 组装完整链路 ── full_chain ( preprocess | intent_classify | branch | postprocess ) return full_chain staticmethod async def _preprocess(state: AgentState) - AgentState: 预处理输入清洗和脱敏。 # 移除多余空白 state.user_input state.user_input.strip() # 截断过长输入 if len(state.user_input) 4000: state.user_input state.user_input[:4000] ... logger.warning(用户输入过长已截断) return state async def _run_branch( self, state: AgentState, branch_name: str ) - AgentState: 执行指定分支的链路。 try: if branch_name search: chain await search_chain_factory(self._llm) elif branch_name codegen: chain await codegen_chain_factory(self._llm) elif branch_name analyze: chain await chat_chain_factory(self._llm) # 简化 else: chain await chat_chain_factory(self._llm) response await chain.ainvoke({user_input: state.user_input}) state.response response logger.info(分支 %s 执行完成, branch_name) except Exception as e: logger.error(分支 %s 执行失败: %s, branch_name, e) state.error f分支 {branch_name} 异常: {e} state.response 抱歉系统处理您的请求时遇到问题请稍后重试。 return state staticmethod async def _postprocess(state: AgentState) - AgentState: 后处理格式校验和日志记录。 # 响应长度检查 if len(state.response) 8000: state.response state.response[:8000] \n\n回答过长已截断 # 敏感信息检查 import re phone_pattern r1[3-9]\d{9} state.response re.sub( phone_pattern, [PHONE_HIDDEN], state.response ) # 日志 logger.info( Agent 处理完成: intent%s, response_len%d, error%s, state.intent, len(state.response), bool(state.error), ) return state # ─── 流式输出支持 ─── async def stream_response( orchestrator: AgentOrchestrator, user_input: str ): 支持流式输出的 Agent 调用。 chain await orchestrator.build() state AgentState(user_inputuser_input) # 非流式一次性返回 result await chain.ainvoke(state) return result.response # ─── 使用示例 ─── async def main(): llm ChatOpenAI(modelgpt-4o, temperature0) orchestrator AgentOrchestrator(llmllm) test_inputs [ Python 中 async 和 await 的用法, 帮我写一个 FastAPI 的健康检查接口, 今天天气真好, ] for user_input in test_inputs: response await stream_response(orchestrator, user_input) print(f\n输入: {user_input}) print(f输出: {response[:100]}...) if __name__ __main__: asyncio.run(main())代码中几个关键的设计RunnableLambda 包装意图识别、各分支处理、后处理都用RunnableLambda包装为 Runnable保证全链路一致性。分支条件函数RunnableBranch的条件函数接收完整的AgentState不只是意图字段。这意味着你可以基于用户权限、时间段、历史行为做更复杂的路由判断。后处理统一所有分支的输出都经过同一个_postprocess保证格式一致性、敏感信息脱敏、日志记录。流式支持RunnableBranch 天然支持 streaming——如果 LLM 配置了streamingTrue整个链路会自动流式输出。四、边界分析与架构权衡4.1 RunnableBranch 的局限性RunnableBranch不是一个图调度器——它的路由是线性的、一次性的。一旦路由到某个分支就不会再跳转到另一个分支。如果你的场景需要Agent 可以在多个能力间来回切换类似于 ReAct Agent 的 Thought-Action-Observation 循环那应该用 LangGraph 而不是 RunnableBranch。4.2 复杂路由的层级组织当意图种类超过 10 个时单一层的RunnableBranch变得难以维护。建议用层级路由# 第一层粗粒度路由 top_branch RunnableBranch( (lambda s: s.intent in [search, qa], search_qa_branch), # 第二层 (lambda s: s.intent in [codegen, debug], code_branch), chat_chain, # 默认 ) # 第二层search/qa 的细粒度路由 search_qa_branch RunnableBranch( (lambda s: s.intent search, search_chain), (lambda s: s.intent qa, qa_chain), )4.3 动态路由 vs 声明式路由场景推荐方案固定意图分类5-10 种RunnableBranch声明式意图数量经常变化动态路由LLM 函数调用选择链路意图间有复杂依赖LangGraph 状态图意图需要多轮判断LangGraph4.4 错误处理和降级在 RunnableBranch 中某个分支抛出异常不会影响其他分支——这跟手动 if-else 一致。但不同的是RunnableBranch 的错误可以通过统一的 error handler 捕获chain_with_fallback chain.with_fallbacks([ RunnableLambda(lambda s: _fallback_handler(s)) ])当任何分支失败时自动走降级链路。五、总结RunnableBranch是把 Agent 的决策树从手写 if-else 面条代码变成声明式、可组合、可测试的 LCEL 链的关键工具。使用要点把每个分支封装为独立的 Runnable——独立测试、独立部署、独立迭代。条件函数基于完整的 AgentState——不只是意图还有用户上下文和历史。后处理统一——所有分支的输出经过同一个后处理环节保证输出格式和安全一致。超过 10 个分支用层级路由——不要让一个 RunnableBranch 处理所有情况。LangChain 的 LCEL 生态里RunnableBranch 是最被低估的组件之一。很多人抱怨LangChain 太抽象不如自己写 if-else——但如果你的 Agent 系统要长期维护、持续演进声明式的条件路由比手写的 if-else 有质的区别。抽象不是为了炫技是为了让系统在下一次迭代时不需要重写调度逻辑。下一篇预告Python 部署优化使用 Docker 多阶段构建缩小 RAG 服务镜像体积。
