企业官网建设流程全解析

1. 为什么“看懂架构”比“跑通Demo”更重要：一个被低估的起点

我带过十几支用LangChain落地AI应用的团队，几乎每支队伍都会在第三周左右集体卡住——不是模型调不通，不是API报错，而是所有人突然发现：自己写的链（Chain）越改越乱，加个新节点就崩，调试时根本不知道数据在哪一层被篡改，日志里全是<langchain_core.runnables.base.RunnableLambda object at 0x...>这种无法定位的地址。后来我挨个翻代码，发现90%的问题根源只有一个：没人真正读过langchain_core.runnables模块的源码，更没人把Runnable接口当成一条设计铁律来遵守。

这恰恰是“LangChain之一：整体架构总览”这个标题最该说清的事——它不是一张静态的PPT图，不是让你背下“Model、Retriever、OutputParser”这几个名词，而是要建立一套可预测、可拆解、可验证的数据流契约。你不需要第一天就手写RunnableParallel，但必须清楚：当你调用.invoke()时，背后触发的是__call__还是ainvoke？config参数传进去后，会在哪一层被merge_configs合并？stream返回的AsyncIterator和Iterator在内存占用上差多少倍？这些不是细节，是架构的呼吸节奏。

LangChain的文档里藏着一句没加粗的话：“Everything is a Runnable.” 这句话不是修辞，是工程约束。它意味着：你写的每个函数、每个LLM封装、每个数据库查询类，只要实现了invoke/ainvoke/stream/astream这四个方法，就能无缝接入整个生态。而LCEL（LangChain Expression Language）就是这套契约的DSL表达——它不定义功能，只定义组合规则。就像乐高积木的凸点与凹槽，形状决定能否拼接，而不是颜色决定好不好看。

所以这篇“总览”，我们不画框图，不列模块，直接从Runnable的5个核心契约开始切：输入契约、输出契约、配置契约、流式契约、错误契约。每一个契约背后，都对应着你在实际项目中踩过的坑。比如，你是否遇到过RunnablePassthrough在RunnableParallel里突然丢掉字段？那是因为你没理解“输入契约”对dict键名的强约束；你是否试过stream=True却收不到chunk？那大概率是RunnableBinding里漏写了_stream方法的重载。这些都不是Bug，是契约未被尊重的必然结果。

提示：本文所有代码片段均基于langchain-core==0.3.12和langchain==0.3.12实测，所有行为差异均标注Python版本兼容性（如3.9+）。不依赖任何第三方插件或非官方分支，所有结论均可在官方GitHub仓库的langchain-core/langchain_core/runnables/路径下验证。

2. Runnable：五个不可妥协的契约，决定了你的代码能否存活超过三个月

LangChain的架构不是自上而下设计出来的宏伟蓝图，而是从Runnable这个最小单元，像细胞分裂一样自然生长出来的。它的五个核心契约，不是文档里的可选建议，而是运行时强制校验的接口协议。忽略任何一个，你的代码可能在本地跑通，但在生产环境的并发压测中崩溃，或者在升级到下一个minor版本时静默失效。

2.1 输入契约：不是“能接收什么”，而是“必须声明接收什么”

Runnable的输入契约，由InputType泛型参数和input_schema()方法共同定义。很多人以为invoke(input)的input可以是任意类型，实则不然。LangChain在调用前会执行self.input_schema().validate(input)，而默认schema是Any——但这只是兜底，不是自由。

真正的约束来自你显式声明的类型。例如：

from typing import Dict, Any, Union from langchain_core.runnables import RunnableLambda # 错误示范：看似灵活，实则埋雷 def bad_handler(x): return {"result": x["query"] + " processed"} bad_chain = RunnableLambda(bad_handler) # 正确示范：声明输入结构，获得自动校验 from pydantic import BaseModel class SearchInput(BaseModel): query: str filters: Dict[str, Any] = {} def good_handler(input: SearchInput) -> Dict[str, str]: return {"result": input.query + " processed"} good_chain = RunnableLambda(good_handler)

关键区别在哪？当你调用bad_chain.invoke({"query": "test"})时，一切正常；但若传入{"query": "test", "extra": "field"}，bad_handler照常执行，不会报错。而good_chain.invoke({"query": "test", "extra": "field"})会立刻抛出ValidationError，提示extra字段不在schema中。

这看起来是“多此一举”，但实际价值巨大：

