ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

LangChain架构本质:RUNNABLE协议与LCEL组合范式

LangChain架构本质:RUNNABLE协议与LCEL组合范式

1. 这不是“框架说明书”,而是你第一次真正看懂LangChain的起点

如果你刚点开LangChain官网,被满屏的Chain、Agent、Retriever、Tool、Callback、Memory、LLM这些词砸得头晕,翻了几页文档发现全是API签名和类型定义,最后默默关掉页面——别急,这不是你理解力的问题,是LangChain从设计第一天起就没打算让你“按图索骥”地学。它根本不是一个传统意义上的“框架”,而是一套面向LLM应用开发者的协议层与组合范式。我带过十几支团队落地LangChain项目,最常听到的抱怨不是“功能不行”,而是“不知道该从哪下手”、“改一行代码,整个链路就崩”。原因很简单:LangChain不教你怎么写代码,它先逼你重构思维——把“我要调用一个大模型”这件事,拆解成“输入怎么来、中间怎么流、输出怎么走、错误怎么兜、状态怎么存”这五个不可绕开的原子问题。而所谓“整体架构”,就是这五个问题在LangChain里被标准化、可插拔、可复用的解法集合。RUNNABLE不是个接口名,它是整个体系的“宪法条款”;LCEL不是语法糖,它是把“组合逻辑”从代码里抽出来、变成声明式表达的唯一出口;LangGraph不是LangChain的升级版,它是当RUNNABLE遇上状态机时,被迫长出的第三条腿。你看不到“架构图”,是因为它根本不是画出来的,而是由你写的每一行|(管道符)、每一个.with_config()、每一次RunnableLambda的封装,一砖一瓦垒出来的。所以这篇总览不列模块图、不背概念定义,只做一件事:告诉你LangChain的骨架是怎么撑起来的,哪些骨头必须连着,哪些关节可以拆换,以及——为什么你昨天写的那段看似能跑的代码,今天加了个Memory就报错。

2. 架构不是画出来的,是RUNNABLE协议撑起来的

2.1 RUNNABLE:不是接口,是契约,更是底线

很多人把Runnable当成Java里的Interface或Python里的ABC,以为只是个类型标记。错了。Runnable是LangChain整个生态的最小共识单元,它的存在意义,不是为了让你“实现它”,而是为了让你“相信它”。当你看到一个对象标着Runnable[Input, Output],你立刻能断定三件事:第一,它一定有.invoke()方法,且入参类型是Input,返回类型是Output;第二,它一定支持.batch()批量调用,且行为可预测;第三,它一定可以通过.with_config()注入运行时上下文(比如trace_id、max_retries)。这三条,就是LangChain敢说“所有组件可互换”的底气。举个反例:你写了个自定义LLM类,继承了BaseLLM,但没实现.batch(),或者.invoke()返回类型和泛型声明不一致——它在LangChain里就是个“非法居民”,哪怕它能跑通单次调用,一旦放进RunnableParallel里就会当场崩溃。我见过最典型的踩坑案例,是某团队封装了一个HTTP客户端调用私有模型,为了省事直接返回dict,结果在LCEL链里和其他str输出的组件|在一起时,类型检查直接报错:“Expected str, got dict”。修复方案不是加类型转换,而是重写invoke,让它严格返回str,并在Input泛型里明确声明接受str或dict。RUNNABLE的强制契约,本质上是在LLM这种非确定性系统上,强行建立确定性边界。它不保证结果正确,但保证调用方式统一、错误路径清晰、组合逻辑可推演。这才是“架构”的起点——不是模块怎么分,而是“谁跟谁说话,用什么语言说”。

2.2 LCEL:把“组合”从代码里拎出来,变成可读、可测、可调试的表达式

LCEL(LangChain Expression Language)常被误读为“链式调用的语法糖”。其实它干的是更底层的事:把数据流的拓扑结构,从 imperative(命令式)代码里解放出来,变成 declarative(声明式)表达式。你看这段代码:

chain = ( {"context": retriever, "question": RunnablePassthrough()} | prompt_template | llm | StrOutputParser() )

