在构建AI应用时你是否遇到过这样的困境单个大模型能力有限无法处理复杂任务RAG系统虽然能检索知识但逻辑推理和流程控制能力薄弱想要实现多步骤协作的智能体却不知从何下手网上资料零散且不成体系。如果你正为此烦恼那么恭喜你这篇文章正是为你准备的。本文将带你系统性地掌握LangChain与LangGraph两大核心框架从零构建一个集成了多智能体协作、RAG知识库和MCP工具扩展的实战项目。无论你是希望快速入门的新手还是寻求项目落地的进阶开发者都能在这里找到从环境搭建、核心原理到避坑指南的完整闭环方案。我们将用三天时间彻底吃透Agent、RAG与MCP让你在AI应用开发的道路上少走99%的弯路。1. 背景与核心概念为什么需要LangChain与LangGraph在深入代码之前我们必须先理解我们正在解决的问题以及所使用的工具。AI应用开发早已不是简单地调用一个API而是涉及提示工程、工具调用、记忆管理、流程编排等多个复杂环节的系统工程。1.1 LangChainAI应用开发的“脚手架”LangChain是一个用于开发由语言模型驱动的应用程序的框架。你可以把它想象成乐高积木的基础连接件。它的核心价值在于标准化和模块化。标准化接口它定义了LLM、ChatModel、Tool、Memory、Chain等通用接口。无论底层是OpenAI的GPT、Anthropic的Claude还是开源的Llama、Qwen你都可以用同一套代码来调用极大地降低了切换模型带来的成本。模块化组件它将复杂的AI应用拆解成可复用的组件。例如PromptTemplate负责管理提示词VectorStore负责向量检索Agent负责决策和工具调用。你可以像搭积木一样组合这些组件。核心功能主要包括模型I/O与LLM对话、数据连接文档加载、向量化、链顺序调用、记忆保存对话历史、代理根据目标调用工具等。简单来说LangChain解决了“如何方便地使用大模型”的问题。1.2 LangGraph复杂工作流的“编排引擎”如果说LangChain提供了积木块那么LangGraph就是搭建复杂动态结构的设计图和控制器。它基于状态图StateGraph的概念专门用于构建有状态、多步骤、可能循环或分支的智能体Agent工作流。解决什么问题传统的LangChainAgent或Chain通常是线性的或简单的if-else逻辑。对于需要根据中间结果动态决定下一步、或者多个角色智能体之间需要协作的任务如“写代码-评审代码-执行测试”传统方式会变得非常臃肿和难以维护。核心概念状态State一个共享的字典存储工作流运行过程中的所有信息如用户问题、模型回复、工具执行结果等。节点Node一个执行单元可以是一个函数它读取和更新状态。例如一个“调用LLM”的节点一个“执行搜索”的节点。边Edge决定工作流下一个应该执行哪个节点的规则。可以是固定的always_go_to也可以根据状态内容动态决定conditional_edge。与LangChain Agent的关系你可以把LangGraph看作是构建更强大、更可控Agent的底层框架。一个经典的ReAct Agent可以用LangGraph清晰地表达为“思考-行动-观察”的循环。1.3 RAG、Agent与MCP构建智能应用的“三驾马车”理解了框架我们再来看看构建应用的核心模式。RAG检索增强生成解决大模型“知识陈旧”和“幻觉”问题。通过将外部知识库如文档、数据库向量化在回答问题时先进行相关检索再将检索到的片段作为上下文提供给模型从而生成更准确、更可靠的答案。Agent智能体一个能感知环境、进行决策并执行行动以实现目标的系统。在LLM语境下Agent通常指一个能调用工具如计算器、搜索引擎、API的LLM。多智能体则是由多个具有不同角色和能力的Agent协作完成任务。MCP模型上下文协议这是一个由Anthropic提出的新兴协议旨在标准化LLM与外部工具/数据源之间的连接方式。你可以把它理解为LLM界的“USB协议”。MCP Server提供工具和数据MCP Client如LangChain可以动态发现并使用这些工具无需硬编码。这极大地增强了AI应用的可扩展性和灵活性。总结一下我们将使用LangChain作为基础组件库使用LangGraph来编排一个包含RAG检索和多智能体协作的复杂工作流并探讨如何通过MCP来动态集成外部工具最终构建一个功能强大、可维护的AI应用。2. 环境准备与版本说明工欲善其事必先利其器。为了避免版本依赖冲突我们使用conda创建独立的Python环境并锁定核心库的版本。2.1 创建并激活Conda环境# 创建名为 langgraph-tutorial 的Python 3.10环境 conda create -n langgraph-tutorial python3.10 -y # 激活环境 conda activate langgraph-tutorial2.2 安装核心依赖创建一个requirements.txt文件内容如下# 核心框架 langchain0.1.0 langchain-community0.0.10 langgraph0.0.26 # 向量数据库与嵌入模型以Chroma和OpenAI为例 chromadb0.4.22 langchain-openai0.0.5 openai1.6.1 tiktoken0.5.2 # 文档处理 pypdf3.17.4 unstructured0.12.2 # 可选用于MCP示例如使用SQLite工具 mcp0.1.0 # 注意MCP生态正在快速发展请关注官方GitHub获取最新版本 # 其他工具 python-dotenv1.0.0然后安装它们pip install -r requirements.txt2.3 配置API密钥我们需要一个LLM提供商。本文以OpenAI为例你也可以替换为其他兼容OpenAI API的模型如DeepSeek、Ollama本地模型等。获取OpenAI API Key访问 platform.openai.com 。在项目根目录创建.env文件并填入你的密钥OPENAI_API_KEYsk-your-actual-api-key-here安装python-dotenv已在上述步骤完成我们将在代码中加载它。2.4 项目结构预览在开始编码前先规划好项目结构这有助于代码管理。langgraph_agent_project/ ├── .env # 环境变量API密钥等 ├── requirements.txt # 项目依赖 ├── main.py # 主程序入口 ├── core/ # 核心模块 │ ├── __init__.py │ ├── graph_builder.py # LangGraph图定义 │ ├── agents.py # 各个智能体定义 │ └── state.py # 自定义状态定义 ├── knowledge/ # RAG知识库相关 │ ├── __init__.py │ ├── vector_store.py # 向量库初始化与检索 │ └── docs/ # 存放知识库文档PDF/TXT │ └── example.pdf ├── tools/ # 自定义工具 │ ├── __init__.py │ └── calculator.py # 示例工具 └── utils/ # 工具函数 ├── __init__.py └── config.py # 配置加载3. 核心原理与组件拆解3.1 LangGraph的核心状态图StateGraph一切工作流都围绕State展开。我们首先需要定义状态的“形状”。# core/state.py from typing import TypedDict, List, Annotated import operator from langchain_core.messages import AnyMessage class AgentState(TypedDict): 定义多智能体工作流的共享状态。 # 用户输入的问题 input: str # 存放所有消息历史用户、AI、工具 messages: Annotated[List[AnyMessage], operator.add] # RAG检索到的相关文档片段 retrieved_docs: List[str] # 当前负责处理的智能体名称 current_agent: str # 最终答案 final_answer: strTypedDict为状态字典提供类型提示使开发更清晰。Annotated[List[AnyMessage], operator.add]这是LangGraph的魔法所在。它声明messages字段是一个列表并且当多个节点修改它时默认使用operator.add即列表的extend操作来合并更新而不是覆盖。这完美契合了对话消息追加的场景。3.2 构建智能体Agent一个智能体通常由三部分组成LLMPromptTools。我们创建一个“研究员”智能体它擅长利用RAG知识库回答问题。# core/agents.py from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.agents import create_react_agent, AgentExecutor from langchain_core.tools import Tool from knowledge.vector_store import retrieve_docs # 假设我们有一个检索函数 def create_research_agent(): 创建一个具备RAG检索能力的研究员智能体。 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 定义工具RAG检索工具 retrieval_tool Tool( nameknowledge_base_search, funclambda query: retrieve_docs(query, k3), # 检索top3相关片段 description当用户的问题涉及公司知识、产品文档或历史记录时使用此工具从知识库中搜索相关信息。输入应为清晰的搜索查询语句。 ) # 构建提示词引导AI使用工具 prompt ChatPromptTemplate.from_messages([ (system, 你是一位严谨的研究员。你的任务是利用所有可用工具为用户提供准确、有依据的回答。 如果你从知识库中找到了相关信息请引用它。如果没找到或信息不足请如实说明。), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), # 用于ReAct格式的思考过程 ]) # 使用LangChain的create_react_agent创建智能体 agent create_react_agent(llmllm, tools[retrieval_tool], promptprompt) # 包装成执行器 agent_executor AgentExecutor(agentagent, tools[retrieval_tool], verboseTrue, handle_parsing_errorsTrue) return agent_executor3.3 构建RAG知识库RAG是增强模型回答准确性的关键。我们以处理PDF文档为例。# knowledge/vector_store.py import os from langchain_community.document_loaders import PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain_community.embeddings import CacheBackedEmbeddings from langchain.storage import LocalFileStore # 初始化嵌入模型和缓存 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) store LocalFileStore(./.cache/embeddings) cached_embedder CacheBackedEmbeddings.from_bytes_store( embeddings, store, namespaceembeddings.model ) # 持久化向量数据库路径 PERSIST_DIRECTORY ./knowledge/chroma_db def init_vector_store(docs_dir./knowledge/docs): 初始化或加载向量数据库。 if os.path.exists(PERSIST_DIRECTORY): # 如果已存在则直接加载 print(加载已有向量数据库...) vectorstore Chroma( persist_directoryPERSIST_DIRECTORY, embedding_functioncached_embedder ) else: # 否则读取文档并创建 print(创建新的向量数据库...) documents [] for filename in os.listdir(docs_dir): if filename.endswith(.pdf): file_path os.path.join(docs_dir, filename) loader PyPDFLoader(file_path) documents.extend(loader.load()) # 分割文本 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200, length_functionlen, is_separator_regexFalse, ) splits text_splitter.split_documents(documents) # 创建向量存储 vectorstore Chroma.from_documents( documentssplits, embeddingcached_embedder, persist_directoryPERSIST_DIRECTORY ) vectorstore.persist() return vectorstore # 全局向量存储对象 _vectorstore None def get_vector_store(): global _vectorstore if _vectorstore is None: _vectorstore init_vector_store() return _vectorstore def retrieve_docs(query: str, k: int 3) - str: 检索相关文档片段。 vectorstore get_vector_store() docs vectorstore.similarity_search(query, kk) # 将文档内容合并成一个字符串返回 return \n\n.join([doc.page_content for doc in docs])4. 完整实战构建多智能体协作工作流现在我们将所有组件串联起来用LangGraph构建一个“问题分类 - 专项处理 - 汇总回答”的多智能体系统。4.1 定义工作流节点节点就是操作状态的函数。我们定义三个节点router路由、research_agent_node研究员、general_agent_node通用助手。# core/graph_builder.py from langgraph.graph import StateGraph, END from .state import AgentState from .agents import create_research_agent, create_general_agent # 假设也有通用助手 from langchain_core.messages import HumanMessage, AIMessage def router_node(state: AgentState) - str: 路由节点根据输入决定下一个执行哪个智能体。 question state[input].lower() # 简单的关键词路由逻辑实际应用中可以用一个分类LLM来实现 if any(keyword in question for keyword in [产品, 文档, 手册, 如何配置]): return research_agent # 交给研究员使用RAG else: return general_agent # 交给通用助手 def research_agent_node(state: AgentState) - dict: 研究员智能体节点。 agent create_research_agent() # 从状态中获取最新的用户消息 user_input state[input] # 调用智能体执行器 result agent.invoke({input: user_input, chat_history: state[messages]}) # 更新状态将AI的回复添加到消息历史中 new_messages [AIMessage(contentresult[output])] return {messages: new_messages, final_answer: result[output], current_agent: research_agent} def general_agent_node(state: AgentState) - dict: 通用助手智能体节点。 agent create_general_agent() # 一个没有RAG工具的简单助手 user_input state[input] result agent.invoke({input: user_input, chat_history: state[messages]}) new_messages [AIMessage(contentresult[output])] return {messages: new_messages, final_answer: result[output], current_agent: general_agent}4.2 构建并编译图这是LangGraph的核心步骤我们将节点和边组装成一个可执行的工作流。# core/graph_builder.py (续) def create_agent_workflow() - StateGraph: 创建并返回编译好的多智能体工作流图。 # 1. 创建图并指定状态类型 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(router, router_node) workflow.add_node(research_agent, research_agent_node) workflow.add_node(general_agent, general_agent_node) # 3. 设置入口点 workflow.set_entry_point(router) # 4. 添加边路由逻辑 workflow.add_conditional_edges( router, router_node, # 路由函数同时也作为条件判断函数 { research_agent: research_agent, general_agent: general_agent } ) # 5. 从智能体节点到结束 workflow.add_edge(research_agent, END) workflow.add_edge(general_agent, END) # 6. 编译图 return workflow.compile() # 创建全局图应用实例 app create_agent_workflow()4.3 主程序与运行验证现在让我们写一个主程序来运行这个工作流。# main.py import asyncio from dotenv import load_dotenv from core.graph_builder import app from core.state import AgentState load_dotenv() # 加载环境变量 async def main(): print( 多智能体RAG系统启动 ) # 初始化状态 initial_state: AgentState { input: , messages: [], retrieved_docs: [], current_agent: , final_answer: } while True: user_input input(\n请输入您的问题 (输入 quit 退出): ) if user_input.lower() quit: break # 准备本次调用的状态 config {configurable: {thread_id: user_session_1}} # 支持多会话 inputs {**initial_state, input: user_input, messages: []} # 每次新问题清空历史消息或根据需求保留 # 调用图工作流 print(f\n[系统] 处理中...) try: # LangGraph应用可以同步调用也支持异步 result await app.ainvoke(inputs, configconfig) # result app.invoke(inputs, configconfig) # 同步方式 print(f\n[最终答案] {result[final_answer]}) print(f[处理智能体] {result[current_agent]}) except Exception as e: print(f\n[错误] 处理过程中发生异常: {e}) if __name__ __main__: asyncio.run(main())4.4 运行与测试确保你的knowledge/docs/目录下有一些PDF文档例如产品手册。在终端运行python main.py首次运行会创建向量数据库稍等片刻。输入问题测试测试RAG路径输入“咱们公司的主要产品是什么”。系统应路由到research_agent并从你提供的PDF中检索信息并回答。测试通用路径输入“讲一个笑话”。系统应路由到general_agent直接调用LLM生成回答。5. 进阶集成MCP模型上下文协议MCP允许我们以标准化的方式动态连接外部工具。假设我们想通过MCP连接一个SQLite数据库。5.1 理解MCP的角色MCP Server提供工具例如执行SQL查询、读取文件列表。我们需要运行或编写一个Server。MCP ClientLangChain可以作为Client动态发现并使用Server提供的工具。5.2 使用现有的MCP Server示例以一个简单的“计算器”和“时间”工具Server为例。首先你需要安装一个MCP Server。这里我们使用一个简单的示例Server你可能需要从社区寻找或自己实现。# 假设我们通过pip安装了一个示例MCP Server # pip install mcp-server-example # 然后运行它通常在某个端口如8080提供SSE服务5.3 在LangGraph中集成MCP工具修改core/agents.py让智能体能够使用MCP工具。# core/agents.py (新增函数) from langchain.agents import Tool import requests import json def create_mcp_tool_client(server_urlhttp://localhost:8080/sse): 创建一个连接到MCP Server并获取其工具列表的客户端。 # 注意这是一个高度简化的示例。实际应使用官方的MCP SDK。 # 这里模拟MCP的SSE连接和工具发现流程。 def list_tools(): # 模拟从MCP Server获取工具列表 # 实际应使用mcp库的Client return [calculator, get_current_time] def execute_tool(tool_name: str, arguments: dict) - str: # 模拟调用MCP工具 if tool_name calculator: expr arguments.get(expression, 0) try: # 警告在生产环境中直接eval是危险的此处仅为演示。 # 真实MCP Server会安全地处理计算。 result eval(expr) return f计算结果: {result} except Exception as e: return f计算错误: {e} elif tool_name get_current_time: from datetime import datetime return f当前时间: {datetime.now().isoformat()} else: return f未知工具: {tool_name} # 动态创建Tool对象 mcp_tools [] for tool_name in list_tools(): mcp_tools.append( Tool( namefmcp_{tool_name}, funclambda args, tntool_name: execute_tool(tn, args), descriptionfMCP Server提供的工具: {tool_name}。, args_schemaNone # 实际应有详细的参数schema ) ) return mcp_tools def create_agent_with_mcp(): 创建一个能使用MCP工具的智能体。 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) mcp_tools create_mcp_tool_client() # 获取MCP工具 all_tools [retrieval_tool] mcp_tools # 组合RAG工具和MCP工具 prompt ChatPromptTemplate.from_messages([ (system, 你可以使用知识库搜索工具和MCP工具如计算器、查询时间来帮助用户。请合理选择工具。), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) agent create_react_agent(llmllm, toolsall_tools, promptprompt) agent_executor AgentExecutor(agentagent, toolsall_tools, verboseTrue, handle_parsing_errorsTrue) return agent_executor然后你可以在graph_builder.py中创建一个新的节点或者替换现有的research_agent_node来使用这个更强大的智能体。6. 常见问题与排查思路在开发过程中你几乎一定会遇到以下问题。这里提供快速排查指南。问题现象常见原因解决思路ModuleNotFoundError: No module named langchain_community依赖未正确安装或版本冲突。1. 确认在正确的conda环境中。2. 运行pip install langchain-community。3. 检查requirements.txt版本尝试安装指定版本。OpenAI API调用超时或报错网络问题、API密钥错误、额度不足。1. 检查.env文件中的OPENAI_API_KEY是否正确。2. 运行curl测试API连通性。3. 登录OpenAI后台检查额度和账单。向量数据库检索结果不相关文档分块策略不当、嵌入模型不匹配、检索参数k太小。1. 调整RecursiveCharacterTextSplitter的chunk_size和chunk_overlap。2. 确保创建和查询时使用相同的嵌入模型。3. 增大检索数量k或尝试不同的相似度搜索方法如MMR。LangGraph节点不执行或流程卡住状态定义错误、边Edge配置错误、节点函数返回值格式不对。1. 检查State的TypedDict定义确保字段名一致。2. 使用app.get_graph().draw_mermaid()输出图结构可视化检查。3. 在节点函数内添加print语句调试确保函数被调用且返回正确的字典。智能体不调用工具直接胡编乱造提示词Prompt未明确要求使用工具、工具描述不清晰、LLM温度temperature过高。1. 在System Prompt中强调“你必须使用工具”。2. 完善工具的description明确使用场景和输入格式。3. 将LLM的temperature调低如0使其更倾向于遵循指令。RuntimeError: This event loop is already running在Jupyter Notebook或已有异步环境中错误地调用了异步方法。1. 在普通脚本中使用asyncio.run(main())。2. 在Jupyter中使用await app.ainvoke(...)并确保在异步函数中运行。MCP连接失败MCP Server未启动、URL错误、协议版本不兼容。1. 确认MCP Server进程正在运行。2. 检查连接URL和端口。3. 查阅MCP Server和Client的文档确保版本兼容。7. 最佳实践与工程建议将原型转化为稳定、可维护的生产级应用需要遵循以下实践。7.1 状态设计与管理最小化状态只将真正需要跨节点共享的数据放入State。避免放入大型对象如整个模型实例。明确更新逻辑善用Annotated注解如operator.add来定义字段的合并策略避免状态覆盖冲突。持久化状态对于需要长期记忆的会话可以将状态存储到数据库如Redis并在每次调用时加载。7.2 智能体与工具设计单一职责每个智能体应专注于一类任务如研究、写作、审核。工具描述应清晰、具体包含输入输出示例。工具验证与安全在工具函数内部务必对输入参数进行严格的验证和清理防止注入攻击尤其是在执行计算、命令、SQL时。优雅降级当某个工具如网络API失败时智能体应有后备方案而不是直接崩溃。7.3 RAG优化分块策略根据文档类型代码、论文、手册选择不同的分块器和分块大小。可以尝试语义分块。元数据过滤在向量化时存储元数据如来源文件、页码检索时可以进行过滤提高精度。重排序Re-ranking在初步向量检索后使用一个更精细的交叉编码器模型对结果进行重排序可以显著提升TOP1答案的相关性。检索后处理对检索到的文档片段进行去重、摘要或关键信息提取再喂给LLM可以节省上下文窗口。7.4 图工作流设计可视化调试务必使用app.get_graph().draw_mermaid()生成流程图这是理解和调试复杂工作流的神器。设置超时与重试对于可能超时的节点如外部API调用在节点逻辑或图层面设置超时和重试机制。实现检查点对于长工作流可以考虑在关键节点后保存状态快照便于错误恢复和继续执行。7.5 生产环境部署配置管理使用环境变量或配置管理工具如pydantic-settings来管理API密钥、模型参数、服务器地址等。日志与监控为每个节点调用、工具调用、LLM调用添加详细的日志。监控Token消耗、延迟和错误率。版本控制对State的定义、图的结构、提示词模板进行版本控制。任何更改都可能影响已有对话的兼容性。测试编写单元测试测试单个工具和节点和集成测试测试整个工作流对典型输入的处理。三天时间我们从LangChain和LangGraph的核心概念出发一步步构建了一个融合RAG检索、多智能体协作和MCP工具扩展的实战项目。你现在应该已经掌握了如何用LangChain组织AI应用的基本组件如何用LangGraph的状态图编排复杂、有状态的工作流如何构建和优化一个RAG系统来增强模型的知识以及如何通过MCP协议让智能体动态获得新能力。这套技术栈代表了当前AI应用工程化的前沿方向。学习的下一步是深入每一个环节探索更高效的向量检索库如FAISS,Weaviate尝试更强大的开源模型如Qwen2.5,Llama3设计更复杂的多智能体协作模式如辩论、评审或是将你的应用封装成API服务部署上线。记住所有的框架和协议都是为了降低复杂性。当你遇到难题时回归本质你的状态是什么你的节点如何操作状态边如何引导流程从这三个问题出发再复杂的智能体也能被清晰地设计和实现。 LangChain与LangGraph实战：构建多智能体协作与RAG知识库的AI应用
在构建AI应用时你是否遇到过这样的困境单个大模型能力有限无法处理复杂任务RAG系统虽然能检索知识但逻辑推理和流程控制能力薄弱想要实现多步骤协作的智能体却不知从何下手网上资料零散且不成体系。如果你正为此烦恼那么恭喜你这篇文章正是为你准备的。本文将带你系统性地掌握LangChain与LangGraph两大核心框架从零构建一个集成了多智能体协作、RAG知识库和MCP工具扩展的实战项目。无论你是希望快速入门的新手还是寻求项目落地的进阶开发者都能在这里找到从环境搭建、核心原理到避坑指南的完整闭环方案。我们将用三天时间彻底吃透Agent、RAG与MCP让你在AI应用开发的道路上少走99%的弯路。1. 背景与核心概念为什么需要LangChain与LangGraph在深入代码之前我们必须先理解我们正在解决的问题以及所使用的工具。AI应用开发早已不是简单地调用一个API而是涉及提示工程、工具调用、记忆管理、流程编排等多个复杂环节的系统工程。1.1 LangChainAI应用开发的“脚手架”LangChain是一个用于开发由语言模型驱动的应用程序的框架。你可以把它想象成乐高积木的基础连接件。它的核心价值在于标准化和模块化。标准化接口它定义了LLM、ChatModel、Tool、Memory、Chain等通用接口。无论底层是OpenAI的GPT、Anthropic的Claude还是开源的Llama、Qwen你都可以用同一套代码来调用极大地降低了切换模型带来的成本。模块化组件它将复杂的AI应用拆解成可复用的组件。例如PromptTemplate负责管理提示词VectorStore负责向量检索Agent负责决策和工具调用。你可以像搭积木一样组合这些组件。核心功能主要包括模型I/O与LLM对话、数据连接文档加载、向量化、链顺序调用、记忆保存对话历史、代理根据目标调用工具等。简单来说LangChain解决了“如何方便地使用大模型”的问题。1.2 LangGraph复杂工作流的“编排引擎”如果说LangChain提供了积木块那么LangGraph就是搭建复杂动态结构的设计图和控制器。它基于状态图StateGraph的概念专门用于构建有状态、多步骤、可能循环或分支的智能体Agent工作流。解决什么问题传统的LangChainAgent或Chain通常是线性的或简单的if-else逻辑。对于需要根据中间结果动态决定下一步、或者多个角色智能体之间需要协作的任务如“写代码-评审代码-执行测试”传统方式会变得非常臃肿和难以维护。核心概念状态State一个共享的字典存储工作流运行过程中的所有信息如用户问题、模型回复、工具执行结果等。节点Node一个执行单元可以是一个函数它读取和更新状态。例如一个“调用LLM”的节点一个“执行搜索”的节点。边Edge决定工作流下一个应该执行哪个节点的规则。可以是固定的always_go_to也可以根据状态内容动态决定conditional_edge。与LangChain Agent的关系你可以把LangGraph看作是构建更强大、更可控Agent的底层框架。一个经典的ReAct Agent可以用LangGraph清晰地表达为“思考-行动-观察”的循环。1.3 RAG、Agent与MCP构建智能应用的“三驾马车”理解了框架我们再来看看构建应用的核心模式。RAG检索增强生成解决大模型“知识陈旧”和“幻觉”问题。通过将外部知识库如文档、数据库向量化在回答问题时先进行相关检索再将检索到的片段作为上下文提供给模型从而生成更准确、更可靠的答案。Agent智能体一个能感知环境、进行决策并执行行动以实现目标的系统。在LLM语境下Agent通常指一个能调用工具如计算器、搜索引擎、API的LLM。多智能体则是由多个具有不同角色和能力的Agent协作完成任务。MCP模型上下文协议这是一个由Anthropic提出的新兴协议旨在标准化LLM与外部工具/数据源之间的连接方式。你可以把它理解为LLM界的“USB协议”。MCP Server提供工具和数据MCP Client如LangChain可以动态发现并使用这些工具无需硬编码。这极大地增强了AI应用的可扩展性和灵活性。总结一下我们将使用LangChain作为基础组件库使用LangGraph来编排一个包含RAG检索和多智能体协作的复杂工作流并探讨如何通过MCP来动态集成外部工具最终构建一个功能强大、可维护的AI应用。2. 环境准备与版本说明工欲善其事必先利其器。为了避免版本依赖冲突我们使用conda创建独立的Python环境并锁定核心库的版本。2.1 创建并激活Conda环境# 创建名为 langgraph-tutorial 的Python 3.10环境 conda create -n langgraph-tutorial python3.10 -y # 激活环境 conda activate langgraph-tutorial2.2 安装核心依赖创建一个requirements.txt文件内容如下# 核心框架 langchain0.1.0 langchain-community0.0.10 langgraph0.0.26 # 向量数据库与嵌入模型以Chroma和OpenAI为例 chromadb0.4.22 langchain-openai0.0.5 openai1.6.1 tiktoken0.5.2 # 文档处理 pypdf3.17.4 unstructured0.12.2 # 可选用于MCP示例如使用SQLite工具 mcp0.1.0 # 注意MCP生态正在快速发展请关注官方GitHub获取最新版本 # 其他工具 python-dotenv1.0.0然后安装它们pip install -r requirements.txt2.3 配置API密钥我们需要一个LLM提供商。本文以OpenAI为例你也可以替换为其他兼容OpenAI API的模型如DeepSeek、Ollama本地模型等。获取OpenAI API Key访问 platform.openai.com 。在项目根目录创建.env文件并填入你的密钥OPENAI_API_KEYsk-your-actual-api-key-here安装python-dotenv已在上述步骤完成我们将在代码中加载它。2.4 项目结构预览在开始编码前先规划好项目结构这有助于代码管理。langgraph_agent_project/ ├── .env # 环境变量API密钥等 ├── requirements.txt # 项目依赖 ├── main.py # 主程序入口 ├── core/ # 核心模块 │ ├── __init__.py │ ├── graph_builder.py # LangGraph图定义 │ ├── agents.py # 各个智能体定义 │ └── state.py # 自定义状态定义 ├── knowledge/ # RAG知识库相关 │ ├── __init__.py │ ├── vector_store.py # 向量库初始化与检索 │ └── docs/ # 存放知识库文档PDF/TXT │ └── example.pdf ├── tools/ # 自定义工具 │ ├── __init__.py │ └── calculator.py # 示例工具 └── utils/ # 工具函数 ├── __init__.py └── config.py # 配置加载3. 核心原理与组件拆解3.1 LangGraph的核心状态图StateGraph一切工作流都围绕State展开。我们首先需要定义状态的“形状”。# core/state.py from typing import TypedDict, List, Annotated import operator from langchain_core.messages import AnyMessage class AgentState(TypedDict): 定义多智能体工作流的共享状态。 # 用户输入的问题 input: str # 存放所有消息历史用户、AI、工具 messages: Annotated[List[AnyMessage], operator.add] # RAG检索到的相关文档片段 retrieved_docs: List[str] # 当前负责处理的智能体名称 current_agent: str # 最终答案 final_answer: strTypedDict为状态字典提供类型提示使开发更清晰。Annotated[List[AnyMessage], operator.add]这是LangGraph的魔法所在。它声明messages字段是一个列表并且当多个节点修改它时默认使用operator.add即列表的extend操作来合并更新而不是覆盖。这完美契合了对话消息追加的场景。3.2 构建智能体Agent一个智能体通常由三部分组成LLMPromptTools。我们创建一个“研究员”智能体它擅长利用RAG知识库回答问题。# core/agents.py from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.agents import create_react_agent, AgentExecutor from langchain_core.tools import Tool from knowledge.vector_store import retrieve_docs # 假设我们有一个检索函数 def create_research_agent(): 创建一个具备RAG检索能力的研究员智能体。 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 定义工具RAG检索工具 retrieval_tool Tool( nameknowledge_base_search, funclambda query: retrieve_docs(query, k3), # 检索top3相关片段 description当用户的问题涉及公司知识、产品文档或历史记录时使用此工具从知识库中搜索相关信息。输入应为清晰的搜索查询语句。 ) # 构建提示词引导AI使用工具 prompt ChatPromptTemplate.from_messages([ (system, 你是一位严谨的研究员。你的任务是利用所有可用工具为用户提供准确、有依据的回答。 如果你从知识库中找到了相关信息请引用它。如果没找到或信息不足请如实说明。), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), # 用于ReAct格式的思考过程 ]) # 使用LangChain的create_react_agent创建智能体 agent create_react_agent(llmllm, tools[retrieval_tool], promptprompt) # 包装成执行器 agent_executor AgentExecutor(agentagent, tools[retrieval_tool], verboseTrue, handle_parsing_errorsTrue) return agent_executor3.3 构建RAG知识库RAG是增强模型回答准确性的关键。我们以处理PDF文档为例。# knowledge/vector_store.py import os from langchain_community.document_loaders import PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain_community.embeddings import CacheBackedEmbeddings from langchain.storage import LocalFileStore # 初始化嵌入模型和缓存 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) store LocalFileStore(./.cache/embeddings) cached_embedder CacheBackedEmbeddings.from_bytes_store( embeddings, store, namespaceembeddings.model ) # 持久化向量数据库路径 PERSIST_DIRECTORY ./knowledge/chroma_db def init_vector_store(docs_dir./knowledge/docs): 初始化或加载向量数据库。 if os.path.exists(PERSIST_DIRECTORY): # 如果已存在则直接加载 print(加载已有向量数据库...) vectorstore Chroma( persist_directoryPERSIST_DIRECTORY, embedding_functioncached_embedder ) else: # 否则读取文档并创建 print(创建新的向量数据库...) documents [] for filename in os.listdir(docs_dir): if filename.endswith(.pdf): file_path os.path.join(docs_dir, filename) loader PyPDFLoader(file_path) documents.extend(loader.load()) # 分割文本 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200, length_functionlen, is_separator_regexFalse, ) splits text_splitter.split_documents(documents) # 创建向量存储 vectorstore Chroma.from_documents( documentssplits, embeddingcached_embedder, persist_directoryPERSIST_DIRECTORY ) vectorstore.persist() return vectorstore # 全局向量存储对象 _vectorstore None def get_vector_store(): global _vectorstore if _vectorstore is None: _vectorstore init_vector_store() return _vectorstore def retrieve_docs(query: str, k: int 3) - str: 检索相关文档片段。 vectorstore get_vector_store() docs vectorstore.similarity_search(query, kk) # 将文档内容合并成一个字符串返回 return \n\n.join([doc.page_content for doc in docs])4. 完整实战构建多智能体协作工作流现在我们将所有组件串联起来用LangGraph构建一个“问题分类 - 专项处理 - 汇总回答”的多智能体系统。4.1 定义工作流节点节点就是操作状态的函数。我们定义三个节点router路由、research_agent_node研究员、general_agent_node通用助手。# core/graph_builder.py from langgraph.graph import StateGraph, END from .state import AgentState from .agents import create_research_agent, create_general_agent # 假设也有通用助手 from langchain_core.messages import HumanMessage, AIMessage def router_node(state: AgentState) - str: 路由节点根据输入决定下一个执行哪个智能体。 question state[input].lower() # 简单的关键词路由逻辑实际应用中可以用一个分类LLM来实现 if any(keyword in question for keyword in [产品, 文档, 手册, 如何配置]): return research_agent # 交给研究员使用RAG else: return general_agent # 交给通用助手 def research_agent_node(state: AgentState) - dict: 研究员智能体节点。 agent create_research_agent() # 从状态中获取最新的用户消息 user_input state[input] # 调用智能体执行器 result agent.invoke({input: user_input, chat_history: state[messages]}) # 更新状态将AI的回复添加到消息历史中 new_messages [AIMessage(contentresult[output])] return {messages: new_messages, final_answer: result[output], current_agent: research_agent} def general_agent_node(state: AgentState) - dict: 通用助手智能体节点。 agent create_general_agent() # 一个没有RAG工具的简单助手 user_input state[input] result agent.invoke({input: user_input, chat_history: state[messages]}) new_messages [AIMessage(contentresult[output])] return {messages: new_messages, final_answer: result[output], current_agent: general_agent}4.2 构建并编译图这是LangGraph的核心步骤我们将节点和边组装成一个可执行的工作流。# core/graph_builder.py (续) def create_agent_workflow() - StateGraph: 创建并返回编译好的多智能体工作流图。 # 1. 创建图并指定状态类型 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(router, router_node) workflow.add_node(research_agent, research_agent_node) workflow.add_node(general_agent, general_agent_node) # 3. 设置入口点 workflow.set_entry_point(router) # 4. 添加边路由逻辑 workflow.add_conditional_edges( router, router_node, # 路由函数同时也作为条件判断函数 { research_agent: research_agent, general_agent: general_agent } ) # 5. 从智能体节点到结束 workflow.add_edge(research_agent, END) workflow.add_edge(general_agent, END) # 6. 编译图 return workflow.compile() # 创建全局图应用实例 app create_agent_workflow()4.3 主程序与运行验证现在让我们写一个主程序来运行这个工作流。# main.py import asyncio from dotenv import load_dotenv from core.graph_builder import app from core.state import AgentState load_dotenv() # 加载环境变量 async def main(): print( 多智能体RAG系统启动 ) # 初始化状态 initial_state: AgentState { input: , messages: [], retrieved_docs: [], current_agent: , final_answer: } while True: user_input input(\n请输入您的问题 (输入 quit 退出): ) if user_input.lower() quit: break # 准备本次调用的状态 config {configurable: {thread_id: user_session_1}} # 支持多会话 inputs {**initial_state, input: user_input, messages: []} # 每次新问题清空历史消息或根据需求保留 # 调用图工作流 print(f\n[系统] 处理中...) try: # LangGraph应用可以同步调用也支持异步 result await app.ainvoke(inputs, configconfig) # result app.invoke(inputs, configconfig) # 同步方式 print(f\n[最终答案] {result[final_answer]}) print(f[处理智能体] {result[current_agent]}) except Exception as e: print(f\n[错误] 处理过程中发生异常: {e}) if __name__ __main__: asyncio.run(main())4.4 运行与测试确保你的knowledge/docs/目录下有一些PDF文档例如产品手册。在终端运行python main.py首次运行会创建向量数据库稍等片刻。输入问题测试测试RAG路径输入“咱们公司的主要产品是什么”。系统应路由到research_agent并从你提供的PDF中检索信息并回答。测试通用路径输入“讲一个笑话”。系统应路由到general_agent直接调用LLM生成回答。5. 进阶集成MCP模型上下文协议MCP允许我们以标准化的方式动态连接外部工具。假设我们想通过MCP连接一个SQLite数据库。5.1 理解MCP的角色MCP Server提供工具例如执行SQL查询、读取文件列表。我们需要运行或编写一个Server。MCP ClientLangChain可以作为Client动态发现并使用Server提供的工具。5.2 使用现有的MCP Server示例以一个简单的“计算器”和“时间”工具Server为例。首先你需要安装一个MCP Server。这里我们使用一个简单的示例Server你可能需要从社区寻找或自己实现。# 假设我们通过pip安装了一个示例MCP Server # pip install mcp-server-example # 然后运行它通常在某个端口如8080提供SSE服务5.3 在LangGraph中集成MCP工具修改core/agents.py让智能体能够使用MCP工具。# core/agents.py (新增函数) from langchain.agents import Tool import requests import json def create_mcp_tool_client(server_urlhttp://localhost:8080/sse): 创建一个连接到MCP Server并获取其工具列表的客户端。 # 注意这是一个高度简化的示例。实际应使用官方的MCP SDK。 # 这里模拟MCP的SSE连接和工具发现流程。 def list_tools(): # 模拟从MCP Server获取工具列表 # 实际应使用mcp库的Client return [calculator, get_current_time] def execute_tool(tool_name: str, arguments: dict) - str: # 模拟调用MCP工具 if tool_name calculator: expr arguments.get(expression, 0) try: # 警告在生产环境中直接eval是危险的此处仅为演示。 # 真实MCP Server会安全地处理计算。 result eval(expr) return f计算结果: {result} except Exception as e: return f计算错误: {e} elif tool_name get_current_time: from datetime import datetime return f当前时间: {datetime.now().isoformat()} else: return f未知工具: {tool_name} # 动态创建Tool对象 mcp_tools [] for tool_name in list_tools(): mcp_tools.append( Tool( namefmcp_{tool_name}, funclambda args, tntool_name: execute_tool(tn, args), descriptionfMCP Server提供的工具: {tool_name}。, args_schemaNone # 实际应有详细的参数schema ) ) return mcp_tools def create_agent_with_mcp(): 创建一个能使用MCP工具的智能体。 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) mcp_tools create_mcp_tool_client() # 获取MCP工具 all_tools [retrieval_tool] mcp_tools # 组合RAG工具和MCP工具 prompt ChatPromptTemplate.from_messages([ (system, 你可以使用知识库搜索工具和MCP工具如计算器、查询时间来帮助用户。请合理选择工具。), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) agent create_react_agent(llmllm, toolsall_tools, promptprompt) agent_executor AgentExecutor(agentagent, toolsall_tools, verboseTrue, handle_parsing_errorsTrue) return agent_executor然后你可以在graph_builder.py中创建一个新的节点或者替换现有的research_agent_node来使用这个更强大的智能体。6. 常见问题与排查思路在开发过程中你几乎一定会遇到以下问题。这里提供快速排查指南。问题现象常见原因解决思路ModuleNotFoundError: No module named langchain_community依赖未正确安装或版本冲突。1. 确认在正确的conda环境中。2. 运行pip install langchain-community。3. 检查requirements.txt版本尝试安装指定版本。OpenAI API调用超时或报错网络问题、API密钥错误、额度不足。1. 检查.env文件中的OPENAI_API_KEY是否正确。2. 运行curl测试API连通性。3. 登录OpenAI后台检查额度和账单。向量数据库检索结果不相关文档分块策略不当、嵌入模型不匹配、检索参数k太小。1. 调整RecursiveCharacterTextSplitter的chunk_size和chunk_overlap。2. 确保创建和查询时使用相同的嵌入模型。3. 增大检索数量k或尝试不同的相似度搜索方法如MMR。LangGraph节点不执行或流程卡住状态定义错误、边Edge配置错误、节点函数返回值格式不对。1. 检查State的TypedDict定义确保字段名一致。2. 使用app.get_graph().draw_mermaid()输出图结构可视化检查。3. 在节点函数内添加print语句调试确保函数被调用且返回正确的字典。智能体不调用工具直接胡编乱造提示词Prompt未明确要求使用工具、工具描述不清晰、LLM温度temperature过高。1. 在System Prompt中强调“你必须使用工具”。2. 完善工具的description明确使用场景和输入格式。3. 将LLM的temperature调低如0使其更倾向于遵循指令。RuntimeError: This event loop is already running在Jupyter Notebook或已有异步环境中错误地调用了异步方法。1. 在普通脚本中使用asyncio.run(main())。2. 在Jupyter中使用await app.ainvoke(...)并确保在异步函数中运行。MCP连接失败MCP Server未启动、URL错误、协议版本不兼容。1. 确认MCP Server进程正在运行。2. 检查连接URL和端口。3. 查阅MCP Server和Client的文档确保版本兼容。7. 最佳实践与工程建议将原型转化为稳定、可维护的生产级应用需要遵循以下实践。7.1 状态设计与管理最小化状态只将真正需要跨节点共享的数据放入State。避免放入大型对象如整个模型实例。明确更新逻辑善用Annotated注解如operator.add来定义字段的合并策略避免状态覆盖冲突。持久化状态对于需要长期记忆的会话可以将状态存储到数据库如Redis并在每次调用时加载。7.2 智能体与工具设计单一职责每个智能体应专注于一类任务如研究、写作、审核。工具描述应清晰、具体包含输入输出示例。工具验证与安全在工具函数内部务必对输入参数进行严格的验证和清理防止注入攻击尤其是在执行计算、命令、SQL时。优雅降级当某个工具如网络API失败时智能体应有后备方案而不是直接崩溃。7.3 RAG优化分块策略根据文档类型代码、论文、手册选择不同的分块器和分块大小。可以尝试语义分块。元数据过滤在向量化时存储元数据如来源文件、页码检索时可以进行过滤提高精度。重排序Re-ranking在初步向量检索后使用一个更精细的交叉编码器模型对结果进行重排序可以显著提升TOP1答案的相关性。检索后处理对检索到的文档片段进行去重、摘要或关键信息提取再喂给LLM可以节省上下文窗口。7.4 图工作流设计可视化调试务必使用app.get_graph().draw_mermaid()生成流程图这是理解和调试复杂工作流的神器。设置超时与重试对于可能超时的节点如外部API调用在节点逻辑或图层面设置超时和重试机制。实现检查点对于长工作流可以考虑在关键节点后保存状态快照便于错误恢复和继续执行。7.5 生产环境部署配置管理使用环境变量或配置管理工具如pydantic-settings来管理API密钥、模型参数、服务器地址等。日志与监控为每个节点调用、工具调用、LLM调用添加详细的日志。监控Token消耗、延迟和错误率。版本控制对State的定义、图的结构、提示词模板进行版本控制。任何更改都可能影响已有对话的兼容性。测试编写单元测试测试单个工具和节点和集成测试测试整个工作流对典型输入的处理。三天时间我们从LangChain和LangGraph的核心概念出发一步步构建了一个融合RAG检索、多智能体协作和MCP工具扩展的实战项目。你现在应该已经掌握了如何用LangChain组织AI应用的基本组件如何用LangGraph的状态图编排复杂、有状态的工作流如何构建和优化一个RAG系统来增强模型的知识以及如何通过MCP协议让智能体动态获得新能力。这套技术栈代表了当前AI应用工程化的前沿方向。学习的下一步是深入每一个环节探索更高效的向量检索库如FAISS,Weaviate尝试更强大的开源模型如Qwen2.5,Llama3设计更复杂的多智能体协作模式如辩论、评审或是将你的应用封装成API服务部署上线。记住所有的框架和协议都是为了降低复杂性。当你遇到难题时回归本质你的状态是什么你的节点如何操作状态边如何引导流程从这三个问题出发再复杂的智能体也能被清晰地设计和实现。