• 调试阶段：前端传参多了个字段，后端日志直接告诉你“谁传错了”，而不是让你在10层嵌套后去查input被谁篡改；
• 协作阶段：其他开发者看到SearchInput类型注解，立刻明白这个链只接受什么结构，无需翻文档；
• 升级阶段：LangChain 0.4.x开始，input_schema()校验默认启用，未声明类型的RunnableLambda将被标记为deprecated。

注意：RunnableLambda的类型推断依赖于函数签名。如果你用lambda x: ...匿名函数，Pydantic无法提取类型，input_schema()将回退到Any。务必使用具名函数并标注类型。

2.2 输出契约：不是“返回什么”，而是“承诺返回什么”

输出契约由OutputType泛型和output_schema()方法定义。它解决的是“下游组件如何信任上游输出”的问题。LangChain中最常见的反模式，就是把Runnable当作普通函数用，忽略了输出结构的契约性。

举个典型场景：你需要把LLM返回的JSON字符串解析成字典，再交给后续步骤处理。

# 危险写法：输出类型模糊 import json from langchain_core.runnables import RunnableLambda parse_json = RunnableLambda(lambda x: json.loads(x)) # 安全写法：明确承诺输出结构 from pydantic import BaseModel class ParsedResponse(BaseModel): answer: str confidence: float def safe_parse_json(input_str: str) -> ParsedResponse: data = json.loads(input_str) return ParsedResponse(**data) safe_parse = RunnableLambda(safe_parse_json)

差别在哪？当parse_json作为RunnableParallel的一部分时，如果LLM返回了非法JSON，json.loads()抛异常，整个并行链失败；而safe_parse在invoke前会先校验输入是否为str，在output_schema()中已声明返回ParsedResponse，LangChain会自动包装异常为OutputParserException，且RunnableParallel能捕获并隔离该错误，不影响其他分支。

更关键的是，safe_parse.output_schema()返回的是ParsedResponse的schema，下游组件（如RunnableBinding）能据此生成准确的config提示，比如自动注入response_format={"type": "json_object"}到LLM调用中。

2.3 配置契约：config不是“传参”，而是“上下文快照”

config参数是LangChain架构中最易被误解的部分。很多人把它当成kwargs的别名，随意传入{"timeout": 30}或{"model": "gpt-4"}，却不知config是一个有严格结构的RunnableConfig对象，其核心是run_id、callbacks、tags、metadata四个字段，以及可选的recursion_limit和max_concurrency。

真正的配置契约体现在merge_configs()方法上。当你构建复杂链时，config会在每一层被合并：

from langchain_core.runnables import RunnablePassthrough from langchain_core.callbacks import BaseCallbackHandler class MyHandler(BaseCallbackHandler): def on_chain_start(self, *args, **kwargs): print("Chain started") base_config = {"callbacks": [MyHandler()], "tags": ["prod"]} chain = RunnablePassthrough() | (lambda x: x.upper()) # 调用时传入config result = chain.invoke("hello", config=base_config)

此时，RunnablePassthrough会将自己的默认config（空字典）与base_config合并，再传递给lambda。合并规则是：

• callbacks：列表合并（去重）
• tags：列表合并（去重）
• metadata：字典深度合并（子键覆盖）
• run_id：若未指定，自动生成UUID

这意味着，如果你在某个子链里写了config={"tags": ["debug"]}，它不会覆盖["prod"]，而是变成["prod", "debug"]。这是设计使然——config是上下文快照，不是覆盖指令。

实测心得：在微服务架构中，我们曾用metadata传递trace_id。但发现RunnableParallel的多个分支会各自生成独立run_id，导致trace断裂。解决方案是：在入口处生成run_id，并通过metadata显式透传，禁用run_id自动生成（config={"run_id": None}），否则LangChain会覆盖你传的metadata。

2.4 流式契约：stream不是“开关”，而是“协议切换”

stream和astream方法的存在，不是为了提供“流式输出”这个功能，而是为了实现零拷贝数据流管道。stream返回Iterator[OutputType]，astream返回AsyncIterator[OutputType]，它们的契约要求：每次yield必须是OutputType的完整实例，不能是partial chunk。

常见错误是试图在stream中yield字符串片段：

# 错误：违反流式契约 def bad_streamer(input: str): for char in input: yield char # ❌ yield str, but OutputType is str? No — it's the full result! # 正确：yield完整输出单元 def good_streamer(input: str) -> Iterator[str]: # 模拟LLM流式返回完整句子 sentences = ["Hello", "world", "this", "is", "a", "test"] for sent in sentences: yield sent # ✅ each yield is a complete str, matching OutputType