表面是链式调用,内核却是函数式编程的组合子(combinator)。|不是方法调用,而是RunnableSequence的构造操作符;{}不是字典,而是RunnableMap的声明语法;RunnablePassthrough()不是个空转组件,而是identity函数的具象化。LCEL的威力,在于它让“数据怎么流动”这件事,脱离了具体执行细节,变成一张可静态分析的图。你可以用chain.get_graph().draw_mermaid()生成流程图(虽然我们禁用mermaid,但原理一样),也可以用chain.input_schema和chain.output_schema精确知道这个链的输入/输出契约。更重要的是,它天然支持局部调试:你想单独测retriever返回的内容是否合理?直接chain.first.invoke("问题")就行;想跳过LLM模拟返回?chain.with_types(input_type=str).with_config(run_name="mock").invoke("问题")。没有LCEL,你得把整个链拆开、注释、重写测试桩;有了LCEL,调试就是切一刀的事。我团队有个硬性规定:新写的链,必须先用LCEL写完,再考虑性能优化。因为LCEL写出来的链,本身就是一份活的、可执行的架构文档。它强迫你思考“这个环节的输入应该是什么格式”、“下游组件能不能消化这个输出”,而不是等跑起来报错才去查schema mismatch。

2.3 为什么必须有“抽象层”?——从LLM调用到应用落地的三道鸿沟

LangChain的架构,本质是在填平LLM能力与实际应用之间的三道鸿沟。第一道鸿沟叫输入鸿沟:大模型只认token,但你的用户输入是自然语言、文件、数据库查询结果、甚至语音转文本。LangChain用Document、BaseRetriever、BasePromptTemplate来标准化输入预处理——Document不是简单的text+metadata,它自带page_content和metadata分离,让检索器能只索引内容,而把来源、时间戳等元信息留给后续逻辑;BasePromptTemplate强制你把变量名、格式、分隔符都声明清楚,避免拼接字符串时的引号灾难。第二道鸿沟叫状态鸿沟:单次调用无状态,但客服对话、多轮问答、数据分析都需要记忆。LangChain不提供“全局state”,而是用BaseChatMessageHistory抽象出“谁能存、怎么存、存多久”,让你自由选InMemoryChatMessageHistory(开发)、PostgresChatMessageHistory(生产)、甚至自己实现RedisChatMessageHistory。关键在于,所有组件都通过config["configurable"]["session_id"]来获取历史,而不是硬编码数据库连接。第三道鸿沟叫可观测鸿沟:LLM调用失败不报错,只是返回乱码;耗时波动大,无法归因。LangChain用CallbackHandler抽象出“事件钩子”,on_llm_start、on_chain_end、on_tool_error……每个事件都带run_id、parent_run_id、tags,让你能把一次完整调用的全链路日志、指标、trace串起来。这三层抽象,不是为了炫技,而是为了让“调用大模型”这件事,能像调用数据库一样,有schema、有事务、有监控。没有它们,你写的每个应用都是孤岛;有了它们,才能谈复用、谈编排、谈治理。

3. 核心模块解剖:不是“组件列表”,而是“协议实现者图谱”

3.1 LLM:不只是模型,是“响应生成器”的协议实现

LangChain里的BaseLLM,名字叫LLM,实则是响应生成协议(Response Generation Protocol)的实现者。它的核心契约只有两条:invoke(input: str) -> str和stream(input: str) -> Iterator[str]。注意,输入是str,不是dict,不是list,不是pydantic.BaseModel——这是硬性约定。为什么?因为所有上游组件(如PromptTemplate)的输出,最终都必须能喂给BaseLLM.invoke()。这就倒逼PromptTemplate必须输出str,倒逼RunnableParallel的输出必须能join()成str。我见过最危险的实践,是有人把BaseLLM子类的invoke改成接收dict,里面塞system_prompt、user_message、history……然后在内部拼接。这直接破坏了协议,导致它无法和任何标准PromptTemplate组合。正确的做法是:用ChatPromptTemplate(它输出List[BaseMessage])配合ChatModel(它接受List[BaseMessage]),或者用StringPromptTemplate(输出str)配合BaseLLM。ChatModel和BaseLLM是平行协议,不是父子关系。ChatModel的invoke接收List[BaseMessage],返回AIMessage;BaseLLM的invoke接收str,返回str。选哪个,取决于你的输入结构。这个选择,不是技术偏好,而是架构决策——它决定了你的整个链路是走“消息流”还是“字符串流”。ChatModel更适合对话场景,BaseLLM更适合单次生成。混用?可以,但必须用RunnableLambda做显式转换,比如RunnableLambda(lambda x: "\n".join([msg.content for msg in x])),把消息列表转成字符串。这种转换不是hack,而是协议桥接的显式声明。

