企业官网建设流程全解析

在构建AI应用时，你是否遇到过这样的困境：单个大模型能力有限，无法处理复杂任务；RAG系统虽然能检索知识，但逻辑推理和流程控制能力薄弱；想要实现多步骤协作的智能体，却不知从何下手，网上资料零散且不成体系。如果你正为此烦恼，那么恭喜你，这篇文章正是为你准备的。

本文将带你系统性地掌握LangChain与LangGraph两大核心框架，从零构建一个集成了多智能体协作、RAG知识库和MCP工具扩展的实战项目。无论你是希望快速入门的新手，还是寻求项目落地的进阶开发者，都能在这里找到从环境搭建、核心原理到避坑指南的完整闭环方案。我们将用三天时间，彻底吃透Agent、RAG与MCP，让你在AI应用开发的道路上少走99%的弯路。

1. 背景与核心概念：为什么需要LangChain与LangGraph？

在深入代码之前，我们必须先理解我们正在解决的问题以及所使用的工具。AI应用开发早已不是简单地调用一个API，而是涉及提示工程、工具调用、记忆管理、流程编排等多个复杂环节的系统工程。

1.1 LangChain：AI应用开发的“脚手架”

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 python=3.10 -y # 激活环境 conda activate langgraph-tutorial

2.2 安装核心依赖

创建一个requirements.txt文件，内容如下：

# 核心框架 langchain==0.1.0 langchain-community==0.0.10 langgraph==0.0.26 # 向量数据库与嵌入模型（以Chroma和OpenAI为例） chromadb==0.4.22 langchain-openai==0.0.5 openai==1.6.1 tiktoken==0.5.2 # 文档处理 pypdf==3.17.4 unstructured==0.12.2 # 可选：用于MCP示例（如使用SQLite工具） mcp==0.1.0 # 注意：MCP生态正在快速发展，请关注官方GitHub获取最新版本 # 其他工具 python-dotenv==1.0.0

然后安装它们：

pip install -r requirements.txt

2.3 配置API密钥

我们需要一个LLM提供商。本文以OpenAI为例，你也可以替换为其他兼容OpenAI API的模型（如DeepSeek、Ollama本地模型等）。

1. 获取OpenAI API Key：访问 platform.openai.com 。
2. 在项目根目录创建.env文件，并填入你的密钥：

   OPENAI_API_KEY=sk-your-actual-api-key-here

3. 安装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: str

• TypedDict：为状态字典提供类型提示，使开发更清晰。
• Annotated[List[AnyMessage], operator.add]：这是LangGraph的魔法所在。它声明messages字段是一个列表，并且当多个节点修改它时，默认使用operator.add（即列表的extend操作）来合并更新，而不是覆盖。这完美契合了对话消息追加的场景。

3.2 构建智能体（Agent）

一个智能体通常由三部分组成：LLM+Prompt+Tools。我们创建一个“研究员”智能体，它擅长利用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(model="gpt-3.5-turbo", temperature=0) # 定义工具：RAG检索工具 retrieval_tool = Tool( name="knowledge_base_search", func=lambda query: retrieve_docs(query, k=3), # 检索top3相关片段 description="""当用户的问题涉及公司知识、产品文档或历史记录时，使用此工具从知识库中搜索相关信息。输入应为清晰的搜索查询语句。""" ) # 构建提示词，引导AI使用工具 prompt = ChatPromptTemplate.from_messages([ ("system", """你是一位严谨的研究员。你的任务是利用所有可用工具，为用户提供准确、有依据的回答。 如果你从知识库中找到了相关信息，请引用它。如果没找到或信息不足，请如实说明。"""), MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), # 用于ReAct格式的思考过程 ]) # 使用LangChain的create_react_agent创建智能体 agent = create_react_agent(llm=llm, tools=[retrieval_tool], prompt=prompt) # 包装成执行器 agent_executor = AgentExecutor(agent=agent, tools=[retrieval_tool], verbose=True, handle_parsing_errors=True) return agent_executor

3.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(model="text-embedding-3-small") store = LocalFileStore("./.cache/embeddings") cached_embedder = CacheBackedEmbeddings.from_bytes_store( embeddings, store, namespace=embeddings.model ) # 持久化向量数据库路径 PERSIST_DIRECTORY = "./knowledge/chroma_db" def init_vector_store(docs_dir="./knowledge/docs"): """初始化或加载向量数据库。""" if os.path.exists(PERSIST_DIRECTORY): # 如果已存在，则直接加载 print("加载已有向量数据库...") vectorstore = Chroma( persist_directory=PERSIST_DIRECTORY, embedding_function=cached_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_size=1000, chunk_overlap=200, length_function=len, is_separator_regex=False, ) splits = text_splitter.split_documents(documents) # 创建向量存储 vectorstore = Chroma.from_documents( documents=splits, embedding=cached_embedder, persist_directory=PERSIST_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, k=k) # 将文档内容合并成一个字符串返回 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(content=result["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(content=result["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, config=config) # result = app.invoke(inputs, config=config) # 同步方式 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 运行与测试

1. 确保你的knowledge/docs/目录下有一些PDF文档（例如产品手册）。
2. 在终端运行：

   python main.py

3. 首次运行会创建向量数据库，稍等片刻。
4. 输入问题测试：
   • 测试RAG路径：输入“咱们公司的主要产品是什么？”。系统应路由到research_agent，并从你提供的PDF中检索信息并回答。
   • 测试通用路径：输入“讲一个笑话”。系统应路由到general_agent，直接调用LLM生成回答。

5. 进阶：集成MCP（模型上下文协议）

MCP允许我们以标准化的方式动态连接外部工具。假设我们想通过MCP连接一个SQLite数据库。

5.1 理解MCP的角色

• MCP Server：提供工具（例如，执行SQL查询、读取文件列表）。我们需要运行或编写一个Server。
• MCP Client：LangChain可以作为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_url="http://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( name=f"mcp_{tool_name}", func=lambda args, tn=tool_name: execute_tool(tn, args), description=f"MCP Server提供的工具: {tool_name}。", args_schema=None # 实际应有详细的参数schema ) ) return mcp_tools def create_agent_with_mcp(): """创建一个能使用MCP工具的智能体。""" llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) mcp_tools = create_mcp_tool_client() # 获取MCP工具 all_tools = [retrieval_tool] + mcp_tools # 组合RAG工具和MCP工具 prompt = ChatPromptTemplate.from_messages([ ("system", "你可以使用知识库搜索工具和MCP工具（如计算器、查询时间）来帮助用户。请合理选择工具。"), MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), ]) agent = create_react_agent(llm=llm, tools=all_tools, prompt=prompt) agent_executor = AgentExecutor(agent=agent, tools=all_tools, verbose=True, handle_parsing_errors=True) return agent_executor

然后，你可以在graph_builder.py中创建一个新的节点，或者替换现有的research_agent_node来使用这个更强大的智能体。

6. 常见问题与排查思路

在开发过程中，你几乎一定会遇到以下问题。这里提供快速排查指南。

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服务部署上线。

记住，所有的框架和协议都是为了降低复杂性。当你遇到难题时，回归本质：你的状态是什么？你的节点如何操作状态？边如何引导流程？从这三个问题出发，再复杂的智能体也能被清晰地设计和实现。