这类工具最值得先看的不是功能列表而是能不能在普通环境里稳定跑起来以及从单任务到批量任务、再到生产化部署的完整路径是否清晰。LangChain 和 LangGraph 作为当前构建 AI 应用和智能体的核心框架热度很高但很多教程要么停留在概念要么代码跑不通要么没讲清楚从 Demo 到真实项目的关键步骤。我建议把学习过程拆成三步先搞懂 LangChain 和 LangGraph 各自解决什么问题、怎么配合再动手搭一个能跑通的 RAG 或 Agent 最小原型最后才是考虑如何用 MCP 扩展工具、管理长期记忆、处理复杂工作流。下面我会按这个顺序结合常见环境本地 Python、Ollama 或在线 API和实际踩坑经验把从零到一的实战流程拆解清楚。1. 先理清 LangChain 和 LangGraph 的分工别混为一谈很多人一上来就被各种概念搞晕把 LangChain 和 LangGraph 当成一个东西或者觉得 LangGraph 是 LangChain 的升级版。其实它们定位不同组合使用才能发挥最大价值。1.1 LangChain你的“应用组装车间”你可以把 LangChain 理解为一个提供了大量标准化“零件”和“组装流水线”的车间。它的核心价值是标准化和简化 AI 应用的构建过程。标准化零件Components比如各种大模型LLM的调用接口OpenAI、Ollama、通义千问等、文本分割器Text Splitter、向量数据库连接器Chroma、Pinecone、文档加载器PDF、TXT、网页爬取等。LangChain 把这些常用功能封装成了统一的类和方法你不用再为每个模型或数据库写一堆适配代码。组装流水线Chains这是 LangChain 的精华。它提供了一套声明式的、可组合的方式来把多个“零件”串联成一个完整的处理流程。最经典的例子就是 RAG检索增强生成链文档加载 - 文本分割 - 向量化存储 - 用户提问 - 检索相关片段 - 组合提示词 - 调用 LLM 生成答案。用 LangChain你几行代码就能把这个流程搭起来。关键判断如果你的需求是“快速搭建一个基于文档问答的聊天机器人”、“做一个简单的文本总结工具”或者“把不同 AI 服务串起来”那么直接从 LangChain 入手是最快路径。它的学习曲线前半段比较平缓。1.2 LangGraph你的“智能体调度中心”LangGraph 解决的是另一个问题如何让 AI 智能体Agent具备复杂的、有状态的、可循环的工作流。如果说 LangChain 是组装静态流水线那 LangGraph 就是设计动态的、带“决策回路”的机器人。核心是“图”Graph和“状态”State你把智能体的每个步骤如“思考”、“调用工具”、“等待用户输入”定义成图中的一个节点Node用边Edge来连接它们决定流程走向。整个图共享一个状态对象记录当前对话历史、工具调用结果、中间结论等。支持循环和条件分支这是普通 Chain 难以做到的。比如一个分析报告的智能体可以1. 理解任务2. 决定需要搜索网络3. 调用搜索工具4. 判断信息是否足够如果不够回到步骤25. 信息足够后开始撰写报告。这种带“循环判断”的流程用 LangGraph 来建模非常自然。长期记忆Long-term Memory的天然载体因为状态可以被持久化到数据库智能体的“记忆”可以跨越单次会话。这对于构建像“个人工作助理”这类需要记住用户偏好和历史任务的智能体至关重要。关键判断当你的需求超出简单的“一问一答”或“固定流程处理”涉及到“根据中间结果决定下一步做什么”、“需要多次调用工具并汇总信息”、“要维护跨会话的记忆”时就必须引入 LangGraph或类似的智能体框架。简单总结LangChain用来快速搭建标准化任务流水线如 RAGLangGraph用来设计和运行具备决策能力的智能体工作流。在复杂项目中你通常会先用 LangChain 的组件如LLM调用、文档处理再用 LangGraph 把它们编排成智能体。2. 环境准备与最小原型跑通第一个 RAG 应用理论清楚了接下来就要动手。我强烈建议从 RAG 开始因为它涵盖了 LangChain 最核心的几大模块且效果立竿见影。这里以使用本地模型通过 Ollama为例这是成本最低、最可控的方式。2.1 基础环境搭建确保你的机器已经准备好Python 环境推荐 Python 3.9。使用conda或venv创建独立的虚拟环境是好习惯。conda create -n langchain-demo python3.10 conda activate langchain-demo安装核心库pip install langchain langchain-community langchain-corelangchain是主包langchain-community包含很多第三方集成如Ollamalangchain-core是基础运行时。安装向量数据库本地开发首选Chroma轻量且无需额外服务。pip install chromadb安装 Ollama 并拉取模型到 Ollama 官网下载安装。然后拉取一个适合你电脑配置的模型比如qwen2.5:7b7B参数对CPU/内存要求相对友好或llama3.2:3b更小。ollama pull qwen2.5:7b运行ollama run qwen2.5:7b测试模型是否能正常对话。2.2 四步构建你的第一个 RAG 链现在在一个 Python 文件如demo_rag.py中跟着以下步骤操作步骤一初始化 LLM 和 Embedding 模型from langchain_community.llms import Ollama from langchain_community.embeddings import OllamaEmbeddings # 连接到本地 Ollama 服务使用我们拉取的模型 llm Ollama(modelqwen2.5:7b, base_urlhttp://localhost:11434) # Embedding 模型用于将文本转换为向量通常使用专门的嵌入模型这里用同一个模型也可以 embeddings OllamaEmbeddings(modelqwen2.5:7b, base_urlhttp://localhost:11434)注意base_url是 Ollama 默认的本地服务地址。如果 Ollama 运行在其他地方需要修改。步骤二加载并处理你的文档假设你有一个knowledge.txt文件里面是一些技术文档。from langchain_community.document_loaders import TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter # 1. 加载文档 loader TextLoader(./knowledge.txt) documents loader.load() # 2. 分割文档。这是 RAG 的关键分割大小影响检索质量。 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个片段约500字符 chunk_overlap50 # 片段间重叠50字符保持上下文连贯 ) docs text_splitter.split_documents(documents) print(f原始文档被分割成了 {len(docs)} 个片段)步骤三创建向量数据库向量存储from langchain_community.vectorstores import Chroma # 将分割后的文档片段转换为向量并存入 Chroma # persist_directory 指定持久化目录这样下次启动无需重新生成 vectorstore Chroma.from_documents( documentsdocs, embeddingembeddings, persist_directory./chroma_db ) # 创建一个检索器Retriever用于后续查询 retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 检索最相关的3个片段步骤四组装 RAG 链并进行问答from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough # 1. 定义提示词模板 template 请根据以下上下文来回答问题。如果你不知道答案就说不知道不要编造。 上下文{context} 问题{question} 请用中文给出有帮助的答案 prompt ChatPromptTemplate.from_template(template) # 2. 组装链输入问题 - 检索上下文 - 格式化提示词 - 调用LLM - 解析输出 rag_chain ( {context: retriever, question: RunnablePassthrough()} | prompt | llm | StrOutputParser() ) # 3. 提问 question LangChain 是做什么用的 answer rag_chain.invoke(question) print(f问题{question}) print(f答案{answer})运行这个脚本。如果一切顺利你会看到模型基于你的knowledge.txt文件内容生成的答案。恭喜你的第一个 RAG 应用跑通了实测要点首次运行较慢因为要计算所有文档片段的向量并存入数据库。后续提问很快因为只需要检索和生成。检查chroma_db目录里面保存了向量数据库文件下次启动可以这样加载避免重复计算vectorstore Chroma(persist_directory./chroma_db, embedding_functionembeddings) retriever vectorstore.as_retriever()3. 从链到图用 LangGraph 构建你的第一个智能体当你的问题变得复杂比如“帮我分析一下最近三天的项目日志总结出错误趋势并写一份报告草稿”这就需要智能体了。我们用 LangGraph 构建一个最简单的“思考-行动-观察”循环智能体。3.1 理解智能体的基本循环一个典型的 ReActReasoning Acting智能体工作流如下思考根据目标和当前信息决定下一步该做什么是调用工具还是直接给出最终答案。行动如果决定调用工具就执行对应的工具如搜索、计算、查询数据库。观察获取工具执行的结果并将其作为新信息加入到上下文中。循环回到“思考”步骤直到智能体认为可以给出最终答案。3.2 用 LangGraph 实现一个计算器智能体我们实现一个能进行多步数学运算的智能体。首先安装 LangGraphpip install langgraph然后创建demo_agent.py步骤一定义智能体的状态状态是一个字典记录整个工作流的所有信息。from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END # 定义状态结构 class AgentState(TypedDict): question: str # 用户原始问题 scratchpad: List[str] # 思考过程记录 current_step: str # 当前步骤”think“, “act”, “finish” tool_input: str # 给工具的输入 tool_output: str # 工具的输出 final_answer: str # 最终答案步骤二定义工具这里我们定义一个简单的计算器工具。from langchain.tools import tool tool def calculator(expression: str) - str: 计算一个数学表达式的值。例如3 5 * 2 try: # 警告实际生产中不要用 eval这里仅为演示 result eval(expression) return str(result) except Exception as e: return f计算错误{e}步骤三定义图节点Nodes我们需要三个节点think思考act行动finish结束。from langchain_community.llms import Ollama llm Ollama(modelqwen2.5:7b) def think_node(state: AgentState) - dict: 思考节点决定下一步是调用工具还是结束 # 构建提示词让 LLM 做决策 prompt f 你是一个数学助手。当前问题{state[question]} 已有的思考记录{state.get(scratchpad, [])} 你需要决定下一步 - 如果问题已经解决或者不需要计算就回复 FINISH。 - 如果还需要计算就回复 CALCULATE: 表达式例如 CALCULATE: 3 5。 只回复 FINISH 或 CALCULATE: 表达式。 response llm.invoke(prompt).strip() scratchpad state.get(scratchpad, []) scratchpad.append(f思考{response}) if response.startswith(FINISH): return {current_step: finish, scratchpad: scratchpad} elif response.startswith(CALCULATE:): expression response.split(CALCULATE:)[1].strip() return {current_step: act, tool_input: expression, scratchpad: scratchpad} else: # 如果 LLM 回复不符合格式默认结束 return {current_step: finish, scratchpad: scratchpad} def act_node(state: AgentState) - dict: 行动节点执行工具 expression state[tool_input] result calculator.invoke({expression: expression}) scratchpad state.get(scratchpad, []) scratchpad.append(f行动计算 {expression} {result}) return {tool_output: result, scratchpad: scratchpad, current_step: think} def finish_node(state: AgentState) - dict: 结束节点生成最终答案 prompt f 基于以下信息给出最终答案。 问题{state[question]} 思考与计算过程{state.get(scratchpad, [])} 请用清晰的一句话总结答案。 final_answer llm.invoke(prompt) return {final_answer: final_answer, current_step: END}步骤四构建图并设置路由# 创建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(think, think_node) workflow.add_node(act, act_node) workflow.add_node(finish, finish_node) # 设置起始节点 workflow.set_entry_point(think) # 设置边路由逻辑 workflow.add_conditional_edges( think, # 根据 think_node 返回的 state[current_step] 决定下一个节点 lambda state: state[current_step], { act: act, finish: finish } ) workflow.add_edge(act, think) # 执行完行动后继续思考 workflow.add_edge(finish, END) # 结束节点指向图终点 # 编译图 app workflow.compile()步骤五运行智能体# 初始化状态 initial_state AgentState( question3 加上 5 乘以 2 等于多少, scratchpad[], current_stepthink, tool_input, tool_output, final_answer ) # 运行图 final_state app.invoke(initial_state) print(最终答案, final_state.get(final_answer)) print(\n完整思考过程) for step in final_state.get(scratchpad, []): print(f- {step})运行这个脚本你会看到智能体经历了“思考 - 决定计算‘5*2’ - 行动 - 再思考 - 决定计算‘310’ - 行动 - 再思考 - 结束”的完整过程并给出答案。关键点状态驱动整个流程由state字典的演变来控制。可循环think和act之间形成了循环。可扩展你可以轻松添加更多工具如搜索、查数据库和决策分支。4. 进阶整合RAG Agent MCP 构建强大智能体现在我们把前面学的拼起来并引入 MCPModel Context Protocol构建一个更实用的智能体它能查询本地知识库RAG也能调用外部工具通过MCP并自主决策工作流。4.1 什么是 MCP为什么需要它MCP 是一个协议它允许 AI 应用如你的智能体以标准化的方式发现、描述和调用外部工具、数据源或服务。你可以把它想象成智能体的“插件系统”。对开发者的价值你不用为每个外部 API 或数据库都写一遍复杂的集成代码。只要对方提供了 MCP 服务器MCPServer你的智能体就能通过标准协议调用它。对智能体的价值智能体可以动态地“知道”自己有哪些工具可用通过读取 MCP 服务器的工具清单并在需要时调用。热门集成像 Figma、蓝湖设计协作平台、数据库等都有社区或官方在探索 MCP 集成让智能体可以直接操作设计稿或查询业务数据。4.2 搭建一个具备 RAG 和外部工具能力的智能体这个例子将结合RAG基于我们之前建的chroma_db回答公司内部知识。MCP 工具假设我们有一个提供“天气查询”和“股票价格”的 MCP 服务器。LangGraph 智能体协调决策决定何时用 RAG何时调用外部工具。步骤一准备 MCP 工具模拟由于搭建真实 MCP 服务器较复杂我们模拟两个工具函数# mcp_tools.py def get_weather(city: str) - str: 模拟获取天气的 MCP 工具 # 这里应该是调用真实 MCP 服务器我们模拟返回 weather_data { 北京: 晴15-25°C, 上海: 多云18-28°C, 深圳: 阵雨22-30°C } return weather_data.get(city, f未找到{city}的天气信息) def get_stock_price(symbol: str) - str: 模拟获取股票价格的 MCP 工具 # 模拟返回 price_data { AAPL: 172.31 USD, MSFT: 415.86 USD, GOOGL: 155.49 USD } return price_data.get(symbol.upper(), f未找到股票代码{symbol}的实时价格)步骤二定义更复杂的智能体状态和工具集# advanced_agent.py from typing import TypedDict, List, Literal from langgraph.graph import StateGraph, END from langchain_community.llms import Ollama from langchain_community.vectorstores import Chroma from langchain_community.embeddings import OllamaEmbeddings from mcp_tools import get_weather, get_stock_price # 导入模拟工具 llm Ollama(modelqwen2.5:7b) embeddings OllamaEmbeddings(modelqwen2.5:7b) # 加载之前创建的 RAG 向量库 vectorstore Chroma(persist_directory./chroma_db, embedding_functionembeddings) rag_retriever vectorstore.as_retriever() class AdvancedAgentState(TypedDict): messages: List[dict] # 对话历史格式 [{role: user/assistant, content: ...}, ...] next_step: Literal[rag_query, call_tool, final_answer] # 下一步决策 tool_name: str # 要调用的工具名 tool_input: str # 工具输入 tool_output: str # 工具输出 rag_context: str # RAG 检索到的上下文 final_response: str # 最终回复 # 可用工具清单 TOOLS { get_weather: { function: get_weather, description: 根据城市名查询天气输入格式北京 }, get_stock_price: { function: get_stock_price, description: 根据股票代码查询实时价格输入格式AAPL }, query_knowledge_base: { function: rag_retriever, # 这里实际是检索器调用方式特殊 description: 查询公司内部知识库输入是一个问题 } }步骤三实现决策节点路由这是智能体的大脑决定下一步做什么。def router_node(state: AdvancedAgentState) - dict: 路由节点分析最新用户消息决定下一步 last_message state[messages][-1][content] # 简单的基于关键词的路由逻辑。实际应用中这里应该用 LLM 做更智能的决策。 if any(word in last_message.lower() for word in [天气, weather]): return {next_step: call_tool, tool_name: get_weather} elif any(word in last_message.lower() for word in [股票, 股价, stock]): return {next_step: call_tool, tool_name: get_stock_price} elif any(word in last_message.lower() for word in [公司, 制度, 文档, how to, what is]): return {next_step: rag_query} else: # 默认尝试用知识库回答 return {next_step: rag_query}步骤四实现 RAG 查询节点和工具调用节点def rag_query_node(state: AdvancedAgentState) - dict: 执行 RAG 查询 last_message state[messages][-1][content] # 检索相关文档 docs rag_retriever.invoke(last_message) context \n.join([doc.page_content for doc in docs]) # 组合提示词让 LLM 基于上下文回答 prompt f基于以下公司知识库上下文回答用户问题。如果上下文不相关或没有答案请如实告知。 上下文 {context} 用户问题{last_message} 助手回答 response llm.invoke(prompt) new_messages state[messages] [{role: assistant, content: response}] return {rag_context: context, messages: new_messages, next_step: final_answer} def call_tool_node(state: AdvancedAgentState) - dict: 调用外部工具 tool_name state[tool_name] last_message state[messages][-1][content] tool_info TOOLS[tool_name] tool_func tool_info[function] # 简陋的参数提取实际应用应用 LLM 或更复杂的解析器 if tool_name get_weather: # 假设用户消息是“北京天气怎么样” tool_input 北京 # 这里应做实体识别简化处理 elif tool_name get_stock_price: tool_input AAPL # 简化处理 else: tool_input last_message # 调用工具 tool_output tool_func(tool_input) if tool_name ! query_knowledge_base else 工具调用方式特殊已在rag_query节点处理 # 将工具结果加入对话历史让 LLM 消化 tool_message f[工具 {tool_name} 调用结果] {tool_output} new_messages state[messages] [{role: user, content: tool_message}] return {tool_output: tool_output, messages: new_messages, next_step: final_answer}步骤五实现最终回答节点并组装图def final_answer_node(state: AdvancedAgentState) - dict: 生成最终回答如果需要的话。这里我们简单地将最后一条消息作为回答。 # 在我们的简单流程中rag_query_node 和 call_tool_node 已经通过 LLM 生成了回答。 # 所以最终回答就是对话历史中最后一条助手消息。 last_msg state[messages][-1] if last_msg[role] assistant: final_response last_msg[content] else: # 如果最后一条不是助手消息比如只有工具输出则让 LLM 总结一下 prompt f对话历史{state[messages]}\n请基于以上信息生成对用户的最终回复。 final_response llm.invoke(prompt) return {final_response: final_response, next_step: END} # 构建图 workflow StateGraph(AdvancedAgentState) workflow.add_node(router, router_node) workflow.add_node(rag_query, rag_query_node) workflow.add_node(call_tool, call_tool_node) workflow.add_node(final_answer, final_answer_node) workflow.set_entry_point(router) # 设置条件路由 workflow.add_conditional_edges( router, lambda state: state[next_step], { rag_query: rag_query, call_tool: call_tool, final_answer: final_answer } ) workflow.add_edge(rag_query, final_answer) workflow.add_edge(call_tool, final_answer) workflow.add_edge(final_answer, END) app workflow.compile()步骤六测试智能体def run_agent_conv(question: str): initial_state AdvancedAgentState( messages[{role: user, content: question}], next_step, tool_name, tool_input, tool_output, rag_context, final_response ) result app.invoke(initial_state) print(f用户{question}) print(f智能体{result[final_response]}) print(- * 50) # 测试 run_agent_conv(北京今天天气怎么样) run_agent_conv(AAPL的股价是多少) run_agent_conv(公司的请假制度是什么) # 这个问题会触发 RAG从你的 knowledge.txt 中找答案这个例子虽然简化了比如工具参数提取很粗糙但它清晰地展示了如何将 RAG、外部工具MCP 理念和 LangGraph 智能体工作流结合起来。在实际项目中你可以用更强大的 LLM 来做路由决策和参数提取并集成真实的 MCP 服务器。5. 生产化考量与常见避坑指南把 Demo 跑起来只是第一步。要真正用于生产或复杂项目以下几个点必须提前规划。5.1 性能与资源优化向量数据库选择Chroma适合开发和轻量生产。数据量大、要求高并发时考虑Weaviate、Qdrant或Pinecone云服务。Embedding 模型Ollama 上的通用 LLM 做 Embedding 通常效果和速度不是最优。考虑专门的开源 Embedding 模型如bge-small-zh-v1.5或商用 API如 OpenAItext-embedding-3-small。LLM 调用超时与重试网络不稳定或模型负载高时必须设置超时和重试逻辑。流式输出对于长文本生成使用流式Streaming以提升用户体验。缓存对频繁相同的查询引入 LLM 调用缓存如LangChain的CacheBacked能极大节省成本和时间。图执行优化LangGraph 支持检查点Checkpointing可以将智能体的状态持久化用于暂停、恢复或异步执行长任务。5.2 稳定性与可观测性错误处理在图中的每个节点都要用try...except包裹处理好工具调用失败、LLM 返回格式错误、网络异常等情况并能够将错误信息反馈到状态中让智能体有机会尝试其他路径。日志记录详细记录每个节点的输入输出、工具调用详情、LLM 的请求和响应。这对于调试复杂的工作流至关重要。考虑结构化日志如 JSON 格式。限制与防护循环限制为智能体的“思考-行动”循环设置最大迭代次数防止陷入死循环。令牌限制控制每次调用 LLM 的上下文长度避免成本过高或超载。工具权限对于 MCP 工具要有明确的权限控制防止智能体执行危险操作。5.3 开发与部署流程版本控制将你的智能体图workflow定义、提示词模板、工具配置都纳入 Git 管理。配置化将模型名称、API Key、数据库连接字符串等抽离为配置文件或环境变量。测试单元测试测试每个工具函数、每个节点函数。集成测试测试整个图对典型输入的处理。回归测试当更新提示词或 LLM 模型后用一组标准问题验证效果是否下降。部署Web 服务使用FastAPI将你的智能体图包装成 REST API。LangGraph StudioLangChain 提供的可视化开发和调试工具可以直观地看到图的执行流程和状态变化非常适合调试。容器化使用 Docker 打包你的应用和环境确保一致性。5.4 最常见的“坑”与排查顺序Ollama 服务未启动或模型未下载症状是连接超时或模型找不到。首先运行ollama list确认模型存在然后运行ollama serve确保服务在运行。向量数据库路径或权限问题症状是检索不到内容或报错。检查persist_directory路径是否存在、是否有写入权限。首次运行后确认chroma_db目录下生成了文件。提示词Prompt效果差症状是 LLM 回答不相关或格式错误。这是最常见的问题。不要只改一两个词要系统性地设计提示词结构指令、上下文、示例、格式要求并进行多轮测试。使用LangChain的PromptTemplate和ChatPromptTemplate来管理。文本分割Chunk不合理症状是 RAG 检索的片段不完整或丢失关键信息。调整chunk_size和chunk_overlap参数。对于中文可能需要按句号或语义分割而不是简单的字符数。LangGraph 图陷入死循环症状是程序长时间不结束。检查你的条件路由逻辑确保存在通往END节点的路径。务必设置interrupt_before或interrupt_after来添加手动断点或者设置最大循环次数。工具调用参数错误症状是工具执行失败或返回意外结果。确保传递给工具的输入格式与工具函数定义的参数类型完全匹配。使用 Pydantic 模型来严格定义工具输入模式是很好的实践。依赖库版本冲突langchain和langgraph生态更新很快。建议使用requirements.txt或pyproject.toml精确锁定版本特别是生产环境。我个人更建议在把一个基于 LangChain 和 LangGraph 的智能体投入实际使用前先用一个简单的“测试套件”跑一遍从环境检查、单条问答、到批量压力测试、再到模拟异常输入把整个流程走稳。这比盲目追求功能的复杂程度要重要得多。