3.2 Retrievers:不是“搜索器”,是“相关性排序器”的协议

BaseRetriever的契约比BaseLLM更微妙:invoke(query: str) -> List[Document]。关键在List[Document]——它不承诺排序,不承诺数量,不承诺相关性阈值。它只承诺:给你一个query,还你一堆Document。这就解释了为什么VectorStoreRetriever默认返回4个结果,而BM25Retriever可能返回10个;为什么SelfQueryRetriever能根据语义过滤,而TimeWeightedVectorStoreRetriever会按时间衰减打分。它们都是BaseRetriever的合法实现,但行为差异巨大。真正的架构约束,在于Document本身。Document的page_content字段,是所有下游组件(尤其是PromptTemplate)默认读取的主体内容;metadata字段,则是留给RunnableLambda做后处理的“富信息区”。比如,你用WebBaseLoader加载网页,metadata里自动带source、title、date;用PyPDFLoader解析PDF,metadata里有page、total_pages。这些信息不参与向量化,但能在PromptTemplate里用{document.metadata.title}引用。我团队有个血泪教训:早期用Chroma做向量库,add_documents时没传metadata,结果RAG链里所有引用{document.metadata.source}的地方都报KeyError。修复不是改模板,而是重构Document生成逻辑,确保metadata必填。Retriever的协议价值,正在于此——它把“怎么找”和“找回来怎么用”彻底解耦。你可以今天用ElasticSearchRetriever,明天换成PineconeRetriever,只要它们都返回符合Document契约的列表,上层链路完全不用动。

3.3 Chains:不是“流程”,是“组合策略”的声明式表达

Chain这个词最容易误导人。LangChain里没有“Chain类”,只有RunnableSequence、RunnableParallel、RunnableBranch这些具体的组合器。SequentialChain、TransformChain这些老类,早已被LCEL取代。现在的“链”,就是|操作符构建的Runnable实例。RunnableSequence的契约是:输入喂给第一个组件,输出喂给第二个,以此类推,最终返回最后一个组件的输出。它不关心中间结果长什么样,只保证类型传递。RunnableParallel的契约是:把输入同时喂给多个组件,等全部返回后,合并成dict。这里的关键是“合并规则”:默认是dict,但你可以用RunnableAssign定制,比如{"answer": chain, "sources": retriever}。RunnableBranch的契约是:根据条件函数的返回值,选择执行哪个分支。条件函数必须返回str(分支名)或Runnable(分支本身)。我见过最精妙的RunnableBranch用法,是做“意图识别路由”:条件函数用一个小的分类模型判断用户query是"qa"、"summarize"还是"translate",然后分别路由到对应的Runnable链。这种路由不是if-else硬编码,而是Runnable的原生能力。Chain的本质,是把控制流(sequence)、数据流(parallel)、决策流(branch)这三种基本编程范式,映射成Runnable的组合操作。它不替代业务逻辑,而是为业务逻辑提供可组合、可嵌套、可复用的容器。一个RunnableBranch里,完全可以嵌套一个RunnableSequence,再嵌套一个RunnableParallel——这就是LCEL的嵌套能力,也是架构伸缩性的来源。

3.4 Tools & Agents:不是“智能体”,是“工具调用协议”的执行引擎

Tool和Agent是LangChain里最易被神化的部分。BaseTool的契约极其简单:invoke(input: str) -> str。它不关心你调的是API、数据库还是本地脚本,只要输入是str,输出是str,它就认。Agent的契约同样简单:invoke(input: Dict[str, Any]) -> Dict[str, Any]。输入是{"input": "用户问题", "chat_history": [...]},输出是{"output": "最终回答"}。真正的复杂性,在于AgentExecutor——它才是那个把Tool、LLM、PromptTemplate粘合成“能自主决策的智能体”的胶水。AgentExecutor的协议是:它接收一个Agent(负责决策)和一个Tool列表(负责执行),然后循环执行“LLM生成tool_call -> 执行tool -> 把结果喂回LLM”这个过程,直到LLM决定"Final Answer"。这里的关键约束是Tool的args_schema:它必须是一个pydantic.BaseModel,定义input字段的类型和校验规则。比如,一个天气查询Tool,args_schema必须声明location: str,这样Agent生成的tool_input才会被自动校验和序列化。没有args_schema,AgentExecutor连参数都解析不了。我团队做过对比:用StructuredTool(带args_schema)和Tool(不带)实现同一个功能,前者在Agent调用时能自动捕获location为空的错误,后者直接抛KeyError。Agent架构的价值,不在于它多“智能”,而在于它把“调用外部系统”这件事,标准化成Tool协议,并把“何时调、调谁、怎么调”这个决策逻辑,交给LLM和PromptTemplate去动态生成。你不需要写if-else判断用户要不要查天气,只需要注册一个WeatherTool,Agent自己会决定。

