你有没有过这样的经历想用大模型做个能真正干活的应用比如让它帮你分析文档、自动处理工单或者做个能聊天的智能客服。你兴致勃勃地打开文档照着教程跑通了第一个“Hello World”示例感觉一切尽在掌握。但当你试图把它变成一个能稳定运行、处理复杂逻辑、甚至能记住上下文的应用时却发现代码迅速膨胀状态管理混乱回调函数满天飞调试起来像在迷宫里找出口。这不是你的问题。早期的 AI 应用开发就像用一堆乐高积木搭建城堡但缺少一张清晰的建筑图纸和稳定的连接件。你手里有强大的大模型LLM有各种工具Tools也有存储Vector DB但如何让它们协同工作处理分支、循环、等待和状态持久化这就是LangChain和LangGraph要解决的核心问题。它们不是两个并列的框架而是一个从“链式思维”到“图式思维”的进化是 AI 应用从玩具走向工程的关键一步。很多人把 LangChain 当作一个“工具调用库”把 LangGraph 当作一个“画流程图的工具”这其实低估了它们真正的价值。它们的核心是为不确定的、基于自然语言的交互过程提供确定性的、可编程的编排框架。今天我们不谈那些浮于表面的概念直接深入到 2026 年这个节点从工程实践的角度拆解如何用 LangChain LangGraph 构建真正可用的 Agent、RAG 和 MCP 应用。你会发现重点不在于调用哪个 API而在于如何设计一个健壮、可维护、可扩展的工作流。1. 重新理解 LangChain 与 LangGraph从“链条”到“大脑”在深入实战之前我们必须先打破一个常见的误解LangGraph 不是 LangChain 的替代品而是它的“操作系统升级”。理解这一点决定了你能否用好这套工具。1.1 LangChain优秀的“连接器”与“标准化”尝试LangChain 最初解决的是一个“连接”问题。在大模型生态早期开发者需要手动拼接Prompt 模板把用户输入、上下文、指令拼成一个字符串。模型调用处理不同模型供应商OpenAI, Anthropic, 本地模型的 API 差异。输出解析把模型返回的非结构化文本解析成结构化的数据如 JSON、列表。工具调用让模型决定何时、如何调用外部工具搜索、计算、数据库。LangChain 通过LCELLangChain Expression Language将这些环节串联成“链”Chain。它的巨大贡献在于标准化。它定义了Runnable接口让 Prompt、LLM、Parser、Tool 都成为可以组合的组件。这就像为乐高积木定义了统一的接口让拼接变得容易。但是链式结构有其天然局限线性思维一个输入经过一系列步骤得到一个输出。难以处理“根据中间结果选择不同路径”的情况。状态管理困难链中的状态传递是隐式的复杂的多轮对话或长期任务的状态管理会变得非常棘手。缺乏控制流难以实现条件判断if-else、循环for/while、并行执行、等待外部事件等逻辑。当你试图用纯 LangChain 构建一个复杂的客服 Agent需要根据用户问题决定是查知识库、转人工还是询问更多信息时代码很快就会变成一堆嵌套的回调和条件判断难以阅读和维护。1.2 LangGraph为 AI 应用引入“确定性工作流引擎”LangGraph 的诞生正是为了弥补链的不足。它的核心抽象是图Graph和状态State。图Graph由节点Nodes和边Edges组成。节点代表一个执行单元如调用 LLM、运行工具边代表执行路径。这让你可以直观地设计包含分支、循环、并发的复杂工作流。状态State一个贯穿整个图执行过程的共享数据存储。所有节点都读取和更新这个状态。这解决了链式结构中状态传递混乱的问题。最关键的思想转变是LangGraph 将 AI 应用从一个“黑箱函数调用”变成了一个“白箱状态机”。你不再只是问模型“接下来怎么办”而是设计好所有可能的路径节点让模型根据当前状态State来决定走哪条路边。程序的流程是确定的但路径的选择是动态的、由模型驱动的。用一个类比来理解LangChain像是一本标准操作手册告诉你第一步做什么第二步做什么每一步都有标准工具。LangGraph像是一个智能调度中心它有一个中央控制面板State看着当前情况用户问题、已有信息、工具结果按照你预设的交通规则Graph指挥不同的工人Nodes去干活并决定下一步派谁去。所以在 2026 年的开发实践中我们的基础认知应该是用 LangChain 来标准化你的“积木块”Prompt, LLM, Tools, Retrievers用 LangGraph 来设计和运行你的“建筑图纸”工作流。2. 构建你的第一个智能体Agent超越简单的工具调用网上大多数 Agent 教程止步于“让模型调用搜索工具”。这只是一个开始。一个真正的、有用的 Agent需要具备决策能力、记忆能力和异常处理能力。我们用 LangGraph 来一步步实现。2.1 定义状态Agent 的“记忆黑板”状态是 LangGraph 的灵魂。设计一个好的状态结构后续开发会事半功倍。对于一个通用对话 Agent状态通常包括from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 对话历史LangGraph 提供了便捷的消息管理 messages: Annotated[List, add_messages] # 用户当前输入的问题 user_input: str # Agent 的“思考过程”或中间步骤用于调试和观察 scratchpad: List[str] # 本次对话中已经调用过的工具及其结果避免重复调用 used_tools: List[dict] # 最终要返回给用户的结果 final_output: str # 一个标志位控制图何时结束运行 should_continue: boolAnnotated和add_messages是 LangGraph 提供的语法糖它自动帮你处理消息列表的追加非常方便。scratchpad和used_tools是工程上的好习惯它们让 Agent 的“思考”过程变得可追溯、可调试。2.2 创建节点Agent 的“功能模块”节点是执行具体任务的地方。每个节点接收整个状态并返回更新后的状态。1. 路由节点Router Node决定下一步做什么这是 Agent 的“大脑”。它查看当前对话历史和用户输入决定下一步是直接回答、调用工具还是结束对话。from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 定义路由提示词 router_prompt ChatPromptTemplate.from_messages([ (system, 你是一个助手。根据对话历史和用户最新问题决定下一步行动。 可选项 - direct_answer: 如果你能直接回答用户问题无需额外信息。 - use_tool: 如果你需要调用工具如搜索、计算来获取信息。 - end_conversation: 如果用户的问题已解决或对话可以结束。 只返回选项名称。), (placeholder, {messages}), # LangGraph 会自动填充消息历史 ]) llm ChatOpenAI(modelgpt-4o-mini, temperature0) router_chain router_prompt | llm def router_node(state: AgentState) - dict: 路由节点分析状态返回下一步动作 # 准备路由决策的输入 router_input {messages: state[messages]} decision router_chain.invoke(router_input).content.strip().lower() # 根据决策更新状态中的“下一步”标志 new_state state.copy() if decision direct_answer: new_state[next_action] answer elif decision use_tool: new_state[next_action] call_tool elif decision end_conversation: new_state[should_continue] False new_state[final_output] 对话结束。 else: # 默认降级处理 new_state[next_action] answer new_state[scratchpad].append(f路由决策异常: {decision}降级为直接回答。) return new_state2. 工具调用节点Tool Node执行具体操作这里我们集成 LangChain 定义的工具。from langchain.tools import Tool from langchain_community.tools import DuckDuckGoSearchRun from langchain_community.utilities import ArxivAPIWrapper # 定义工具 search DuckDuckGoSearchRun() arxiv ArxivAPIWrapper() tools [ Tool( nameweb_search, funcsearch.run, description当需要获取最新的、实时的或未包含在知识库中的信息时使用此工具。输入是一个搜索查询词。 ), Tool( namearxiv_search, funcarxiv.run, description当需要查找学术论文、研究资料时使用此工具。输入是论文标题、作者或关键词。 ) ] def tool_node(state: AgentState) - dict: 工具调用节点让LLM选择并调用工具 from langchain.agents import create_tool_calling_agent, AgentExecutor # 1. 创建Agent它负责让LLM决定用哪个工具、输入是什么 agent_prompt ChatPromptTemplate.from_messages([...]) # 省略具体提示词 agent create_tool_calling_agent(llmllm, toolstools, promptagent_prompt) # 2. 执行Agent agent_executor AgentExecutor(agentagent, toolstools, verboseFalse) # 取最新的用户消息作为输入 latest_input state[messages][-1].content if state[messages] else state[user_input] tool_result agent_executor.invoke({input: latest_input}) # 3. 更新状态 new_state state.copy() new_state[scratchpad].append(f调用工具: {tool_result.get(intermediate_steps, [])}) # 将工具返回的结果添加到对话历史中让后续节点能看到 new_state[messages].append((assistant, f我查到的信息是{tool_result[output]})) # 工具调用后默认回到路由节点决定下一步 new_state[next_action] route return new_state3. 回答节点Answer Node生成最终回复def answer_node(state: AgentState) - dict: 回答节点基于当前所有信息生成给用户的回复 answer_prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的助手。请根据对话历史和所有可用信息生成友好、准确的回复。), (placeholder, {messages}), ]) answer_chain answer_prompt | llm response answer_chain.invoke({messages: state[messages]}) new_state state.copy() new_state[messages].append((assistant, response.content)) new_state[final_output] response.content # 回答完毕后一般也回到路由节点等待用户下一个输入 new_state[next_action] route return new_state2.3 编排成图定义 Agent 的“行为逻辑”有了节点我们需要用边把它们连接起来形成完整的工作流。from langgraph.graph import StateGraph, END # 1. 创建图 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(router, router_node) workflow.add_node(call_tool, tool_node) workflow.add_node(answer, answer_node) # 3. 设置入口点 workflow.set_entry_point(router) # 4. 定义条件边Conditional Edge根据状态决定下一步走向 def decide_next_step(state: AgentState) - str: 决定下一个节点是什么 # 如果应该结束则走向 END if not state.get(should_continue, True): return END # 否则看 next_action 字段 next_action state.get(next_action, route) if next_action call_tool: return call_tool elif next_action answer: return answer else: # 默认回到路由节点 return router # 5. 为路由节点添加条件边 workflow.add_conditional_edges( router, decide_next_step, { call_tool: call_tool, answer: answer, END: END, router: router # 自循环等待下一个用户输入 } ) # 6. 为工具节点和回答节点添加固定边执行完后固定回到路由节点 workflow.add_edge(call_tool, router) workflow.add_edge(answer, router) # 7. 编译图 agent_app workflow.compile()现在一个具备基本决策能力的 Agent 就构建好了。你可以通过agent_app.invoke()来运行它。与简单的链式调用相比这个 Agent 的优势在于流程清晰所有可能的状态转移都在图中明确定义。易于调试通过scratchpad和used_tools可以完整追溯 Agent 的思考过程。易于扩展要增加新功能如访问数据库只需添加新的工具和节点并在路由逻辑中增加选项即可。3. 实现生产级 RAG不仅仅是向量检索RAG检索增强生成是当前 AI 应用的核心模式。但很多实现停留在“切块-嵌入-检索-生成”的简单流水线忽略了生产环境中的关键问题检索质量、上下文管理和生成幻觉。我们用 LangChain 组件 LangGraph 工作流来构建一个更健壮的 RAG 系统。3.1 文档处理与索引质量始于源头检索的质量很大程度上取决于索引的质量。LangChain 提供了丰富的文档加载器和文本分割器但关键是如何组合。from langchain_community.document_loaders import PyPDFLoader, UnstructuredFileLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings def build_rag_index(doc_paths: List[str], persist_directory: str ./chroma_db): 构建 RAG 向量索引生产环境建议 all_splits [] for path in doc_paths: # 1. 根据文件类型选择加载器 if path.endswith(.pdf): loader PyPDFLoader(path) else: # Unstructured 能处理多种格式docx, pptx, html等 loader UnstructuredFileLoader(path) docs loader.load() # 2. 智能文本分割 # 关键不要只用字符长度分割要尝试保留语义段落 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 块大小 chunk_overlap200, # 重叠部分避免切断上下文 separators[\n\n, \n, 。, , , , , , ], # 中文友好分隔符 length_functionlen, ) splits text_splitter.split_documents(docs) all_splits.extend(splits) # 3. 去重可选但重要 # 简单基于内容哈希去重 unique_texts {} unique_splits [] for split in all_splits: text_hash hash(split.page_content[:500]) # 取前500字符哈希 if text_hash not in unique_texts: unique_texts[text_hash] True unique_splits.append(split) # 4. 创建向量存储 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore Chroma.from_documents( documentsunique_splits, embeddingembeddings, persist_directorypersist_directory ) vectorstore.persist() return vectorstore关键点重叠分割避免关键信息被切断。语义分割优先按段落、标题等自然边界分割。去重避免相同内容多次检索浪费上下文窗口。3.2 检索与重排提升召回精度简单返回相似度最高的 K 个片段往往不够。我们需要“检索-重排”两步走。from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor from langchain.retrievers.ensemble import EnsembleRetriever from langchain_community.retrievers import BM25Retriever def create_advanced_retriever(vectorstore): 创建高级检索器混合检索 重排 # 1. 基础向量检索器 vector_retriever vectorstore.as_retriever( search_typesimilarity, search_kwargs{k: 10} # 先召回较多候选 ) # 2. 关键词检索器BM25作为补充 # 注意需要将文档转换为纯文本列表 from langchain_community.retrievers.bm25 import BM25Retriever as LangchainBM25 # 这里简化处理实际需从vectorstore中提取文本 # bm25_retriever LangchainBM25.from_documents(docs) # 3. 混合检索器Ensemble # ensemble_retriever EnsembleRetriever( # retrievers[vector_retriever, bm25_retriever], # weights[0.7, 0.3] # 向量检索权重更高 # ) # 4. 上下文压缩重排器用LLM对召回结果进行精炼只保留相关部分 llm ChatOpenAI(temperature0, modelgpt-4o-mini) compressor LLMChainExtractor.from_llm(llm) compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrievervector_retriever # 可以用 ensemble_retriever ) return compression_retriever这种设计的好处是先用向量/关键词检索广泛召回再用一个轻量级 LLM 对每个片段进行“相关性打分”或“内容提取”只把最相关的部分送入最终的生成环节极大节省了上下文窗口。3.3 用 LangGraph 编排 RAG 工作流简单的 RAG 是一条链但复杂的 RAG 可能涉及多轮追问、结果验证、来源追溯等。这时就需要图。from typing import TypedDict import operator class RAGState(TypedDict): question: str retrieved_docs: List # 检索到的文档 condensed_context: str # 压缩/重排后的上下文 answer: str need_follow_up: bool # 是否需要追问澄清 follow_up_question: str # 追问的问题 def retrieve_node(state: RAGState): 检索节点 retriever create_advanced_retriever(vectorstore) docs retriever.invoke(state[question]) return {retrieved_docs: docs} def condense_node(state: RAGState): 上下文压缩节点如果检索结果太多进行总结提炼 if len(state[retrieved_docs]) 5: # 调用LLM总结这些文档的核心信息 condense_prompt ChatPromptTemplate.from_template( 请根据以下文档片段提炼出与问题“{question}”最相关的核心信息合成一段连贯的文本。 文档片段 {docs} 核心信息提炼) chain condense_prompt | llm condensed chain.invoke({question: state[question], docs: state[retrieved_docs]}) return {condensed_context: condensed.content} else: return {condensed_context: \n\n.join([doc.page_content for doc in state[retrieved_docs]])} def generate_node(state: RAGState): 生成答案节点 qa_prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的助手请严格根据提供的上下文信息回答问题。如果上下文没有明确答案请说“根据已有信息无法回答”不要编造信息。), (human, 上下文{context}\n\n问题{question}) ]) qa_chain qa_prompt | llm answer qa_chain.invoke({context: state[condensed_context], question: state[question]}) return {answer: answer.content} def validate_node(state: RAGState): 答案验证节点检查生成答案是否基于上下文是否需要追问 validation_prompt ChatPromptTemplate.from_template( 请判断以下答案是否严格基于提供的上下文。同时判断用户的问题是否模糊、需要进一步澄清。 上下文{context} 问题{question} 答案{answer} 请用JSON格式回答{{is_grounded: true/false, need_clarification: true/false, clarification_question: 如果需要澄清请提出一个问题}} ) chain validation_prompt | llm judgment chain.invoke(state) # 解析JSON结果更新状态 # ... 解析逻辑 # 如果 need_clarification 为 True则更新 follow_up_question return {need_follow_up: True, follow_up_question: 您能具体说明一下XX方面吗} # 构建 RAG 图 rag_workflow StateGraph(RAGState) rag_workflow.add_node(retrieve, retrieve_node) rag_workflow.add_node(condense, condense_node) rag_workflow.add_node(generate, generate_node) rag_workflow.add_node(validate, validate_node) rag_workflow.set_entry_point(retrieve) rag_workflow.add_edge(retrieve, condense) rag_workflow.add_edge(condense, generate) rag_workflow.add_edge(generate, validate) # 条件边验证后如果不需要追问结束如果需要则进入一个“等待用户澄清”的节点或循环。 def after_validate(state: RAGState): if state.get(need_follow_up): return ask_for_clarification # 假设有这样一个节点 else: return END rag_workflow.add_conditional_edges(validate, after_validate) rag_app rag_workflow.compile()这个 RAG 工作流实现了检索增强使用高级检索器。上下文管理在文档过多时进行压缩。答案验证检查幻觉并判断是否需要追问用户澄清。流程可观测每个节点的输入输出都清晰可见。这才是面向生产的 RAG它不仅仅是检索和生成而是一个受控的、可观测的、具备反馈循环的问答系统。4. 集成 MCPModel Context Protocol连接外部世界的“标准插座”MCP 是一个新兴但非常重要的协议它旨在标准化 AI 应用与外部工具、数据源之间的通信方式。你可以把它理解为 AI 世界的“USB-C 接口”。之前每个工具都需要为每个 AI 框架LangChain, LlamaIndex, AutoGen写一套适配器。有了 MCP工具只需实现一次 MCP 服务端Server任何支持 MCP 的客户端如 LangGraph都可以即插即用。4.1 MCP 核心概念MCP Server一个提供工具Tools和资源Resources的进程。例如一个数据库 MCP Server 可以提供“执行 SQL 查询”的工具和“表结构”资源。MCP ClientLangGraph 可以作为 MCP Client动态发现并调用 Server 提供的工具。Stdio/SSE 传输Client 和 Server 通过标准输入输出或 Server-Sent Events 通信无需 HTTP API。4.2 在 LangGraph 中使用 MCP 工具假设我们有一个本地的“文件系统 MCP Server”它提供了list_directory和read_file两个工具。# 假设我们使用 langchain-mcp 适配库请关注官方或社区更新 # 以下代码展示概念性流程 import asyncio from langchain_mcp import MCPServer # 1. 启动或连接一个 MCP Server # 通常通过子进程启动一个 Server 二进制文件 async def setup_mcp_tools(): # 创建 MCP 客户端连接到 Server # 这里概念性展示实际 API 可能不同 # async with MCPServer(path/to/your/mcp-server-binary) as client: # # 2. 发现 Server 提供的所有工具 # tools await client.list_tools() # # 3. 将这些工具转换成 LangChain Tool 对象 # langchain_tools [] # for tool_info in tools: # langchain_tools.append( # Tool( # nametool_info.name, # funclambda params, tool_infotool_info: asyncio.run(client.call_tool(tool_info.name, params)), # descriptiontool_info.description # ) # ) # return langchain_tools # 示例返回 return [ Tool( namelist_directory, funclambda path: fListing dir: {path}, # 模拟函数 description列出指定目录下的文件和文件夹 ), Tool( nameread_file, funclambda filepath: fContents of {filepath}, # 模拟函数 description读取指定文件的内容 ) ] # 在 LangGraph 的 Agent 中我们可以动态加载这些工具 class DynamicAgentState(TypedDict): messages: List available_tools: List[Tool] # 状态中保存可用工具列表 async def dynamic_agent_workflow(): # 初始化时或运行时加载 MCP 工具 mcp_tools await setup_mcp_tools() # 创建图并将工具传递给相关节点 workflow StateGraph(DynamicAgentState) # 定义一个节点它可以使用 state 中的 available_tools def agent_node(state: DynamicAgentState): # 这里的 agent 可以使用 state[available_tools] 来调用工具 # ... agent 逻辑 pass # ... 构建图MCP 的优势解耦工具开发与 AI 应用开发分离。动态性Agent 可以在运行时发现新工具无需重启。标准化统一的工具描述、调用和结果返回格式。安全性可以通过 Server 实现权限控制和审计。对于企业环境MCP 意味着你可以让运维团队维护一个“数据库 MCP Server”让业务团队维护一个“CRM MCP Server”AI 应用开发者无需关心底层连接细节只需像调用本地函数一样使用这些工具。这是构建复杂、安全的企业级 Agent 的关键基础设施。5. 从开发到部署工程化实践与避坑指南构建一个能跑的 Demo 和构建一个能服务成百上千用户的生产系统是两回事。以下是基于经验的关键实践。5.1 状态持久化与长期记忆LangGraph 的State默认在内存中。生产环境需要持久化。方案一使用Checkpointer。LangGraph 内置了检查点机制可以将状态保存到数据库如 PostgreSQL, Redis。方案二自定义持久化节点。在图的关键节点如一轮对话结束将状态序列化后存入你的业务数据库。from langgraph.checkpoint import MemorySaver # 使用内存检查点适合开发 memory MemorySaver() app workflow.compile(checkpointermemory) # 使用时会传入一个 thread_id用于恢复状态 config {configurable: {thread_id: user_123_session_456}} initial_state {messages: [(user, 你好)]} result app.invoke(initial_state, configconfig) # 下次可以用相同的 thread_id 恢复对话5.2 超时、重试与容错网络调用和模型调用可能失败。为 LLM 调用和工具调用设置超时使用asyncio.timeout或tenacity库进行重试。图的错误处理LangGraph 允许你为节点设置try...except将错误信息捕获并放入状态引导流程走向一个“错误处理节点”而不是整个应用崩溃。def safe_tool_node(state: State): try: # 可能失败的工具调用 result some_unreliable_api(state[input]) return {output: result} except TimeoutError: return {error: 工具调用超时, next_action: handle_error} except Exception as e: return {error: str(e), next_action: handle_error}5.3 监控、日志与可观测性这是生产系统的生命线。结构化日志在每个节点的开始和结束记录日志包含节点名、状态摘要、耗时、错误信息。使用logging库并输出 JSON 格式便于接入 ELK 或 Datadog。追踪Tracing利用 LangSmithLangChain 官方平台或 OpenTelemetry 对每次调用进行追踪可视化整个图的执行路径查看每个节点的输入输出。这对于调试复杂工作流至关重要。关键指标记录每个请求的 token 消耗、耗时、工具调用次数、缓存命中率等。5.4 性能优化缓存对 LLM 的相同提示词、对检索器的相同查询进行缓存。LangChain 内置了InMemoryCache、RedisCache等。异步执行如果图中有可以并行执行的节点如同时调用多个不依赖的工具使用langgraph.graph的StateGraph的add_edge和并发控制来实现。精简上下文这是成本控制和效果提升的核心。严格过滤检索结果使用LLMChainExtractor等压缩技术只把必要信息送给 LLM。5.5 安全与权限工具权限不是所有 Agent 都能调用所有工具。根据用户身份或会话上下文动态过滤可用的工具列表state[available_tools]。输入输出过滤对用户输入和模型输出进行内容安全过滤防止提示词注入或生成有害内容。数据隔离在多租户环境中确保通过thread_id或其他标识严格隔离不同用户的数据和状态。6. 总结LangChain LangGraph 的思维转变回顾整篇文章我们从 LangChain 与 LangGraph 的本质区别开始到构建具备决策能力的 Agent实现生产级 RAG集成标准化工具协议 MCP最后探讨工程化实践。这条路径的核心其实是一场思维转变从“如何调用模型”转变为“如何设计一个由模型驱动的、确定性的工作流”。LangChain 给了你标准化的积木而 LangGraph 给了你设计图纸和组装流水线。在 2026 年AI 应用的竞争将不再是“谁能调通 API”而是“谁能设计出更稳健、更高效、更易维护的智能工作流”。所以当你下次再启动一个 AI 项目时不要一上来就写llm.invoke()。先拿出一张白纸或者打开绘图工具问自己几个问题我的应用有哪些状态用户输入、对话历史、中间结果、工具输出……有哪些节点理解意图、检索、推理、执行工具、生成回复、验证……节点之间如何流转什么条件下走哪条路错误可能发生在哪里如何恢复如何让这个流程被观察和调试把这些问题想清楚画成图再用 LangGraph 把它实现出来。这才是通往下一代 AI 应用的正确路径。工具和技术迭代很快但这种基于状态和流程的系统设计能力才是你真正需要积累的核心资产。 
长河君 · 2026/8/25 3:59:52 · 阅读 1.2k LangChain与LangGraph实战：构建可编排的AI智能体与生产级RAG系统
你有没有过这样的经历想用大模型做个能真正干活的应用比如让它帮你分析文档、自动处理工单或者做个能聊天的智能客服。你兴致勃勃地打开文档照着教程跑通了第一个“Hello World”示例感觉一切尽在掌握。但当你试图把它变成一个能稳定运行、处理复杂逻辑、甚至能记住上下文的应用时却发现代码迅速膨胀状态管理混乱回调函数满天飞调试起来像在迷宫里找出口。这不是你的问题。早期的 AI 应用开发就像用一堆乐高积木搭建城堡但缺少一张清晰的建筑图纸和稳定的连接件。你手里有强大的大模型LLM有各种工具Tools也有存储Vector DB但如何让它们协同工作处理分支、循环、等待和状态持久化这就是LangChain和LangGraph要解决的核心问题。它们不是两个并列的框架而是一个从“链式思维”到“图式思维”的进化是 AI 应用从玩具走向工程的关键一步。很多人把 LangChain 当作一个“工具调用库”把 LangGraph 当作一个“画流程图的工具”这其实低估了它们真正的价值。它们的核心是为不确定的、基于自然语言的交互过程提供确定性的、可编程的编排框架。今天我们不谈那些浮于表面的概念直接深入到 2026 年这个节点从工程实践的角度拆解如何用 LangChain LangGraph 构建真正可用的 Agent、RAG 和 MCP 应用。你会发现重点不在于调用哪个 API而在于如何设计一个健壮、可维护、可扩展的工作流。1. 重新理解 LangChain 与 LangGraph从“链条”到“大脑”在深入实战之前我们必须先打破一个常见的误解LangGraph 不是 LangChain 的替代品而是它的“操作系统升级”。理解这一点决定了你能否用好这套工具。1.1 LangChain优秀的“连接器”与“标准化”尝试LangChain 最初解决的是一个“连接”问题。在大模型生态早期开发者需要手动拼接Prompt 模板把用户输入、上下文、指令拼成一个字符串。模型调用处理不同模型供应商OpenAI, Anthropic, 本地模型的 API 差异。输出解析把模型返回的非结构化文本解析成结构化的数据如 JSON、列表。工具调用让模型决定何时、如何调用外部工具搜索、计算、数据库。LangChain 通过LCELLangChain Expression Language将这些环节串联成“链”Chain。它的巨大贡献在于标准化。它定义了Runnable接口让 Prompt、LLM、Parser、Tool 都成为可以组合的组件。这就像为乐高积木定义了统一的接口让拼接变得容易。但是链式结构有其天然局限线性思维一个输入经过一系列步骤得到一个输出。难以处理“根据中间结果选择不同路径”的情况。状态管理困难链中的状态传递是隐式的复杂的多轮对话或长期任务的状态管理会变得非常棘手。缺乏控制流难以实现条件判断if-else、循环for/while、并行执行、等待外部事件等逻辑。当你试图用纯 LangChain 构建一个复杂的客服 Agent需要根据用户问题决定是查知识库、转人工还是询问更多信息时代码很快就会变成一堆嵌套的回调和条件判断难以阅读和维护。1.2 LangGraph为 AI 应用引入“确定性工作流引擎”LangGraph 的诞生正是为了弥补链的不足。它的核心抽象是图Graph和状态State。图Graph由节点Nodes和边Edges组成。节点代表一个执行单元如调用 LLM、运行工具边代表执行路径。这让你可以直观地设计包含分支、循环、并发的复杂工作流。状态State一个贯穿整个图执行过程的共享数据存储。所有节点都读取和更新这个状态。这解决了链式结构中状态传递混乱的问题。最关键的思想转变是LangGraph 将 AI 应用从一个“黑箱函数调用”变成了一个“白箱状态机”。你不再只是问模型“接下来怎么办”而是设计好所有可能的路径节点让模型根据当前状态State来决定走哪条路边。程序的流程是确定的但路径的选择是动态的、由模型驱动的。用一个类比来理解LangChain像是一本标准操作手册告诉你第一步做什么第二步做什么每一步都有标准工具。LangGraph像是一个智能调度中心它有一个中央控制面板State看着当前情况用户问题、已有信息、工具结果按照你预设的交通规则Graph指挥不同的工人Nodes去干活并决定下一步派谁去。所以在 2026 年的开发实践中我们的基础认知应该是用 LangChain 来标准化你的“积木块”Prompt, LLM, Tools, Retrievers用 LangGraph 来设计和运行你的“建筑图纸”工作流。2. 构建你的第一个智能体Agent超越简单的工具调用网上大多数 Agent 教程止步于“让模型调用搜索工具”。这只是一个开始。一个真正的、有用的 Agent需要具备决策能力、记忆能力和异常处理能力。我们用 LangGraph 来一步步实现。2.1 定义状态Agent 的“记忆黑板”状态是 LangGraph 的灵魂。设计一个好的状态结构后续开发会事半功倍。对于一个通用对话 Agent状态通常包括from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 对话历史LangGraph 提供了便捷的消息管理 messages: Annotated[List, add_messages] # 用户当前输入的问题 user_input: str # Agent 的“思考过程”或中间步骤用于调试和观察 scratchpad: List[str] # 本次对话中已经调用过的工具及其结果避免重复调用 used_tools: List[dict] # 最终要返回给用户的结果 final_output: str # 一个标志位控制图何时结束运行 should_continue: boolAnnotated和add_messages是 LangGraph 提供的语法糖它自动帮你处理消息列表的追加非常方便。scratchpad和used_tools是工程上的好习惯它们让 Agent 的“思考”过程变得可追溯、可调试。2.2 创建节点Agent 的“功能模块”节点是执行具体任务的地方。每个节点接收整个状态并返回更新后的状态。1. 路由节点Router Node决定下一步做什么这是 Agent 的“大脑”。它查看当前对话历史和用户输入决定下一步是直接回答、调用工具还是结束对话。from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 定义路由提示词 router_prompt ChatPromptTemplate.from_messages([ (system, 你是一个助手。根据对话历史和用户最新问题决定下一步行动。 可选项 - direct_answer: 如果你能直接回答用户问题无需额外信息。 - use_tool: 如果你需要调用工具如搜索、计算来获取信息。 - end_conversation: 如果用户的问题已解决或对话可以结束。 只返回选项名称。), (placeholder, {messages}), # LangGraph 会自动填充消息历史 ]) llm ChatOpenAI(modelgpt-4o-mini, temperature0) router_chain router_prompt | llm def router_node(state: AgentState) - dict: 路由节点分析状态返回下一步动作 # 准备路由决策的输入 router_input {messages: state[messages]} decision router_chain.invoke(router_input).content.strip().lower() # 根据决策更新状态中的“下一步”标志 new_state state.copy() if decision direct_answer: new_state[next_action] answer elif decision use_tool: new_state[next_action] call_tool elif decision end_conversation: new_state[should_continue] False new_state[final_output] 对话结束。 else: # 默认降级处理 new_state[next_action] answer new_state[scratchpad].append(f路由决策异常: {decision}降级为直接回答。) return new_state2. 工具调用节点Tool Node执行具体操作这里我们集成 LangChain 定义的工具。from langchain.tools import Tool from langchain_community.tools import DuckDuckGoSearchRun from langchain_community.utilities import ArxivAPIWrapper # 定义工具 search DuckDuckGoSearchRun() arxiv ArxivAPIWrapper() tools [ Tool( nameweb_search, funcsearch.run, description当需要获取最新的、实时的或未包含在知识库中的信息时使用此工具。输入是一个搜索查询词。 ), Tool( namearxiv_search, funcarxiv.run, description当需要查找学术论文、研究资料时使用此工具。输入是论文标题、作者或关键词。 ) ] def tool_node(state: AgentState) - dict: 工具调用节点让LLM选择并调用工具 from langchain.agents import create_tool_calling_agent, AgentExecutor # 1. 创建Agent它负责让LLM决定用哪个工具、输入是什么 agent_prompt ChatPromptTemplate.from_messages([...]) # 省略具体提示词 agent create_tool_calling_agent(llmllm, toolstools, promptagent_prompt) # 2. 执行Agent agent_executor AgentExecutor(agentagent, toolstools, verboseFalse) # 取最新的用户消息作为输入 latest_input state[messages][-1].content if state[messages] else state[user_input] tool_result agent_executor.invoke({input: latest_input}) # 3. 更新状态 new_state state.copy() new_state[scratchpad].append(f调用工具: {tool_result.get(intermediate_steps, [])}) # 将工具返回的结果添加到对话历史中让后续节点能看到 new_state[messages].append((assistant, f我查到的信息是{tool_result[output]})) # 工具调用后默认回到路由节点决定下一步 new_state[next_action] route return new_state3. 回答节点Answer Node生成最终回复def answer_node(state: AgentState) - dict: 回答节点基于当前所有信息生成给用户的回复 answer_prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的助手。请根据对话历史和所有可用信息生成友好、准确的回复。), (placeholder, {messages}), ]) answer_chain answer_prompt | llm response answer_chain.invoke({messages: state[messages]}) new_state state.copy() new_state[messages].append((assistant, response.content)) new_state[final_output] response.content # 回答完毕后一般也回到路由节点等待用户下一个输入 new_state[next_action] route return new_state2.3 编排成图定义 Agent 的“行为逻辑”有了节点我们需要用边把它们连接起来形成完整的工作流。from langgraph.graph import StateGraph, END # 1. 创建图 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(router, router_node) workflow.add_node(call_tool, tool_node) workflow.add_node(answer, answer_node) # 3. 设置入口点 workflow.set_entry_point(router) # 4. 定义条件边Conditional Edge根据状态决定下一步走向 def decide_next_step(state: AgentState) - str: 决定下一个节点是什么 # 如果应该结束则走向 END if not state.get(should_continue, True): return END # 否则看 next_action 字段 next_action state.get(next_action, route) if next_action call_tool: return call_tool elif next_action answer: return answer else: # 默认回到路由节点 return router # 5. 为路由节点添加条件边 workflow.add_conditional_edges( router, decide_next_step, { call_tool: call_tool, answer: answer, END: END, router: router # 自循环等待下一个用户输入 } ) # 6. 为工具节点和回答节点添加固定边执行完后固定回到路由节点 workflow.add_edge(call_tool, router) workflow.add_edge(answer, router) # 7. 编译图 agent_app workflow.compile()现在一个具备基本决策能力的 Agent 就构建好了。你可以通过agent_app.invoke()来运行它。与简单的链式调用相比这个 Agent 的优势在于流程清晰所有可能的状态转移都在图中明确定义。易于调试通过scratchpad和used_tools可以完整追溯 Agent 的思考过程。易于扩展要增加新功能如访问数据库只需添加新的工具和节点并在路由逻辑中增加选项即可。3. 实现生产级 RAG不仅仅是向量检索RAG检索增强生成是当前 AI 应用的核心模式。但很多实现停留在“切块-嵌入-检索-生成”的简单流水线忽略了生产环境中的关键问题检索质量、上下文管理和生成幻觉。我们用 LangChain 组件 LangGraph 工作流来构建一个更健壮的 RAG 系统。3.1 文档处理与索引质量始于源头检索的质量很大程度上取决于索引的质量。LangChain 提供了丰富的文档加载器和文本分割器但关键是如何组合。from langchain_community.document_loaders import PyPDFLoader, UnstructuredFileLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings def build_rag_index(doc_paths: List[str], persist_directory: str ./chroma_db): 构建 RAG 向量索引生产环境建议 all_splits [] for path in doc_paths: # 1. 根据文件类型选择加载器 if path.endswith(.pdf): loader PyPDFLoader(path) else: # Unstructured 能处理多种格式docx, pptx, html等 loader UnstructuredFileLoader(path) docs loader.load() # 2. 智能文本分割 # 关键不要只用字符长度分割要尝试保留语义段落 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 块大小 chunk_overlap200, # 重叠部分避免切断上下文 separators[\n\n, \n, 。, , , , , , ], # 中文友好分隔符 length_functionlen, ) splits text_splitter.split_documents(docs) all_splits.extend(splits) # 3. 去重可选但重要 # 简单基于内容哈希去重 unique_texts {} unique_splits [] for split in all_splits: text_hash hash(split.page_content[:500]) # 取前500字符哈希 if text_hash not in unique_texts: unique_texts[text_hash] True unique_splits.append(split) # 4. 创建向量存储 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore Chroma.from_documents( documentsunique_splits, embeddingembeddings, persist_directorypersist_directory ) vectorstore.persist() return vectorstore关键点重叠分割避免关键信息被切断。语义分割优先按段落、标题等自然边界分割。去重避免相同内容多次检索浪费上下文窗口。3.2 检索与重排提升召回精度简单返回相似度最高的 K 个片段往往不够。我们需要“检索-重排”两步走。from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor from langchain.retrievers.ensemble import EnsembleRetriever from langchain_community.retrievers import BM25Retriever def create_advanced_retriever(vectorstore): 创建高级检索器混合检索 重排 # 1. 基础向量检索器 vector_retriever vectorstore.as_retriever( search_typesimilarity, search_kwargs{k: 10} # 先召回较多候选 ) # 2. 关键词检索器BM25作为补充 # 注意需要将文档转换为纯文本列表 from langchain_community.retrievers.bm25 import BM25Retriever as LangchainBM25 # 这里简化处理实际需从vectorstore中提取文本 # bm25_retriever LangchainBM25.from_documents(docs) # 3. 混合检索器Ensemble # ensemble_retriever EnsembleRetriever( # retrievers[vector_retriever, bm25_retriever], # weights[0.7, 0.3] # 向量检索权重更高 # ) # 4. 上下文压缩重排器用LLM对召回结果进行精炼只保留相关部分 llm ChatOpenAI(temperature0, modelgpt-4o-mini) compressor LLMChainExtractor.from_llm(llm) compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrievervector_retriever # 可以用 ensemble_retriever ) return compression_retriever这种设计的好处是先用向量/关键词检索广泛召回再用一个轻量级 LLM 对每个片段进行“相关性打分”或“内容提取”只把最相关的部分送入最终的生成环节极大节省了上下文窗口。3.3 用 LangGraph 编排 RAG 工作流简单的 RAG 是一条链但复杂的 RAG 可能涉及多轮追问、结果验证、来源追溯等。这时就需要图。from typing import TypedDict import operator class RAGState(TypedDict): question: str retrieved_docs: List # 检索到的文档 condensed_context: str # 压缩/重排后的上下文 answer: str need_follow_up: bool # 是否需要追问澄清 follow_up_question: str # 追问的问题 def retrieve_node(state: RAGState): 检索节点 retriever create_advanced_retriever(vectorstore) docs retriever.invoke(state[question]) return {retrieved_docs: docs} def condense_node(state: RAGState): 上下文压缩节点如果检索结果太多进行总结提炼 if len(state[retrieved_docs]) 5: # 调用LLM总结这些文档的核心信息 condense_prompt ChatPromptTemplate.from_template( 请根据以下文档片段提炼出与问题“{question}”最相关的核心信息合成一段连贯的文本。 文档片段 {docs} 核心信息提炼) chain condense_prompt | llm condensed chain.invoke({question: state[question], docs: state[retrieved_docs]}) return {condensed_context: condensed.content} else: return {condensed_context: \n\n.join([doc.page_content for doc in state[retrieved_docs]])} def generate_node(state: RAGState): 生成答案节点 qa_prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的助手请严格根据提供的上下文信息回答问题。如果上下文没有明确答案请说“根据已有信息无法回答”不要编造信息。), (human, 上下文{context}\n\n问题{question}) ]) qa_chain qa_prompt | llm answer qa_chain.invoke({context: state[condensed_context], question: state[question]}) return {answer: answer.content} def validate_node(state: RAGState): 答案验证节点检查生成答案是否基于上下文是否需要追问 validation_prompt ChatPromptTemplate.from_template( 请判断以下答案是否严格基于提供的上下文。同时判断用户的问题是否模糊、需要进一步澄清。 上下文{context} 问题{question} 答案{answer} 请用JSON格式回答{{is_grounded: true/false, need_clarification: true/false, clarification_question: 如果需要澄清请提出一个问题}} ) chain validation_prompt | llm judgment chain.invoke(state) # 解析JSON结果更新状态 # ... 解析逻辑 # 如果 need_clarification 为 True则更新 follow_up_question return {need_follow_up: True, follow_up_question: 您能具体说明一下XX方面吗} # 构建 RAG 图 rag_workflow StateGraph(RAGState) rag_workflow.add_node(retrieve, retrieve_node) rag_workflow.add_node(condense, condense_node) rag_workflow.add_node(generate, generate_node) rag_workflow.add_node(validate, validate_node) rag_workflow.set_entry_point(retrieve) rag_workflow.add_edge(retrieve, condense) rag_workflow.add_edge(condense, generate) rag_workflow.add_edge(generate, validate) # 条件边验证后如果不需要追问结束如果需要则进入一个“等待用户澄清”的节点或循环。 def after_validate(state: RAGState): if state.get(need_follow_up): return ask_for_clarification # 假设有这样一个节点 else: return END rag_workflow.add_conditional_edges(validate, after_validate) rag_app rag_workflow.compile()这个 RAG 工作流实现了检索增强使用高级检索器。上下文管理在文档过多时进行压缩。答案验证检查幻觉并判断是否需要追问用户澄清。流程可观测每个节点的输入输出都清晰可见。这才是面向生产的 RAG它不仅仅是检索和生成而是一个受控的、可观测的、具备反馈循环的问答系统。4. 集成 MCPModel Context Protocol连接外部世界的“标准插座”MCP 是一个新兴但非常重要的协议它旨在标准化 AI 应用与外部工具、数据源之间的通信方式。你可以把它理解为 AI 世界的“USB-C 接口”。之前每个工具都需要为每个 AI 框架LangChain, LlamaIndex, AutoGen写一套适配器。有了 MCP工具只需实现一次 MCP 服务端Server任何支持 MCP 的客户端如 LangGraph都可以即插即用。4.1 MCP 核心概念MCP Server一个提供工具Tools和资源Resources的进程。例如一个数据库 MCP Server 可以提供“执行 SQL 查询”的工具和“表结构”资源。MCP ClientLangGraph 可以作为 MCP Client动态发现并调用 Server 提供的工具。Stdio/SSE 传输Client 和 Server 通过标准输入输出或 Server-Sent Events 通信无需 HTTP API。4.2 在 LangGraph 中使用 MCP 工具假设我们有一个本地的“文件系统 MCP Server”它提供了list_directory和read_file两个工具。# 假设我们使用 langchain-mcp 适配库请关注官方或社区更新 # 以下代码展示概念性流程 import asyncio from langchain_mcp import MCPServer # 1. 启动或连接一个 MCP Server # 通常通过子进程启动一个 Server 二进制文件 async def setup_mcp_tools(): # 创建 MCP 客户端连接到 Server # 这里概念性展示实际 API 可能不同 # async with MCPServer(path/to/your/mcp-server-binary) as client: # # 2. 发现 Server 提供的所有工具 # tools await client.list_tools() # # 3. 将这些工具转换成 LangChain Tool 对象 # langchain_tools [] # for tool_info in tools: # langchain_tools.append( # Tool( # nametool_info.name, # funclambda params, tool_infotool_info: asyncio.run(client.call_tool(tool_info.name, params)), # descriptiontool_info.description # ) # ) # return langchain_tools # 示例返回 return [ Tool( namelist_directory, funclambda path: fListing dir: {path}, # 模拟函数 description列出指定目录下的文件和文件夹 ), Tool( nameread_file, funclambda filepath: fContents of {filepath}, # 模拟函数 description读取指定文件的内容 ) ] # 在 LangGraph 的 Agent 中我们可以动态加载这些工具 class DynamicAgentState(TypedDict): messages: List available_tools: List[Tool] # 状态中保存可用工具列表 async def dynamic_agent_workflow(): # 初始化时或运行时加载 MCP 工具 mcp_tools await setup_mcp_tools() # 创建图并将工具传递给相关节点 workflow StateGraph(DynamicAgentState) # 定义一个节点它可以使用 state 中的 available_tools def agent_node(state: DynamicAgentState): # 这里的 agent 可以使用 state[available_tools] 来调用工具 # ... agent 逻辑 pass # ... 构建图MCP 的优势解耦工具开发与 AI 应用开发分离。动态性Agent 可以在运行时发现新工具无需重启。标准化统一的工具描述、调用和结果返回格式。安全性可以通过 Server 实现权限控制和审计。对于企业环境MCP 意味着你可以让运维团队维护一个“数据库 MCP Server”让业务团队维护一个“CRM MCP Server”AI 应用开发者无需关心底层连接细节只需像调用本地函数一样使用这些工具。这是构建复杂、安全的企业级 Agent 的关键基础设施。5. 从开发到部署工程化实践与避坑指南构建一个能跑的 Demo 和构建一个能服务成百上千用户的生产系统是两回事。以下是基于经验的关键实践。5.1 状态持久化与长期记忆LangGraph 的State默认在内存中。生产环境需要持久化。方案一使用Checkpointer。LangGraph 内置了检查点机制可以将状态保存到数据库如 PostgreSQL, Redis。方案二自定义持久化节点。在图的关键节点如一轮对话结束将状态序列化后存入你的业务数据库。from langgraph.checkpoint import MemorySaver # 使用内存检查点适合开发 memory MemorySaver() app workflow.compile(checkpointermemory) # 使用时会传入一个 thread_id用于恢复状态 config {configurable: {thread_id: user_123_session_456}} initial_state {messages: [(user, 你好)]} result app.invoke(initial_state, configconfig) # 下次可以用相同的 thread_id 恢复对话5.2 超时、重试与容错网络调用和模型调用可能失败。为 LLM 调用和工具调用设置超时使用asyncio.timeout或tenacity库进行重试。图的错误处理LangGraph 允许你为节点设置try...except将错误信息捕获并放入状态引导流程走向一个“错误处理节点”而不是整个应用崩溃。def safe_tool_node(state: State): try: # 可能失败的工具调用 result some_unreliable_api(state[input]) return {output: result} except TimeoutError: return {error: 工具调用超时, next_action: handle_error} except Exception as e: return {error: str(e), next_action: handle_error}5.3 监控、日志与可观测性这是生产系统的生命线。结构化日志在每个节点的开始和结束记录日志包含节点名、状态摘要、耗时、错误信息。使用logging库并输出 JSON 格式便于接入 ELK 或 Datadog。追踪Tracing利用 LangSmithLangChain 官方平台或 OpenTelemetry 对每次调用进行追踪可视化整个图的执行路径查看每个节点的输入输出。这对于调试复杂工作流至关重要。关键指标记录每个请求的 token 消耗、耗时、工具调用次数、缓存命中率等。5.4 性能优化缓存对 LLM 的相同提示词、对检索器的相同查询进行缓存。LangChain 内置了InMemoryCache、RedisCache等。异步执行如果图中有可以并行执行的节点如同时调用多个不依赖的工具使用langgraph.graph的StateGraph的add_edge和并发控制来实现。精简上下文这是成本控制和效果提升的核心。严格过滤检索结果使用LLMChainExtractor等压缩技术只把必要信息送给 LLM。5.5 安全与权限工具权限不是所有 Agent 都能调用所有工具。根据用户身份或会话上下文动态过滤可用的工具列表state[available_tools]。输入输出过滤对用户输入和模型输出进行内容安全过滤防止提示词注入或生成有害内容。数据隔离在多租户环境中确保通过thread_id或其他标识严格隔离不同用户的数据和状态。6. 总结LangChain LangGraph 的思维转变回顾整篇文章我们从 LangChain 与 LangGraph 的本质区别开始到构建具备决策能力的 Agent实现生产级 RAG集成标准化工具协议 MCP最后探讨工程化实践。这条路径的核心其实是一场思维转变从“如何调用模型”转变为“如何设计一个由模型驱动的、确定性的工作流”。LangChain 给了你标准化的积木而 LangGraph 给了你设计图纸和组装流水线。在 2026 年AI 应用的竞争将不再是“谁能调通 API”而是“谁能设计出更稳健、更高效、更易维护的智能工作流”。所以当你下次再启动一个 AI 项目时不要一上来就写llm.invoke()。先拿出一张白纸或者打开绘图工具问自己几个问题我的应用有哪些状态用户输入、对话历史、中间结果、工具输出……有哪些节点理解意图、检索、推理、执行工具、生成回复、验证……节点之间如何流转什么条件下走哪条路错误可能发生在哪里如何恢复如何让这个流程被观察和调试把这些问题想清楚画成图再用 LangGraph 把它实现出来。这才是通往下一代 AI 应用的正确路径。工具和技术迭代很快但这种基于状态和流程的系统设计能力才是你真正需要积累的核心资产。