为什么这么严格？因为RunnableParallel、RunnableMap等组合器，需要在流式模式下保证各分支的yield事件时间对齐。如果A分支yield"Hel"，B分支yield"World"，组合器无法判断这是两个独立结果，还是同一结果的碎片——它只能假设每个yield都是原子单元。

LCEL中|操作符在流式模式下的行为，完全取决于右侧Runnable是否实现了stream。如果没实现，|会自动降级为invoke，失去流式能力。因此，检查一个链是否支持流式，不是看它有没有stream=True参数，而是看链中所有节点是否都实现了stream方法。

2.5 错误契约：异常不是“中断”，而是“结构化信号”

Runnable的错误契约规定：所有异常必须继承自BaseException，且推荐使用LangChain内置异常类（如OutputParserException、RetrieverError）。这不是形式主义，而是为了让RunnableTry、RetryPolicy等容错组件能精准识别错误类型。

例如，你封装一个向量库检索器：

from langchain_core.runnables import Runnable from langchain_core.exceptions import RetrieverError class VectorDBRetriever(Runnable): def __init__(self, db): self.db = db def invoke(self, input: str, config=None): try: return self.db.search(input) except ConnectionError as e: # ❌ 不要 raise e — 类型不匹配 # ✅ 必须转换为LangChain标准异常 raise RetrieverError(f"VectorDB connection failed: {e}") from e

这样，当上游使用RunnableTry时：