4. 实操全景:从零搭建一个可调试、可监控、可上线的RAG链

4.1 第一步:定义输入/输出契约,不是写代码,是画接口图

在敲任何一行代码前,先用纸笔(或白板)画出你的RAG链的端到端契约。假设我们要做一个“技术文档问答助手”,输入是用户问题str,输出是带引用的答案str。那么契约就是:

  • 输入:{"question": str}
  • 输出:{"answer": str, "sources": List[Dict[str, str]]}(sources里包含title、url、snippet)

这个契约决定了所有组件的选型。Retriever必须返回Document,且Document.metadata里要有title、url;LLM必须能理解{context}和{question}的模板;OutputParser必须能把LLM的原始输出,结构化成{"answer": ..., "sources": ...}。我坚持让团队先写pydantic.BaseModel定义输入输出:

from pydantic import BaseModel, Field class RAGInput(BaseModel): question: str = Field(..., description="用户提出的技术问题") class RAGOutput(BaseModel): answer: str = Field(..., description="简洁准确的回答") sources: list[dict] = Field(..., description="引用的文档来源列表,每个含title/url/snippet")

然后用RunnableWithMessageHistory包装整个链,确保configurable里有session_id。这一步看似繁琐,但它堵死了90%的后期返工——因为所有组件的input_schema和output_schema,都必须和这个契约对齐。PromptTemplate的变量名必须是question和context;StrOutputParser不行,得用JsonOutputParser(pydantic_object=RAGOutput);Retriever返回的Document,metadata字段必须包含title和url。契约先行,是LangChain项目不崩盘的第一铁律。

4.2 第二步:用LCEL搭骨架,拒绝“先写再组合”

骨架代码必须严格遵循LCEL语法,禁用任何class MyChain(Chain)的旧写法。我们的RAG链骨架如下:

from langchain_core.runnables import RunnablePassthrough, RunnableParallel from langchain_core.output_parsers import JsonOutputParser from langchain_core.prompts import ChatPromptTemplate # 1. 定义PromptTemplate,严格匹配RAGInput契约 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个资深技术文档助手。请基于以下上下文回答问题,答案必须简洁准确,并在末尾列出引用的文档标题和URL。"), ("human", "上下文:{context}\n问题:{question}") ]) # 2. 构建核心链:输入 -> 检索 -> 注入上下文 -> 调用LLM -> 解析输出 rag_chain = ( # 输入:RAGInput -> 提取question {"question": RunnablePassthrough(), "context": retriever} | prompt | llm | JsonOutputParser(pydantic_object=RAGOutput) )

注意三个关键点:第一,{"question": ..., "context": ...}是RunnableParallel的声明,它把question原样透传,同时用retriever生成context;第二,prompt接收dict,所以retriever必须返回List[Document],prompt内部会自动format;第三,JsonOutputParser的pydantic_object必须是前面定义的RAGOutput,这样输出类型才严格受控。这个骨架,已经能跑通基础流程。但还没完——要加上RunnableWithMessageHistory支持多轮:

from langchain_core.runnables.history import RunnableWithMessageHistory # 用lambda函数从config中提取session_id def get_session_history(session_id: str): return InMemoryChatMessageHistory() # 生产环境换成Redis rag_chain_with_history = RunnableWithMessageHistory( rag_chain, get_session_history, input_messages_key="question", history_messages_key="chat_history" )

现在,调用时传config={"configurable": {"session_id": "abc123"}},历史就自动管理了。整个骨架,没有一行if、没有一行for,全是声明式组合。这就是LCEL的力量:骨架即契约,契约即代码。

4.3 第三步:注入可观测性,不是加日志,是埋事件钩子

可观测性不是事后补救,而是从第一行invoke开始。LangChain的CallbackHandler是事件驱动的,必须在invoke时显式传入。我们用LangChainTracer(集成LangSmith)和自定义LoggingCallbackHandler:

