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把“组合”从代码里拎出来变成可读、可测、可调试的表达式LCELLangChain 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_typestr).with_config(run_namemock).invoke(问题)。没有LCEL你得把整个链拆开、注释、重写测试桩有了LCEL调试就是切一刀的事。我团队有个硬性规定新写的链必须先用LCEL写完再考虑性能优化。因为LCEL写出来的链本身就是一份活的、可执行的架构文档。它强迫你思考“这个环节的输入应该是什么格式”、“下游组件能不能消化这个输出”而不是等跑起来报错才去查schema mismatch。2.3 为什么必须有“抽象层”——从LLM调用到应用落地的三道鸿沟LangChain的架构本质是在填平LLM能力与实际应用之间的三道鸿沟。第一道鸿沟叫输入鸿沟大模型只认token但你的用户输入是自然语言、文件、数据库查询结果、甚至语音转文本。LangChain用Document、BaseRetriever、BasePromptTemplate来标准化输入预处理——Document不是简单的textmetadata它自带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]返回AIMessageBaseLLM的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解析PDFmetadata里有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字段的类型和校验规则。比如一个天气查询Toolargs_schema必须声明location: str这样Agent生成的tool_input才会被自动校验和序列化。没有args_schemaAgentExecutor连参数都解析不了。我团队做过对比用StructuredTool带args_schema和Tool不带实现同一个功能前者在Agent调用时能自动捕获location为空的错误后者直接抛KeyError。Agent架构的价值不在于它多“智能”而在于它把“调用外部系统”这件事标准化成Tool协议并把“何时调、调谁、怎么调”这个决策逻辑交给LLM和PromptTemplate去动态生成。你不需要写if-else判断用户要不要查天气只需要注册一个WeatherToolAgent自己会决定。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、urlLLM必须能理解{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和contextStrOutputParser不行得用JsonOutputParser(pydantic_objectRAGOutput)Retriever返回的Documentmetadata字段必须包含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_objectRAGOutput) )注意三个关键点第一{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_keyquestion, history_messages_keychat_history )现在调用时传config{configurable: {session_id: abc123}}历史就自动管理了。整个骨架没有一行if、没有一行for全是声明式组合。这就是LCEL的力量骨架即契约契约即代码。4.3 第三步注入可观测性不是加日志是埋事件钩子可观测性不是事后补救而是从第一行invoke开始。LangChain的CallbackHandler是事件驱动的必须在invoke时显式传入。我们用LangChainTracer集成LangSmith和自定义LoggingCallbackHandlerimport 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(fChain start: {serialized[name]}, inputs{inputs}) def on_llm_end(self, response, **kwargs): self.logger.info(fLLM 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}, schemainput_schema) # 应该通过 # 验证输出 output_schema rag_chain.output_schema.schema() validate(instance{answer: test, sources: []}, schemaoutput_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耗时如果retriever500ms说明向量库没建好索引如果llm3000ms说明prompt太长或模型选错。第三关安全安检。Runnable默认不做输入清洗必须手动加RunnableLambdafrom 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必须配BaseLLMChatPromptTemplate输出List[BaseMessage]必须配ChatModel。混用必然报错。解决方案不是改LLM而是统一协议如果你用ChatModel推荐所有PromptTemplate必须是ChatPromptTemplate且retriever返回的Document要用format_document转成HumanMessage或SystemMessage如果你坚持用BaseLLMretriever返回的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” —— 不是没上报是没启用callbackretriever默认不触发CallbackHandler事件除非你显式调用retriever.with_config(callbacks[...])。但更推荐的做法是把retriever作为RunnableParallel的一部分RunnableParallel会自动将父链的callbacks透传给所有子组件。所以永远不要单独调用retriever.invoke()而是把它放在LCEL链里让|操作符帮你搞定回调透传。5.4 “Agent死循环” —— 不是LLM发疯是tool没返回有效结果AgentExecutor的循环终止条件是LLM生成Final Answer:。但如果Tool返回空字符串、或返回格式错误的JSONAgent会认为没得到结果继续循环。典型场景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我查BB结果触发CC失败要重试DD成功后发邮件并更新数据库”这种场景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%的接口用LCEL20%的核心业务流用LangGraph两者通过Runnable协议完美协同。我在实际项目中发现最有效的学习路径不是从文档开始而是从chain.get_graph().to_json()开始。每次写完一段LCEL立刻dump出它的graph JSON对照着看输入怎么进来、中间怎么流转、输出怎么出去。这个JSON就是你亲手写的架构图。它不漂亮但绝对真实它不抽象但直击本质。LangChain的架构从来不在纸上而在你每一行|、每一个RunnableLambda、每一次with_config里。当你不再问“LangChain怎么用”而是问“我的这个需求需要哪几个RUNNABLE协议来满足”你就真正入门了。