企业官网建设流程全解析

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-else：

from 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 的深层差异

三、生产级代码实现

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_factory=dict) 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 = r"1[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_input=user_input) # 非流式：一次性返回 result = await chain.ainvoke(state) return result.response # ─── 使用示例 ─── async def main(): llm = ChatOpenAI(model="gpt-4o", temperature=0) orchestrator = AgentOrchestrator(llm=llm) 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 配置了streaming=True，整个链路会自动流式输出。

四、边界分析与架构权衡

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 声明式路由

4.4 错误处理和降级

在 RunnableBranch 中，某个分支抛出异常不会影响其他分支——这跟手动 if-else 一致。但不同的是，RunnableBranch 的错误可以通过统一的 error handler 捕获：

chain_with_fallback = chain.with_fallbacks([ RunnableLambda(lambda s: _fallback_handler(s)) ])

当任何分支失败时，自动走降级链路。

五、总结

RunnableBranch是把 Agent 的决策树从"手写 if-else 面条代码"变成"声明式、可组合、可测试的 LCEL 链"的关键工具。

使用要点：

1. 把每个分支封装为独立的 Runnable——独立测试、独立部署、独立迭代。
2. 条件函数基于完整的 AgentState——不只是意图，还有用户上下文和历史。
3. 后处理统一——所有分支的输出经过同一个后处理环节，保证输出格式和安全一致。
4. 超过 10 个分支用层级路由——不要让一个 RunnableBranch 处理所有情况。

LangChain 的 LCEL 生态里，RunnableBranch 是最被低估的组件之一。很多人抱怨"LangChain 太抽象，不如自己写 if-else"——但如果你的 Agent 系统要长期维护、持续演进，声明式的条件路由比手写的 if-else 有质的区别。抽象不是为了炫技，是为了让系统在下一次迭代时不需要重写调度逻辑。

下一篇预告：Python 部署优化，使用 Docker 多阶段构建缩小 RAG 服务镜像体积。