import logging from langchain.callbacks.base import BaseCallbackHandler class LoggingCallbackHandler(BaseCallbackHandler): def __init__(self): self.logger = logging.getLogger("rag_chain") def on_chain_start(self, serialized, inputs, **kwargs): self.logger.info(f"Chain start: {serialized['name']}, inputs={inputs}") def on_llm_end(self, response, **kwargs): self.logger.info(f"LLM end: {response.llm_output.get('token_usage', {})}") # 调用时传入 result = rag_chain_with_history.invoke( {"question": "LangChain的RUNNABLE协议是什么?"}, config={ "configurable": {"session_id": "test_001"}, "callbacks": [LangChainTracer(), LoggingCallbackHandler()] } )

关键点在于callbacks必须是list,且每个CallbackHandler必须实现对应事件方法。LangChainTracer会自动上报到LangSmith,生成完整的trace图;LoggingCallbackHandler则把关键事件打到日志。注意on_llm_end里取response.llm_output['token_usage'],这是OpenAI等模型返回的原生token统计,不是LangChain计算的。可观测性埋点,必须深入到每个组件的生命周期,而不是只在链首尾加log。这样,当retriever慢、llm卡、parser失败时,你都能在trace里精准定位到哪个节点、哪个run_id出了问题。

4.4 第四步:上线前的三道安检——类型、性能、安全

上线前,必须过三关。第一关:类型安检。用rag_chain.input_schema.schema()和rag_chain.output_schema.schema()导出JSON Schema,用jsonschema.validate()验证真实输入输出:

import jsonschema from jsonschema import validate # 验证输入 input_schema = rag_chain.input_schema.schema() validate(instance={"question": "test"}, schema=input_schema) # 应该通过 # 验证输出 output_schema = rag_chain.output_schema.schema() validate(instance={"answer": "test", "sources": []}, schema=output_schema) # 应该通过

第二关:性能安检。用langchain_community.utils.timing测各环节耗时:

from langchain_community.utils.timing import time_it # 测retriever time_it(retriever.invoke)("LangChain RUNNABLE") # 返回 (result, elapsed_ms) # 测整个链 time_it(rag_chain.invoke)({"question": "test"}) # 返回 (result, elapsed_ms)

重点看retriever和llm的P95耗时,如果retriever>500ms,说明向量库没建好索引;如果llm>3000ms,说明prompt太长或模型选错。第三关:安全安检。Runnable默认不做输入清洗,必须手动加RunnableLambda:

from langchain_core.runnables import RunnableLambda def sanitize_input(inputs: dict) -> dict: # 基础清洗:去空格、截断超长输入 question = inputs.get("question", "").strip()[:1000] if not question: raise ValueError("Question cannot be empty") return {"question": question} safe_rag_chain = RunnableLambda(sanitize_input) | rag_chain_with_history

这三道安检,不是可选项,是上线红线。我团队的CI/CD流水线里,这三关任一失败,自动阻断发布。

5. 常见问题与避坑指南:那些文档里不会写的实战真相

5.1 “类型错误:Expected str, got dict” —— 不是bug,是契约警告

这是新手遇到最多的问题,99%源于PromptTemplate和LLM的协议错配。StringPromptTemplate输出str,必须配BaseLLM;ChatPromptTemplate输出List[BaseMessage],必须配ChatModel。混用必然报错。解决方案不是改LLM,而是统一协议:

  • 如果你用ChatModel(推荐),所有PromptTemplate必须是ChatPromptTemplate,且retriever返回的Document要用format_document转成HumanMessage或SystemMessage;
  • 如果你坚持用BaseLLM,retriever返回的Document必须用RunnableLambda转成str,比如lambda docs: "\n\n".join([d.page_content for d in docs])。

提示:用chain.get_graph().to_json()查看链的输入/输出类型,比猜快十倍。

5.2 “Memory不生效” —— 不是配置错,是session_id没传对

RunnableWithMessageHistory依赖configurable["session_id"]。常见错误有三:第一,调用时没传config;第二,config里"configurable"键名写错(比如写成"config");第三,get_session_history函数里,session_id参数名和configurable里不一致。最隐蔽的错误是:前端传session_id是字符串,但后端get_session_history函数签名写成def get_session_history(session_id: int),导致类型转换失败,session_id变成None,所有历史都存到None这个key下。解决方案:在get_session_history开头加日志,打印session_id类型和值。