from langchain_core.runnables import RunnableTry retriever = VectorDBRetriever(my_db) fallback = RunnableLambda(lambda _: [{"content": "default"}]) robust_retriever = RunnableTry( runnable=retriever, fallback=fallback, # 只对RetrieverError重试，对ValueError不重试 retry_errors=[RetrieverError] )

如果没有遵循错误契约，RunnableTry无法区分ConnectionError和业务逻辑错误（如ValueError），可能导致无限重试或错误静默。

3. LCEL：不是语法糖，而是运行时编译器的DSL

很多初学者把LCEL（LangChain Expression Language）当成|符号的链式调用语法糖，认为llm | parser和llm.invoke(...) | parser.invoke(...)只是写法不同。这是致命误解。LCEL的本质，是在调用前将链编译为一个可复用、可序列化、可 introspect 的Runnable对象，而非运行时动态拼接函数。

3.1 编译期 vs 运行期：LCEL的两次生命

LCEL表达式在首次创建时经历编译期，生成一个RunnableSequence实例；在每次invoke时经历运行期，执行编译好的执行计划。

from langchain_core.runnables import RunnableLambda from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 编译期：创建RunnableSequence对象 prompt = ChatPromptTemplate.from_template("Tell me about {topic}") llm = ChatOpenAI(model="gpt-3.5-turbo") chain = prompt | llm | (lambda x: x.content.upper()) # 此时chain已是完整Runnable，可序列化 print(type(chain)) # <class 'langchain_core.runnables.base.RunnableSequence'> print(chain.get_graph().draw_mermaid()) # 可视化执行图

关键点在于：chain不是三个函数的简单管道，而是一个拥有自己input_schema、output_schema、config_schema的独立实体。你可以对它做：

• chain.with_config({"tags": ["lcel-demo"]})—— 返回新Runnable，不修改原链；
• chain.with_types(input_type=Dict[str, str])—— 强制重写输入类型；
• chain.get_graph().to_json()—— 导出为JSON，供前端可视化。

而手动拼接：

# 运行期：每次调用都重新执行 def manual_chain(topic: str): messages = prompt.invoke({"topic": topic}) response = llm.invoke(messages) return response.content.upper()

manual_chain只是一个普通函数，没有schema、没有config、无法introspect、无法序列化。它在LangChain生态中是“黑盒”，无法被RunnableParallel调度，也无法被LangGraph的State管理。

3.2 LCEL的四大运算符：每个都对应一个编译规则

LCEL的|、+、&、@运算符，不是随意设计的符号，而是对应四种不同的Runnable组合模式，每种都有严格的编译规则：

其中&和@最易被误用。&不是“and”，而是“assign”——它要求左侧必须是dict，右侧必须是Runnable，且右侧的输出会被赋值给左侧的键名：

# 正确：assign返回新dict from langchain_core.runnables import RunnablePassthrough from operator import itemgetter # 将retriever结果赋给"context"键 assign_context = {"context": retriever} & {"question": itemgetter("question")} # 错误：&右侧不是Runnable # {"context": retriever} & {"question": "what is AI?"} # ❌ TypeError

@也不是“at”，而是“pick”——它从dict中提取单个键，如果键不存在，抛KeyError。它不支持嵌套路径（如"data.answer"），必须用itemgetter组合：

# 正确：提取单层键 chain @ "answer" # 提取嵌套字段：需组合 from operator import itemgetter chain | itemgetter("data", "answer") # ✅ 返回data["answer"]

3.3 LCEL的隐式转换：何时自动包装，何时拒绝编译

LCEL编译器会自动将常见Python对象转换为Runnable，但有严格边界：

• 函数→RunnableLambda：仅当函数有明确签名（非lambda）且类型可推断；
• 字符串模板→PromptTemplate：仅当使用ChatPromptTemplate.from_template()等工厂方法；
• 字典→RunnablePassthrough：仅当字典值是Runnable（如{"key": runnable}）；
• 列表→RunnableParallel：仅当列表元素全是Runnable。

而以下情况会编译失败：

• lambda x: x—— 类型无法推断，编译器拒绝；
• "hello {name}"—— 字符串字面量，不是PromptTemplate实例；
• {"key": "value"}—— 字典值不是Runnable，编译器报TypeError。

这意味着，LCEL不是“让代码更短”，而是“让意图更明确”。当你写prompt | llm时，编译器知道你要构建一个RunnableSequence；当你写{"context": retriever}时，编译器知道你要构建RunnableAssign。这种明确性，是手动拼接永远无法提供的。

4. LangChain Core模块全景：从runnables到callbacks的职责地图

LangChain的架构不是按“功能域”（如LLM、Retriever）划分的，而是按数据流生命周期组织的。langchain-core包的模块结构，就是一张清晰的职责地图。理解这张地图，比死记硬背每个类更重要。

4.1runnables：数据流的骨架与肌肉

langchain_core.runnables是整个架构的基石，包含三类核心组件：

• 基类：Runnable（抽象基类）、RunnableBinding（绑定配置）、RunnableBranch（条件分支）；
• 组合器：RunnableSequence（线性）、RunnableParallel（并行）、RunnableMap（映射）、RunnableTry（容错）；
• 工具类：RunnableLambda（函数包装）、RunnablePassthrough（透传）、RunnableAssign（赋值）。

关键洞察：runnables模块不依赖任何模型或工具。你可以只安装langchain-core，就能构建完整的Runnable链。这意味着，runnables是纯协议层，而langchain（非-core）包才是实现层。

实操中，我们常把runnables作为“胶水层”：

• 在FastAPI路由中，直接chain.invoke(request)，不碰LLM；
• 在Celery任务中，序列化RunnableSequence对象，跨进程执行；
• 在测试中，用RunnableLambda(lambda x: {"mock": True})替换真实LLM，隔离外部依赖。

4.2callbacks：数据流的神经系统

langchain_core.callbacks不是简单的日志记录器，而是数据流的观察者模式实现。它定义了BaseCallbackHandler抽象类，所有回调（如StreamingStdOutCallbackHandler、LangChainTracer）都必须实现其方法。

回调的触发时机，严格绑定在Runnable的生命周期上：

• on_chain_start：RunnableSequence开始执行时；
• on_llm_start：ChatOpenAI准备发送请求时；
• on_retriever_end：VectorStoreRetriever返回结果后；
• on_chain_end：整个链完成时。

重要的是，callbacks是异步安全的。BaseCallbackHandler的on_llm_new_token等方法，被设计为可await，因此在astream模式下，token能实时推送至前端。

我们曾用callbacks实现“AI客服响应质量监控”：

• 自定义QualityCallbackHandler，在on_chain_end中分析output的长度、关键词密度、情感得分；
• 若得分低于阈值，自动触发on_chain_error，记录到告警系统；
• 所有回调通过config={"callbacks": [handler]}注入，无需修改业务链代码。

4.3messages与documents：数据流的血液与细胞

langchain_core.messages和langchain_core.documents定义了LangChain的两种核心数据载体：

• messages：BaseMessage及其子类（HumanMessage、AIMessage、SystemMessage），用于对话状态管理；
• documents：Document类，用于RAG场景的文本块表示。

它们的关键设计是不可变性（Immutable）。Document的page_content、metadata一旦创建，不可修改。这保证了在RunnableParallel中，多个分支读取同一Document时，不会因某一分支篡改内容而影响其他分支。

因此，当你需要“修改”Document时，必须创建新实例：

# ❌ 错误：直接修改 doc.metadata["score"] = 0.9 # ✅ 正确：创建新Document new_doc = Document( page_content=doc.page_content, metadata={**doc.metadata, "score": 0.9} )

这种设计看似繁琐，却是LangChain能支持RunnableParallel、RunnableMap等高级组合器的根基——数据流中的每个节点，都持有数据的“快照”，而非引用。

4.4prompts：数据流的起搏器

langchain_core.prompts模块负责将原始输入（如用户query）转化为模型可理解的messages序列。它的核心是BasePromptTemplate抽象类，所有模板（ChatPromptTemplate、PromptTemplate）都继承于此。

PromptTemplate的format方法，本质是Runnable的invoke实现。当你写：

template = PromptTemplate.from_template("Answer: {question}") chain = template | llm

template就是一个Runnable，其invoke接收{"question": "..."}，返回"Answer: ..."字符串；llm的invoke接收该字符串，返回AIMessage。

这种设计让Prompt成为可组合、可测试的单元。你可以单独测试template.invoke({"question": "test"})，确保格式正确，而不必启动LLM。

5. 架构陷阱：那些官方文档不会告诉你的“合理默认值”

LangChain的架构设计充满“合理默认值”，它们让新手快速上手，却在复杂场景中成为隐形陷阱。识别并主动覆盖这些默认值，是进阶的关键。

5.1RunnableParallel的并发默认值：不是CPU核心数，而是16

RunnableParallel的max_concurrency默认值是16，无论你的机器有多少CPU核心。这意味着，在8核机器上，并行8个LLM调用，会排队等待；在32核机器上，最多只并发16个。

为什么设为16？LangChain团队在issue中解释：这是基于OpenAI API的rate limit经验——大多数免费/入门级key的TPM（Tokens Per Minute）限制在10K左右，16并发足以打满带宽，再高反而触发限流。

但如果你用的是本地部署的Llama3-70B，max_concurrency=16会导致GPU显存爆炸。解决方案：

from langchain_core.runnables import RunnableParallel # 显式设置，避免默认陷阱 parallel_chain = RunnableParallel( summary=summary_chain, entities=entity_chain ).with_config({"max_concurrency": 4}) # 根据GPU显存调整

5.2ChatPromptTemplate的system message默认位置：不是开头，而是末尾

ChatPromptTemplate的format_messages方法，默认将systemmessage放在messages列表的最后，而非传统认知的开头。这是因为LangChain遵循OpenAI的chat.completionsAPI规范：systemmessage必须是第一个，但ChatPromptTemplate生成的messages是供ChatOpenAI内部处理的，它会在发送前自动重排。

但如果你直接用messages调用其他模型（如Ollama），顺序错误会导致system prompt被忽略。验证方法：

template = ChatPromptTemplate.from_messages([ ("system", "You are a helpful assistant."), ("human", "{input}") ]) msgs = template.format_messages(input="Hello") print([msg.type for msg in msgs]) # ['human', 'system'] —— system在末尾！

正确做法：显式指定位置，或用MessagesPlaceholder：

# 方案1：用placeholder确保system在开头 template = ChatPromptTemplate.from_messages([ ("system", "You are a helpful assistant."), MessagesPlaceholder(variable_name="history"), ("human", "{input}") ]) # 方案2：手动构造messages列表 from langchain_core.messages import SystemMessage, HumanMessage messages = [ SystemMessage(content="You are a helpful assistant."), HumanMessage(content="Hello") ]

5.3RunnableRetry的指数退避默认值：base=1.0，不是0.5

RunnableRetry的retry_delay默认是exponential_backoff(base=1.0)，即第一次重试延迟1秒，第二次2秒，第三次4秒……但很多开发者期望“快速失败”，希望第一次就重试。base=1.0意味着最小延迟1秒，对于毫秒级服务（如Redis缓存）过于保守。

调整方法：

from langchain_core.runnables import RunnableRetry from langchain_core.runnables.retry import exponential_backoff # 设置base=0.1，首次延迟0.1秒 retry_policy = RunnableRetry( runnable=my_chain, retry_errors=[ConnectionError], retry_delay=exponential_backoff(base=0.1) )

5.4config的recursion_limit默认值：20，不是无限

Runnable的config中recursion_limit默认为20，防止无限递归。在LangGraph中，state更新可能触发循环，20次是安全上限。但如果你的业务逻辑确实需要更深递归（如复杂决策树），必须显式提高：

chain.invoke(input, config={"recursion_limit": 50})

否则，第21次递归会抛RecursionError，且错误信息不提示recursion_limit，只显示“maximum recursion depth exceeded”，让人误以为是Python全局限制。

实测心得：我们在一个法律条款解析链中，因嵌套引用过深，触发了recursion_limit。调试时花了两天才定位到是LangChain的默认值，而非代码逻辑问题。建议：所有生产环境链，都在入口处显式设置recursion_limit，哪怕设为100，也比默认值更透明。

6. 架构演进：从LangChain到LangGraph，Runnable契约如何延续

LangGraph的出现，不是对LangChain的替代，而是对其Runnable契约的状态化扩展。理解这一点，才能平滑过渡，而非推倒重来。

6.1 LangGraph的State：不是新概念，而是Runnable的增强版

LangGraph的State，本质上是一个dict，但它被Runnable的input_schema和output_schema严格约束。例如：

from typing import TypedDict, Annotated from langgraph.graph import StateGraph from langchain_core.runnables import Runnable class GraphState(TypedDict): question: str context: list answer: str # 这个Runnable的input_type必须是GraphState def retrieve(state: GraphState) -> GraphState: # 从state["question"]检索，写入state["context"] return {"context": [...]} # LangGraph自动将retrieve包装为Runnable，其input_schema=GraphState workflow = StateGraph(GraphState) workflow.add_node("retrieve", retrieve)

这里，retrieve函数被LangGraph自动包装为Runnable，其input_schema就是GraphState，output_schema也是GraphState。LangGraph没有发明新契约，只是把Runnable的输入/输出契约，从“单次调用”扩展到“多次迭代”。

6.2 Edge与Conditional Edge：RunnableBranch的图形化表达

LangGraph的add_conditional_edges，对应RunnableBranch的RunnableBranch类。RunnableBranch的branches参数，是一个(condition, runnable)元组列表，而LangGraph的conditional_edge函数，就是condition的图形化封装。

# LangChain原生写法 from langchain_core.runnables import RunnableBranch branch = RunnableBranch( (lambda x: x["needs_rag"], rag_chain), (lambda x: x["needs_sql"], sql_chain), default_chain ) # LangGraph等价写法 def route(state: GraphState): if state["needs_rag"]: return "rag" elif state["needs_sql"]: return "sql" else: return "__end__" workflow.add_conditional_edges( "router", route, { "rag": "retrieve", "sql": "execute_sql", "__end__": END } )

两者本质相同：都是根据state的某个字段，选择下一个Runnable。LangGraph的优势在于可视化编排和循环控制，但底层仍是Runnable的组合。

6.3 Memory与CheckPoint：RunnableConfig的持久化延伸

LangGraph的checkpointer，是对RunnableConfig中metadata的持久化实现。每次app.invoke()，LangGraph会将config中的metadata（含run_id、thread_id）与state一起存入数据库。这使得Runnable的“上下文快照”能力，从单次请求，扩展到跨会话、跨重启。

因此，迁移策略很清晰：

• 保留所有现有Runnable：llm、retriever、parser等，无需修改；
• 将RunnableSequence替换为StateGraph：用add_node注册原有链；
• 用add_edge替代|：线性链直接add_edge("node1", "node2")；
• 用add_conditional_edges替代RunnableBranch：逻辑不变，只是写法更直观。

架构的连续性，正在于此：LangGraph没有废弃Runnable，而是让它在状态机中焕发新生。你今天写的prompt | llm | parser，明天就能无缝接入LangGraph的StateGraph，只需一行workflow.add_node("answer", prompt | llm | parser)。

我在实际项目中，曾用两周时间，将一个2000行的LangChain链，重构为LangGraph工作流。没有重写任何业务逻辑，只改变了编排方式。上线后，运维同学能直接看到状态流转图，故障定位时间从小时级降到分钟级。这印证了一点：好的架构，不是让你重写代码，而是让你重用代码。

最后分享一个小技巧：在LangGraph中调试时，不要只看app.stream()的输出，更要检查app.get_state(config)返回的Checkpoint。它包含了每一次invoke的config快照、state快照、metadata，是排查“为什么这里没走分支”的终极证据。毕竟，架构的威力，不在于它多炫酷，而在于它多诚实——它从不隐藏数据流向，只要你愿意去看。