LangChain 框架深度解析从 LCEL 到 Agent 架构的核心原理摘要本文深入剖析 LangChain 框架的核心架构与实现原理涵盖 LCELLangChain Expression Language的声明式链式组合机制、Runnable 统一接口设计、Agent 工作流模式ReAct、Tool Calling以及 LangGraph 状态图编排系统。通过源码级分析揭示其设计思想帮助开发者掌握构建生产级 AI 应用架构的关键技术。引言LangChain 自 2022 年发布以来已成为构建 LLM 应用最广泛使用的开源框架之一。2025 年 LangChain 1.0 和 LangGraph 1.0 正式发布标志着框架进入生产就绪阶段。核心问题LangChain 如何实现模块化组件的统一编排LCEL 的声明式语法背后隐藏着什么设计模式Agent 如何实现自主决策与工具调用的闭环LangChain 与 LangGraph 的定位有何不同文章结构首先解析核心组件与 Runnable 接口深入 LCEL 组合机制然后剖析 Agent 架构与工作流模式最后介绍 LangGraph 的状态编排系统。核心组件架构组件层次结构LangChain 采用分层架构设计层次组件职责编排层LCEL / LangGraph组合、流程控制逻辑层Agents决策、工具调用模型层ModelsLLM 接口封装数据层Memory / Retrievers状态持久化、数据检索工具层Tools外部能力扩展Runnable 统一接口Runnable 是 LangChain 的基石接口定义了组件执行的标准协议classRunnable(Generic[Input,Output]):# 核心方法definvoke(self,input:Input,config:RunnableConfig)-Outputdefbatch(self,inputs:List[Input],config:RunnableConfig)-List[Output]defstream(self,input:Input,config:RunnableConfig)-Iterator[Output]# 异步方法asyncdefainvoke(self,input:Input,config:RunnableConfig)-Outputasyncdefabatch(self,inputs:List[Input])-List[Output]asyncdefastream(self,input:Input)-AsyncIterator[Output]# 组合方法defpipe(self,other:Runnable)-RunnableSequencedefbind(self,**kwargs)-Runnabledefwith_retry(self,retry_policy)-Runnabledefwith_fallbacks(self,fallbacks:List[Runnable])-Runnable设计思想统一协议所有组件PromptTemplate、LLM、Retriever、Tool都实现 Runnable实现可插拔架构多态执行invoke单次、batch批量、stream流式三种执行模式异步原生所有方法都有 async 版本支持高并发场景组合优先pipe 方法支持链式组合bind 方法支持参数绑定RunnableSequence 与 RunnableParallelRunnable 的组合通过两个核心原语实现RunnableSequence顺序执行# 使用 | 运算符构建顺序链sequenceRunnableLambda(lambdax:x1)|RunnableLambda(lambdax:x*2)sequence.invoke(1)# 输出: 4sequence.batch([1,2,3])# 输出: [4, 6, 8]RunnableParallel并行执行# 使用字典构建并行分支sequenceRunnableLambda(lambdax:x1)|{mul_2:RunnableLambda(lambdax:x*2),mul_5:RunnableLambda(lambdax:x*5),}sequence.invoke(1)# 输出: {mul_2: 4, mul_5: 10}关键要点Runnable 接口实现了一切皆可组合的设计哲学组合后的 Chain 自动继承 invoke/batch/stream/async 能力| 运算符语法糖使链式组合直观简洁LCEL声明式链式编排LCEL 语法解析LangChain Expression Language (LCEL) 是 Runnable 的声明式 DSLfromlangchain_core.promptsimportPromptTemplatefromlangchain_openaiimportChatOpenAIfromlangchain_core.output_parsersimportStrOutputParser# 定义组件promptPromptTemplate.from_template(Question:{question}Answer:Lets think step by step.)llmChatOpenAI(modelgpt-4,temperature0)parserStrOutputParser()# 使用 LCEL 组合chainprompt|llm|parser# 执行chain.invoke({question:How much is 22?})LCEL 内部实现机制当执行prompt | llm | parser时实际发生第一次 pipeprompt.pipe(llm)→ 创建RunnableSequence([prompt, llm])第二次 pipesequence.pipe(parser)→ 创建RunnableSequence([prompt, llm, parser])invoke 执行依次调用各组件前一组件输出作为后一组件输入流式执行优化LCEL 的 stream 方法不是简单迭代而是实现管道式流式传输# 流式输出每个组件产生的中间结果立即传递给下游forchunkinchain.stream({question:What is AI?}):print(chunk,end,flushTrue)内部实现使用RunnableSequence.stream()每个组件的stream()方法产生迭代器上游迭代器元素直接传递给下游。LCEL 高级组合模式1. 参数绑定bind# 绑定固定参数llm_with_stopllm.bind(stop[ ])chainprompt|llm_with_stop|parser2. 条件分支RunnableBranchfromlangchain_core.runnablesimportRunnableBranch branchRunnableBranch((lambdax:x[type]math,math_chain),(lambdax:x[type]text,text_chain),default_chain)3. 回退机制with_fallbacksrobust_chain(prompt|llm|parser).with_fallbacks([prompt|backup_llm|parser])4. 重试策略with_retryretry_chainchain.with_retry(stop_after_attempt3,wait_exponential_multiplier1000)关键要点LCEL 实现了声明式、可组合的链式编排| 运算符语法糖背后是 RunnableSequence 组合组合后的 Chain 自动支持流式、批量、异步执行Agent 架构深度解析Agent 核心概念Agent 是具有自主决策能力的系统核心特征特征描述自主决策LLM 决定执行哪些动作、顺序工具调用通过 Tool 接口调用外部能力状态感知记忆对话历史、中间结果反闭环观察工具返回结果调整后续决策Agent 工作流模式1. ReAct 模式Reasoning ActingReAct 是经典的 Agent 模式遵循思考-行动-观察循环Thought: 我需要查询天气信息 Action: search_weather(北京) Observation: 北京今天晴气温 25°C Thought: 我已获得天气信息可以回答用户 Answer: 北京今天天气晴朗气温 25°CReAct Agent 实现fromlangchain.agentsimportload_tools,initialize_agent,AgentTypefromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4,temperature0)toolsload_tools([serpapi,llm-math],llmllm)agentinitialize_agent(tools,llm,agentAgentType.ZERO_SHOT_REACT_DESCRIPTION,verboseTrue)agent.run(北京今天天气如何计算25的平方)2. Tool Calling 模式现代 LLM如 GPT-4、Claude原生支持 Function CallingAgent 可直接调用fromlangchain.agentsimportAgentExecutor,create_tool_calling_agentfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain.toolsimportTool# 定义工具defget_weather(city:str)-str:returnf{city}今天晴气温25°Ctools[Tool(nameget_weather,funcget_weather,description获取指定城市的天气信息)]# 创建 AgentllmChatOpenAI(modelgpt-4-turbo,temperature0)promptChatPromptTemplate.from_messages([(system,You are a helpful assistant.),(human,{input}),(placeholder,{agent_scratchpad}),])agentcreate_tool_calling_agent(llm,tools,prompt)agent_executorAgentExecutor(agentagent,toolstools)agent_executor.invoke({input:北京天气如何})Agent 执行循环Agent 的核心是执行循环Agent Loop┌─────────────┐ │ 用户输入 │ └──────┬──────┘ ↓ ┌─────────────┐ │ LLM 决策 │ ← 分析输入决定是否调用工具 └──────┬──────┘ ↓ 有工具调用 ├─ Yes ──→ ┌─────────────┐ │ │ 执行工具 │ │ └──────┬──────┘ │ ↓ │ ┌─────────────┐ │ │ 观察结果 │ → Tool Message │ └──────┬──────┘ │ ↓ │ ┌─────────────┐ │ │ 继续决策 │ ← 将观察结果加入对话 │ └──────┬──────┘ │ │ └─────────────────┘ └─ No ──→ ┌─────────────┐ │ 返回答案 │ └──────┬──────┘ ↓ ┌─────────────┐ │ 执行结束 │ └─────────────┘Tool 接口设计Tool 是 Agent 与外部世界交互的标准接口fromlangchain.toolsimporttooltooldefcalculate(expression:str)-float:计算数学表达式。 Args: expression: 数学表达式如 22 或 sqrt(16) returneval(expression)# Tool 的核心属性# - name: 工具名称# - description: 工具描述LLM 用于决策# - args_schema: 参数 SchemaJSON Schema 格式关键要点Agent 通过 LLM 自主决定执行路径ReAct 模式显式的思考-行动-观察循环Tool Calling 模式利用 LLM 原生 Function Calling执行循环LLM 决策 → 工具执行 → 观察反馈 → 继续决策LangGraph状态图编排系统LangChain vs LangGraph 定位特性LangChainLangGraph定位高层抽象框架低层编排引擎组合方式链式LCEL图式StateGraph控制流线性为主支持复杂循环、分支状态管理隐式Memory显式TypedDict State持久化需额外配置内置 Checkpointer适用场景快速原型、简单流程生产级复杂 AgentLangGraph 核心概念LangGraph 将 Agent 工作流建模为状态图fromlanggraph.graphimportStateGraph,START,ENDfromtyping_extensionsimportTypedDict,Annotatedimportoperator# 1. 定义状态classAgentState(TypedDict):messages:Annotated[list,operator.add]# 消息累加llm_calls:int# LLM 调用次数# 2. 定义节点Nodedefllm_node(state:AgentState)-dict:LLM 决策节点responsellm.invoke(state[messages])return{messages:[response],llm_calls:state.get(llm_calls,0)1}deftool_node(state:AgentState)-dict:工具执行节点results[]fortool_callinstate[messages][-1].tool_calls:resulttools[tool_call[name]].invoke(tool_call[args])results.append(ToolMessage(contentresult,tool_call_idtool_call[id]))return{messages:results}# 3. 定义边Edgedefshould_continue(state:AgentState)-str:条件边决定下一节点ifstate[messages][-1].tool_calls:returntool_nodereturnEND# 4. 构建图graphStateGraph(AgentState)graph.add_node(llm,llm_node)graph.add_node(tool,tool_node)graph.add_edge(START,llm)graph.add_conditional_edges(llm,should_continue,[tool,END])graph.add_edge(tool,llm)# 工具执行后返回 LLM# 5. 编译并执行agentgraph.compile()resultagent.invoke({messages:[HumanMessage(计算 25 的平方)]})LangGraph 架构要素1. State状态classState(TypedDict):messages:Annotated[list[AnyMessage],operator.add]# operator.add 表示新消息累加到列表2. Node节点函数接收当前 State返回 State 更新dict是 Agent 逻辑的封装单元3. Edge边普通边固定流转路径graph.add_edge(A, B)条件边根据状态动态路由graph.add_conditional_edges(A, router, [B, C, END])LangGraph 执行模型LangGraph 的执行模型START → llm_node → [条件判断] ├─ tool_calls 存在 → tool_node → llm_node (循环) └─ 无 tool_calls → END关键特性循环支持tool_node → llm_node形成闭环实现 Agent 的多轮决策状态持久化Checkpointer 可保存执行状态支持暂停恢复Human-in-the-Loop可在任意节点插入人工干预关键要点LangGraph 使用显式 State Node Edge 模型条件边实现复杂路由逻辑循环边支持 Agent 的多轮决策闭环Memory 与 Retrieval 系统Memory 架构LangChain Memory 系统管理对话上下文fromlangchain.memoryimportConversationBufferMemory memoryConversationBufferMemory(memory_keychat_history,return_messagesTrue)# Memory 自动管理对话历史chainConversationChain(llmllm,memorymemory)Memory 类型对比类型特点适用场景ConversationBufferMemory保存完整对话短对话ConversationBufferWindowMemory保留最近 N 轮控制长度ConversationSummaryMemoryLLM 摘要历史长对话VectorStoreMemory向量检索相关历史大规模历史Retriever 接口Retriever 是 Runnable 的特化实现用于数据检索fromlangchain_community.vectorstoresimportFAISSfromlangchain_openaiimportOpenAIEmbeddings vectorstoreFAISS.from_texts(documents,OpenAIEmbeddings())retrievervectorstore.as_retriever(search_kwargs{k:3})# Retriever 本身是 Runnabledocsretriever.invoke(query)# 返回相关文档LCEL 中集成 Retrieverfromlangchain_core.runnablesimportRunnableParallel# RAG Chain 示例rag_chain({context:retriever,question:RunnablePassthrough()}|prompt|llm|parser)实战案例构建完整 Agent场景描述构建一个具备计算、搜索、记忆能力的多功能 Agent。解决方案fromlangchain.toolsimporttoolfromlangchain_openaiimportChatOpenAIfromlangchain.agentsimportAgentExecutor,create_tool_calling_agentfromlangchain_core.promptsimportChatPromptTemplatefromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.graphimportStateGraph,START,END# 1. 定义工具tooldefcalculator(expression:str)-float:计算数学表达式returneval(expression)tooldefsearch(query:str)-str:搜索信息# 实际实现调用搜索 APIreturnf搜索结果:{query}# 2. 配置 LLMllmChatOpenAI(modelgpt-4-turbo,temperature0)tools[calculator,search]llm_with_toolsllm.bind_tools(tools)# 3. 使用 LangGraph 构建持久化 Agentfromlanggraph.graphimportMessagesStatedefagent_node(state:MessagesState):responsellm_with_tools.invoke(state[messages])return{messages:[response]}deftool_node(state:MessagesState):results[]fortcinstate[messages][-1].tool_calls:tool{t.name:tfortintools}[tc[name]]results.append(ToolMessage(contentstr(tool.invoke(tc[args])),tool_call_idtc[id]))return{messages:results}defshould_continue(state):ifstate[messages][-1].tool_calls:returntoolsreturnEND# 构建图builderStateGraph(MessagesState)builder.add_node(agent,agent_node)builder.add_node(tools,tool_node)builder.add_edge(START,agent)builder.add_conditional_edges(agent,should_continue)builder.add_edge(tools,agent)# 添加持久化checkpointerMemorySaver()agentbuilder.compile(checkpointercheckpointer)# 执行带线程 IDresultagent.invoke({messages:[HumanMessage(计算 15 * 3然后搜索 Python 教程)]},config{configurable:{thread_id:session-1}})效果评估工具调用正确性Agent 依次调用 calculator 和 search状态持久化可通过 thread_id 恢复对话流式输出支持实时输出中间结果总结核心要点回顾Runnable 接口LangChain 的统一执行协议实现 invoke/batch/stream/async 多态执行LCEL 语法声明式链式组合| 运算符语法糖自动继承执行能力Agent 架构自主决策 工具调用 反馈闭环ReAct 与 Tool Calling 两种模式LangGraph状态图编排系统显式 State Node Edge支持循环、持久化最佳实践建议简单链优先 LCEL线性流程使用 | 运算符快速组合复杂 Agent 用 LangGraph多轮决策、循环、条件路由用 StateGraph工具定义规范提供清晰的 description 和 args_schema流式优先长输出场景使用 stream 方法提升用户体验持久化配置生产环境使用 LangGraph Checkpointer 或外部存储扩展阅读LangChain 官方文档LangGraph 官方文档Runnable API 参考LangChain 1.0 发布说明参考资料LangChain Core ComponentsLangGraph Architecture OverviewThe Complete Guide to LangChain LangGraph 2025
LangChain 框架深度解析：从 LCEL 到 Agent 架构的核心原理
发布时间：2026/5/27 2:40:12
LangChain 框架深度解析从 LCEL 到 Agent 架构的核心原理摘要本文深入剖析 LangChain 框架的核心架构与实现原理涵盖 LCELLangChain Expression Language的声明式链式组合机制、Runnable 统一接口设计、Agent 工作流模式ReAct、Tool Calling以及 LangGraph 状态图编排系统。通过源码级分析揭示其设计思想帮助开发者掌握构建生产级 AI 应用架构的关键技术。引言LangChain 自 2022 年发布以来已成为构建 LLM 应用最广泛使用的开源框架之一。2025 年 LangChain 1.0 和 LangGraph 1.0 正式发布标志着框架进入生产就绪阶段。核心问题LangChain 如何实现模块化组件的统一编排LCEL 的声明式语法背后隐藏着什么设计模式Agent 如何实现自主决策与工具调用的闭环LangChain 与 LangGraph 的定位有何不同文章结构首先解析核心组件与 Runnable 接口深入 LCEL 组合机制然后剖析 Agent 架构与工作流模式最后介绍 LangGraph 的状态编排系统。核心组件架构组件层次结构LangChain 采用分层架构设计层次组件职责编排层LCEL / LangGraph组合、流程控制逻辑层Agents决策、工具调用模型层ModelsLLM 接口封装数据层Memory / Retrievers状态持久化、数据检索工具层Tools外部能力扩展Runnable 统一接口Runnable 是 LangChain 的基石接口定义了组件执行的标准协议classRunnable(Generic[Input,Output]):# 核心方法definvoke(self,input:Input,config:RunnableConfig)-Outputdefbatch(self,inputs:List[Input],config:RunnableConfig)-List[Output]defstream(self,input:Input,config:RunnableConfig)-Iterator[Output]# 异步方法asyncdefainvoke(self,input:Input,config:RunnableConfig)-Outputasyncdefabatch(self,inputs:List[Input])-List[Output]asyncdefastream(self,input:Input)-AsyncIterator[Output]# 组合方法defpipe(self,other:Runnable)-RunnableSequencedefbind(self,**kwargs)-Runnabledefwith_retry(self,retry_policy)-Runnabledefwith_fallbacks(self,fallbacks:List[Runnable])-Runnable设计思想统一协议所有组件PromptTemplate、LLM、Retriever、Tool都实现 Runnable实现可插拔架构多态执行invoke单次、batch批量、stream流式三种执行模式异步原生所有方法都有 async 版本支持高并发场景组合优先pipe 方法支持链式组合bind 方法支持参数绑定RunnableSequence 与 RunnableParallelRunnable 的组合通过两个核心原语实现RunnableSequence顺序执行# 使用 | 运算符构建顺序链sequenceRunnableLambda(lambdax:x1)|RunnableLambda(lambdax:x*2)sequence.invoke(1)# 输出: 4sequence.batch([1,2,3])# 输出: [4, 6, 8]RunnableParallel并行执行# 使用字典构建并行分支sequenceRunnableLambda(lambdax:x1)|{mul_2:RunnableLambda(lambdax:x*2),mul_5:RunnableLambda(lambdax:x*5),}sequence.invoke(1)# 输出: {mul_2: 4, mul_5: 10}关键要点Runnable 接口实现了一切皆可组合的设计哲学组合后的 Chain 自动继承 invoke/batch/stream/async 能力| 运算符语法糖使链式组合直观简洁LCEL声明式链式编排LCEL 语法解析LangChain Expression Language (LCEL) 是 Runnable 的声明式 DSLfromlangchain_core.promptsimportPromptTemplatefromlangchain_openaiimportChatOpenAIfromlangchain_core.output_parsersimportStrOutputParser# 定义组件promptPromptTemplate.from_template(Question:{question}Answer:Lets think step by step.)llmChatOpenAI(modelgpt-4,temperature0)parserStrOutputParser()# 使用 LCEL 组合chainprompt|llm|parser# 执行chain.invoke({question:How much is 22?})LCEL 内部实现机制当执行prompt | llm | parser时实际发生第一次 pipeprompt.pipe(llm)→ 创建RunnableSequence([prompt, llm])第二次 pipesequence.pipe(parser)→ 创建RunnableSequence([prompt, llm, parser])invoke 执行依次调用各组件前一组件输出作为后一组件输入流式执行优化LCEL 的 stream 方法不是简单迭代而是实现管道式流式传输# 流式输出每个组件产生的中间结果立即传递给下游forchunkinchain.stream({question:What is AI?}):print(chunk,end,flushTrue)内部实现使用RunnableSequence.stream()每个组件的stream()方法产生迭代器上游迭代器元素直接传递给下游。LCEL 高级组合模式1. 参数绑定bind# 绑定固定参数llm_with_stopllm.bind(stop[ ])chainprompt|llm_with_stop|parser2. 条件分支RunnableBranchfromlangchain_core.runnablesimportRunnableBranch branchRunnableBranch((lambdax:x[type]math,math_chain),(lambdax:x[type]text,text_chain),default_chain)3. 回退机制with_fallbacksrobust_chain(prompt|llm|parser).with_fallbacks([prompt|backup_llm|parser])4. 重试策略with_retryretry_chainchain.with_retry(stop_after_attempt3,wait_exponential_multiplier1000)关键要点LCEL 实现了声明式、可组合的链式编排| 运算符语法糖背后是 RunnableSequence 组合组合后的 Chain 自动支持流式、批量、异步执行Agent 架构深度解析Agent 核心概念Agent 是具有自主决策能力的系统核心特征特征描述自主决策LLM 决定执行哪些动作、顺序工具调用通过 Tool 接口调用外部能力状态感知记忆对话历史、中间结果反闭环观察工具返回结果调整后续决策Agent 工作流模式1. ReAct 模式Reasoning ActingReAct 是经典的 Agent 模式遵循思考-行动-观察循环Thought: 我需要查询天气信息 Action: search_weather(北京) Observation: 北京今天晴气温 25°C Thought: 我已获得天气信息可以回答用户 Answer: 北京今天天气晴朗气温 25°CReAct Agent 实现fromlangchain.agentsimportload_tools,initialize_agent,AgentTypefromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4,temperature0)toolsload_tools([serpapi,llm-math],llmllm)agentinitialize_agent(tools,llm,agentAgentType.ZERO_SHOT_REACT_DESCRIPTION,verboseTrue)agent.run(北京今天天气如何计算25的平方)2. Tool Calling 模式现代 LLM如 GPT-4、Claude原生支持 Function CallingAgent 可直接调用fromlangchain.agentsimportAgentExecutor,create_tool_calling_agentfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain.toolsimportTool# 定义工具defget_weather(city:str)-str:returnf{city}今天晴气温25°Ctools[Tool(nameget_weather,funcget_weather,description获取指定城市的天气信息)]# 创建 AgentllmChatOpenAI(modelgpt-4-turbo,temperature0)promptChatPromptTemplate.from_messages([(system,You are a helpful assistant.),(human,{input}),(placeholder,{agent_scratchpad}),])agentcreate_tool_calling_agent(llm,tools,prompt)agent_executorAgentExecutor(agentagent,toolstools)agent_executor.invoke({input:北京天气如何})Agent 执行循环Agent 的核心是执行循环Agent Loop┌─────────────┐ │ 用户输入 │ └──────┬──────┘ ↓ ┌─────────────┐ │ LLM 决策 │ ← 分析输入决定是否调用工具 └──────┬──────┘ ↓ 有工具调用 ├─ Yes ──→ ┌─────────────┐ │ │ 执行工具 │ │ └──────┬──────┘ │ ↓ │ ┌─────────────┐ │ │ 观察结果 │ → Tool Message │ └──────┬──────┘ │ ↓ │ ┌─────────────┐ │ │ 继续决策 │ ← 将观察结果加入对话 │ └──────┬──────┘ │ │ └─────────────────┘ └─ No ──→ ┌─────────────┐ │ 返回答案 │ └──────┬──────┘ ↓ ┌─────────────┐ │ 执行结束 │ └─────────────┘Tool 接口设计Tool 是 Agent 与外部世界交互的标准接口fromlangchain.toolsimporttooltooldefcalculate(expression:str)-float:计算数学表达式。 Args: expression: 数学表达式如 22 或 sqrt(16) returneval(expression)# Tool 的核心属性# - name: 工具名称# - description: 工具描述LLM 用于决策# - args_schema: 参数 SchemaJSON Schema 格式关键要点Agent 通过 LLM 自主决定执行路径ReAct 模式显式的思考-行动-观察循环Tool Calling 模式利用 LLM 原生 Function Calling执行循环LLM 决策 → 工具执行 → 观察反馈 → 继续决策LangGraph状态图编排系统LangChain vs LangGraph 定位特性LangChainLangGraph定位高层抽象框架低层编排引擎组合方式链式LCEL图式StateGraph控制流线性为主支持复杂循环、分支状态管理隐式Memory显式TypedDict State持久化需额外配置内置 Checkpointer适用场景快速原型、简单流程生产级复杂 AgentLangGraph 核心概念LangGraph 将 Agent 工作流建模为状态图fromlanggraph.graphimportStateGraph,START,ENDfromtyping_extensionsimportTypedDict,Annotatedimportoperator# 1. 定义状态classAgentState(TypedDict):messages:Annotated[list,operator.add]# 消息累加llm_calls:int# LLM 调用次数# 2. 定义节点Nodedefllm_node(state:AgentState)-dict:LLM 决策节点responsellm.invoke(state[messages])return{messages:[response],llm_calls:state.get(llm_calls,0)1}deftool_node(state:AgentState)-dict:工具执行节点results[]fortool_callinstate[messages][-1].tool_calls:resulttools[tool_call[name]].invoke(tool_call[args])results.append(ToolMessage(contentresult,tool_call_idtool_call[id]))return{messages:results}# 3. 定义边Edgedefshould_continue(state:AgentState)-str:条件边决定下一节点ifstate[messages][-1].tool_calls:returntool_nodereturnEND# 4. 构建图graphStateGraph(AgentState)graph.add_node(llm,llm_node)graph.add_node(tool,tool_node)graph.add_edge(START,llm)graph.add_conditional_edges(llm,should_continue,[tool,END])graph.add_edge(tool,llm)# 工具执行后返回 LLM# 5. 编译并执行agentgraph.compile()resultagent.invoke({messages:[HumanMessage(计算 25 的平方)]})LangGraph 架构要素1. State状态classState(TypedDict):messages:Annotated[list[AnyMessage],operator.add]# operator.add 表示新消息累加到列表2. Node节点函数接收当前 State返回 State 更新dict是 Agent 逻辑的封装单元3. Edge边普通边固定流转路径graph.add_edge(A, B)条件边根据状态动态路由graph.add_conditional_edges(A, router, [B, C, END])LangGraph 执行模型LangGraph 的执行模型START → llm_node → [条件判断] ├─ tool_calls 存在 → tool_node → llm_node (循环) └─ 无 tool_calls → END关键特性循环支持tool_node → llm_node形成闭环实现 Agent 的多轮决策状态持久化Checkpointer 可保存执行状态支持暂停恢复Human-in-the-Loop可在任意节点插入人工干预关键要点LangGraph 使用显式 State Node Edge 模型条件边实现复杂路由逻辑循环边支持 Agent 的多轮决策闭环Memory 与 Retrieval 系统Memory 架构LangChain Memory 系统管理对话上下文fromlangchain.memoryimportConversationBufferMemory memoryConversationBufferMemory(memory_keychat_history,return_messagesTrue)# Memory 自动管理对话历史chainConversationChain(llmllm,memorymemory)Memory 类型对比类型特点适用场景ConversationBufferMemory保存完整对话短对话ConversationBufferWindowMemory保留最近 N 轮控制长度ConversationSummaryMemoryLLM 摘要历史长对话VectorStoreMemory向量检索相关历史大规模历史Retriever 接口Retriever 是 Runnable 的特化实现用于数据检索fromlangchain_community.vectorstoresimportFAISSfromlangchain_openaiimportOpenAIEmbeddings vectorstoreFAISS.from_texts(documents,OpenAIEmbeddings())retrievervectorstore.as_retriever(search_kwargs{k:3})# Retriever 本身是 Runnabledocsretriever.invoke(query)# 返回相关文档LCEL 中集成 Retrieverfromlangchain_core.runnablesimportRunnableParallel# RAG Chain 示例rag_chain({context:retriever,question:RunnablePassthrough()}|prompt|llm|parser)实战案例构建完整 Agent场景描述构建一个具备计算、搜索、记忆能力的多功能 Agent。解决方案fromlangchain.toolsimporttoolfromlangchain_openaiimportChatOpenAIfromlangchain.agentsimportAgentExecutor,create_tool_calling_agentfromlangchain_core.promptsimportChatPromptTemplatefromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.graphimportStateGraph,START,END# 1. 定义工具tooldefcalculator(expression:str)-float:计算数学表达式returneval(expression)tooldefsearch(query:str)-str:搜索信息# 实际实现调用搜索 APIreturnf搜索结果:{query}# 2. 配置 LLMllmChatOpenAI(modelgpt-4-turbo,temperature0)tools[calculator,search]llm_with_toolsllm.bind_tools(tools)# 3. 使用 LangGraph 构建持久化 Agentfromlanggraph.graphimportMessagesStatedefagent_node(state:MessagesState):responsellm_with_tools.invoke(state[messages])return{messages:[response]}deftool_node(state:MessagesState):results[]fortcinstate[messages][-1].tool_calls:tool{t.name:tfortintools}[tc[name]]results.append(ToolMessage(contentstr(tool.invoke(tc[args])),tool_call_idtc[id]))return{messages:results}defshould_continue(state):ifstate[messages][-1].tool_calls:returntoolsreturnEND# 构建图builderStateGraph(MessagesState)builder.add_node(agent,agent_node)builder.add_node(tools,tool_node)builder.add_edge(START,agent)builder.add_conditional_edges(agent,should_continue)builder.add_edge(tools,agent)# 添加持久化checkpointerMemorySaver()agentbuilder.compile(checkpointercheckpointer)# 执行带线程 IDresultagent.invoke({messages:[HumanMessage(计算 15 * 3然后搜索 Python 教程)]},config{configurable:{thread_id:session-1}})效果评估工具调用正确性Agent 依次调用 calculator 和 search状态持久化可通过 thread_id 恢复对话流式输出支持实时输出中间结果总结核心要点回顾Runnable 接口LangChain 的统一执行协议实现 invoke/batch/stream/async 多态执行LCEL 语法声明式链式组合| 运算符语法糖自动继承执行能力Agent 架构自主决策 工具调用 反馈闭环ReAct 与 Tool Calling 两种模式LangGraph状态图编排系统显式 State Node Edge支持循环、持久化最佳实践建议简单链优先 LCEL线性流程使用 | 运算符快速组合复杂 Agent 用 LangGraph多轮决策、循环、条件路由用 StateGraph工具定义规范提供清晰的 description 和 args_schema流式优先长输出场景使用 stream 方法提升用户体验持久化配置生产环境使用 LangGraph Checkpointer 或外部存储扩展阅读LangChain 官方文档LangGraph 官方文档Runnable API 参考LangChain 1.0 发布说明参考资料LangChain Core ComponentsLangGraph Architecture OverviewThe Complete Guide to LangChain LangGraph 2025