5.3 “LangSmith trace里看不到retriever” —— 不是没上报,是没启用callback

retriever默认不触发CallbackHandler事件,除非你显式调用retriever.with_config(callbacks=[...])。但更推荐的做法是:把retriever作为RunnableParallel的一部分,RunnableParallel会自动将父链的callbacks透传给所有子组件。所以,永远不要单独调用retriever.invoke(),而是把它放在LCEL链里,让|操作符帮你搞定回调透传。

5.4 “Agent死循环” —— 不是LLM发疯,是tool没返回有效结果

AgentExecutor的循环终止条件,是LLM生成"Final Answer:"。但如果Tool返回空字符串、或返回格式错误的JSON,Agent会认为没得到结果,继续循环。典型场景:WeatherTool调用API失败,返回"",Agent就卡在"Thought: I need to call the weather tool... Action: weather Action Input: {...}"无限重复。解决方案:在Tool.invoke()里加防御性编程,确保即使API失败,也返回"Error: weather service unavailable"这样的有效字符串,并在Agent的system prompt里明确要求:“如果tool返回error,必须在Final Answer中告知用户”。

5.5 “LCEL链太慢” —— 不是语法问题,是并行没开对

RunnableParallel默认是并发执行,但如果你的retriever和llm都在同一个线程池里,反而会竞争资源。正确做法:给retriever和llm分别配置不同的thread_pool,或用asyncio异步执行。对于AsyncRetriever,必须用await rag_chain.ainvoke(),不能用invoke()。性能瓶颈往往不在LLM,而在retriever的向量相似度计算。解决方案:对retriever加cache,用InMemoryCache或RedisCache缓存query -> List[Document]结果,命中率高的场景能提速3-5倍。

6. 架构演进:从LCEL到LangGraph,不是替代,是补位

LangGraph的出现,不是因为LangChain不行了,而是因为LCEL遇到了它无法优雅解决的问题:有状态的、循环的、分支复杂的长期运行工作流。LCEL擅长“一次调用,线性流转”,但面对“用户问A,我查B,B结果触发C,C失败要重试D,D成功后发邮件并更新数据库”这种场景,LCEL的RunnableBranch就显得力不从心——它只能做单次分支,无法维持状态、无法循环、无法持久化中间结果。LangGraph用StateGraph解决了这个问题:它把工作流定义成一个有向图,每个节点是一个Runnable,每条边是一个condition函数,状态(State)在节点间流转。State是一个pydantic.BaseModel,你可以定义messages: List[BaseMessage]、tool_calls: List[dict]、retry_count: int等字段,condition函数根据State字段决定走向哪个节点。LangGraph不是LangChain的2.0,它是LangChain的“状态机插件”。你可以把一个LangGraph workflow,当作一个Runnable,无缝接入LCEL链里:

from langgraph.graph import StateGraph # 定义State class GraphState(TypedDict): messages: list[BaseMessage] retry_count: int # 构建Graph workflow = StateGraph(GraphState) workflow.add_node("call_llm", call_llm_node) workflow.add_node("call_tool", call_tool_node) workflow.add_conditional_edges( "call_llm", should_call_tool, { "tool": "call_tool", "end": END } ) app = workflow.compile() # 接入LCEL链 full_chain = ( {"question": RunnablePassthrough()} | prompt | app # 这里app就是一个Runnable | final_parser )

所以,选LangChain还是LangGraph?答案是:LCEL做短平快的API服务,LangGraph做长周期的业务流程。一个客服系统,用户提问走LCEL RAG链;但用户要求“帮我重置密码”,这个跨系统、需人工审核、有状态的流程,就交给LangGraph。架构演进,从来不是推倒重来,而是补位增效。我团队的实践是:80%的接口用LCEL,20%的核心业务流用LangGraph,两者通过Runnable协议完美协同。

我在实际项目中发现,最有效的学习路径不是从文档开始,而是从chain.get_graph().to_json()开始。每次写完一段LCEL,立刻dump出它的graph JSON,对照着看输入怎么进来、中间怎么流转、输出怎么出去。这个JSON,就是你亲手写的架构图。它不漂亮,但绝对真实;它不抽象,但直击本质。LangChain的架构,从来不在纸上,而在你每一行|、每一个RunnableLambda、每一次with_config里。当你不再问“LangChain怎么用”,而是问“我的这个需求,需要哪几个RUNNABLE协议来满足”,你就真正入门了。

返回列表