# LearnAgent **Repository Path**: samsom/learn-agent ## Basic Information - **Project Name**: LearnAgent - **Description**: 一个分阶段学习 LangChain、LangGraph 和 MCP 框架的实践项目，从基础 LLM 调用到构建完整的 RAG、Agent 及多 Agent 协作系统。 - **Primary Language**: Python - **License**: MulanPSL-2.0 - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 1 - **Forks**: 0 - **Created**: 2026-07-17 - **Last Updated**: 2026-09-15 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # LangChain & LangGraph 学习项目 一个分阶段学习 LangChain、LangGraph 和 MCP 框架的实践项目，从基础 LLM 调用到构建完整的 RAG、Agent 及多 Agent 协作系统。 ## 项目结构 ``` LearnProject/ ├── .env # 环境变量配置（API Key） ├── .gitignore ├── file/ │ └── test.txt # RAG 测试文档 ├── README.md ├── langChain_base/ # LangChain 基础学习 │ ├── stage1.py # 基础 LLM 调用 │ ├── stage2.py # 提示词模板 │ ├── stage3.py # 输出解析器与链式调用 │ ├── stage4.py # 对话历史管理 │ ├── stage5.py # RAG（检索增强生成） │ └── stage6.py # 工具与 Agent ├── langGraph_base/ # LangGraph 基础学习 │ ├── stage_graph_1.py # State/Node/Edge 基础 │ ├── stage_graph_2.py # 条件分支路由 │ ├── stage_graph_3.py # 手动构建 ReAct Agent │ ├── stage_graph_4.py # Human-in-the-loop │ └── stage_graph_5.py # 多 Agent 协作 ├── langGraph_mcp/ # LangGraph MCP 学习 │ ├── mcp_math_server.py # MCP 数学服务器 │ ├── mcp_weather_server.py # MCP 天气服务器 │ ├── stage_mcp_single.py # 单 MCP 服务器集成 │ └── stage_mcp_multi.py # 多 MCP 服务器集成 └── smart_assistant/ # 综合实战 └── main.py # 智能助手系统（整合所有知识点） ``` ## 环境要求 - Python 3.8+ - 智谱 AI API Key（或其他兼容 OpenAI 接口的大模型 API） ## 安装依赖 ```bash # LangChain 基础依赖 pip install langchain langchain-openai langchain-community langchain-text-splitters langgraph chromadb python-dotenv # LangGraph MCP 额外依赖 pip install mcp langchain-mcp-adapters ``` ## 配置 在项目根目录创建 `.env` 文件，填入你的 API Key： ``` OPENAI_API_KEY=your_api_key_here ``` ## 学习阶段 ### Stage 1：基础 LLM 调用 演示如何使用 `ChatOpenAI` 调用大语言模型。 **知识要点：** - `load_dotenv()`：从 `.env` 文件加载环境变量，避免硬编码 API Key - `ChatOpenAI`：LangChain 的聊天模型封装，兼容 OpenAI 接口格式 - `model`：指定模型名称，此处使用智谱 GLM-4-Flash - `base_url`：自定义 API 端点，可对接第三方兼容服务 - `temperature`：控制输出随机性，0 为确定性输出，1 为高随机性 - `invoke()`：同步调用模型，返回 `AIMessage` 对象 - `.content`：从 `AIMessage` 中提取纯文本内容 ```bash python stage1.py ``` ### Stage 2：提示词模板 使用 `ChatPromptTemplate` 构建可复用的提示词模板，支持变量替换。 **知识要点：** - `ChatPromptTemplate.from_messages()`：从消息列表创建模板 - 消息角色：`system`（系统指令）、`human`（用户输入）、`ai`（助手回复） - `{变量名}`：模板占位符，调用时传入字典进行替换 - 模板与模型解耦：同一模板可复用于不同模型 - `.invoke(dict)`：将变量填入模板，生成最终消息列表 ```bash python stage2.py ``` ### Stage 3：链式调用（LCEL） 通过管道符 `|` 将 prompt、llm、parser 串联成一个完整的处理链。 **知识要点：** - LCEL（LangChain Expression Language）：LangChain 的声明式链式语法 - 管道符 `|`：将组件串联，前一个组件的输出作为后一个的输入 - `StrOutputParser()`：将 `AIMessage` 转为纯字符串，简化后续处理 - 链式调用优势：代码简洁、易于组合、支持流式输出 - `chain.invoke()`：一次调用跑完整个流程 ```bash python stage3.py ``` ### Stage 4：对话历史 使用 `RunnableWithMessageHistory` 和 `InMemoryChatMessageHistory` 实现多轮对话记忆。 **知识要点：** - `MessagesPlaceholder`：在模板中预留历史消息插入位置 - `InMemoryChatMessageHistory`：内存中的对话历史存储 - `RunnableWithMessageHistory`：自动管理对话历史的包装器 - `session_id`：会话标识符，不同 ID 对应不同对话历史 - `get_session_history`：回调函数，根据 session_id 返回对应的历史存储 - 工作流程：调用前自动读取历史 → 拼入 prompt → 调用模型 → 自动保存新历史 ```bash python stage4.py ``` ### Stage 5：RAG（检索增强生成） 文档加载 → 文本分割 → 向量存储 → 检索 → 生成回答。 **知识要点：** - **文档加载**：`TextLoader` 读取本地文本文件，返回 `Document` 对象 - **文本分割**：`RecursiveCharacterTextSplitter` 按语义递归分割长文本 - `chunk_size`：每个文本块的最大字符数 - `chunk_overlap`：相邻块的重叠字符数，保持上下文连贯 - **向量嵌入**：`OpenAIEmbeddings` 将文本转为向量表示 - **向量存储**：`Chroma` 本地向量数据库，存储和检索嵌入向量 - **检索器**：`as_retriever()` 创建检索接口，`k=3` 返回最相关的 3 个文档块 - **RAG 链**：`retriever | format_docs` 检索文档 → `RunnablePassthrough` 透传问题 → 模型生成答案 ```bash python stage5.py ``` ### Stage 6：工具与 Agent 使用 `@tool` 装饰器定义工具，通过 `create_react_agent` 创建能自主调用工具的 Agent。 **知识要点：** - `@tool` 装饰器：将普通函数转为 LangChain 工具 - **docstring 的关键作用**：Agent 根据文档字符串判断何时调用该工具 - `create_react_agent`：创建 ReAct 模式的 Agent - **ReAct 循环**：思考（Thought）→ 行动（Action）→ 观察（Observation）→ 循环 - `temperature=0`：让 Agent 决策更稳定、确定性更强 - Agent 优势：可根据任务自主选择工具，无需硬编码调用逻辑 - 输入输出：通过 `messages` 列表交互，格式为 `[("human", "问题")]` ```bash python stage6.py ``` --- ## LangGraph 基础学习阶段 ### Stage Graph-1：State/Node/Edge 基础 理解 LangGraph 的三要素：State（状态）、Node（节点）、Edge（边）。 **类比理解：** - **State** = 工作台上的工件（所有节点共享的数据） - **Node** = 工人（处理数据的函数） - **Edge** = 流水线（连接工人之间的传送带） - **Graph** = 整条流水线蓝图 **知识要点：** - `State(TypedDict)`：用 TypedDict 定义状态结构，所有节点共享 - `Node 函数`：接收 State 作为参数，返回要更新的字段字典 - `StateGraph(State)`：图的构建器，传入 State 类型定义 - `add_node("name", func)`：添加节点，起一个名字 - `add_edge(START, "node")`：添加边，定义执行顺序 - `START` / `END`：LangGraph 内置的特殊节点，表示入口和出口 - `compile()`：把构建器变成可执行的图对象 - `graph.invoke(state)`：运行图，传入初始 State ```python # stage_graph_1.py 核心代码 from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): input: str result: str def greeting_node(state: State) -> dict: return {"result": f"你好！你说了：{state['input']}"} graph_builder = StateGraph(State) graph_builder.add_node("greeting", greeting_node) graph_builder.add_edge(START, "greeting") graph_builder.add_edge("greeting", END) graph = graph_builder.compile() result = graph.invoke({"input": "Hello", "result": ""}) ``` ### Stage Graph-2：条件分支路由 根据 State 中的数据决定走哪条路，实现动态路由。 **知识要点：** - `add_conditional_edges(起点, 判断函数, {返回值: 目标节点})`：条件分支 - **判断函数**：接收 State，返回字符串（路由标识） - **路由映射**：字典形式，判断函数返回值 → 目标节点名 - 支持多路分支：math / translate / chat 等 ```python # 条件分支核心代码 def route_by_category(state: State) -> str: return state["category"] # "math" / "translate" / "chat" graph_builder.add_conditional_edges( "classify", # 从 classify 节点出发 route_by_category, # 用这个函数决定走哪条路 { # 返回值 → 目标节点 的映射 "math": "math", "translate": "translate", "chat": "chat", } ) ``` ### Stage Graph-3：手动构建 ReAct Agent 用 LangGraph 手动实现 ReAct 循环（思考 → 行动 → 观察 → 循环），理解底层原理。 **知识要点：** - **循环能力**：LangGraph 的核心优势，LangChain 的 Chain 做不到 - `Annotated[list, add_messages]`：消息列表的更新方式是「追加」而非「替换」 - `llm.bind_tools([...])`：把工具绑定到 LLM，让 LLM 知道有哪些工具可用 - `response.tool_calls`：LLM 返回的工具调用请求列表 - `ToolMessage`：包装工具执行结果的消息类型 - **条件循环**：`should_continue` 函数判断是否需要继续调用工具 ```python # ReAct 循环核心代码 def think_node(state: State) -> dict: response = llm.invoke(state["messages"]) return {"messages": [response]} # 追加 AI 回复（可能包含 tool_calls） def act_node(state: State) -> dict: tool_messages = [] for tool_call in state["messages"][-1].tool_calls: result = TOOLS[tool_call["name"]].invoke(tool_call["args"]) tool_messages.append(ToolMessage(content=str(result), tool_call_id=tool_call["id"])) return {"messages": tool_messages} def should_continue(state: State) -> str: if hasattr(state["messages"][-1], "tool_calls") and state["messages"][-1].tool_calls: return "act" # 继续调用工具 return END # 给出最终回答 # act 完成后回到 think（形成循环！） graph_builder.add_edge("act", "think") ``` ### Stage Graph-4：Human-in-the-loop 在图执行过程中暂停，等待人类确认后再继续。 **使用场景：** - 发邮件前让人类确认内容 - 执行敏感操作前要求审批 - 让人类选择下一步方向 **知识要点：** - `interrupt(value)`：暂停点，图执行到此处会暂停，返回值展示给审批人 - `Command(resume=value)`：恢复执行，传入的值成为 `interrupt()` 的返回值 - `MemorySaver`：内存检查点，保存暂停时的状态（生产环境用数据库） - `graph.stream()` + `interrupt()`：流式执行，遇到 interrupt 自动暂停 - `thread_id`：检查点按 thread_id 保存状态，支持多会话隔离 ```python # Human-in-the-loop 核心代码 from langgraph.checkpoint.memory import MemorySaver from langgraph.types import interrupt, Command def human_approval_node(state: State) -> dict: decision = interrupt({ "question": "请确认是否执行？", "plan": state["plan"] }) # decision 的值来自 Command(resume=...) 传入的内容 return {"approved": decision.lower() == "yes"} memory = MemorySaver() graph = graph_builder.compile(checkpointer=memory) config = {"configurable": {"thread_id": "task_001"}} # 第一次运行：执行到 interrupt 处暂停 for event in graph.stream(initial_state, config): if "__interrupt__" in event: print("图已暂停，等待人类确认...") # 恢复执行 for event in graph.stream(Command(resume="yes"), config): ... ``` ### Stage Graph-5：多 Agent 协作 多个 Agent 协作完成复杂任务，主管分配任务，工人执行，最后汇总。 **架构：** ``` 用户问题 → 主管 Agent（分类）→ 路由到对应 Agent → 汇总输出 ├─ 研究员 → 写手 → 审稿人 ├─ 通用 Agent └─ ... ``` **知识要点：** - **主管 Agent**：分析问题，决定任务类型（research / write / review / general） - **专业 Agent**：每个节点是一个专业 Agent（研究员、写手、审稿人） - **多级路由**：主管分类后路由，研究员完成后再次路由 - **检查点**：`MemorySaver` 保存执行历史，支持多轮对话 - **State 设计**：包含各 Agent 的结果字段，支持结果传递 ```python # 多 Agent 协作核心代码 graph_builder.add_conditional_edges( "supervisor", route_task, {"general": "general", "researcher": "researcher", "reviewer": "reviewer"} ) graph_builder.add_conditional_edges( "researcher", after_research, {"writer": "writer", "reviewer": "reviewer"} ) graph_builder.add_edge("writer", "reviewer") graph_builder.add_edge("reviewer", "merge") graph_builder.add_edge("merge", END) ``` ### Stage Graph 运行说明 ```bash # 基础概念 python stage_graph_1.py # 条件分支 python stage_graph_2.py # ReAct Agent python stage_graph_3.py # Human-in-the-loop python stage_graph_4.py # 多 Agent 协作 python stage_graph_5.py ``` --- ## LangGraph MCP 学习阶段 ### Stage MCP-1：MCP 服务器基础 使用 `FastMCP` 创建 MCP 服务器，暴露工具（Tools）、资源（Resources）和提示词模板（Prompts）。 **知识要点：** - **MCP（Model Context Protocol）**：模型上下文协议，标准化 AI 模型与外部工具/数据源的交互方式 - **FastMCP**：快速创建 MCP 服务器的高级封装类 - **工具（Tools）**：可执行的函数，客户端可以调用（如计算、查询） - `@mcp.tool()` 装饰器：将函数注册为 MCP 工具 - **docstring 的关键作用**：Agent 根据**文档字符串**判断何时调用该工具 - **资源（Resources）**：只读数据，客户端可以读取但不能修改 - `@mcp.resource("uri://pattern")` 装饰器：注册资源端点 - URI 模式支持参数化：`greeting://{name}` - **提示词模板（Prompts）**：预定义的 prompt，客户端可以直接获取使用 - `@mcp.prompt()` 装饰器：注册提示词模板 - **传输方式**： - `stdio`：标准输入输出，适合本地运行 - `streamable-http`：HTTP 协议，适合远程服务 ```python # mcp_math_server.py 核心代码 from mcp.server.fastmcp import FastMCP mcp = FastMCP("Math") @mcp.tool() def add(a: int, b: int) -> int: """计算两个整数的和""" return a + b @mcp.resource("config://math") def get_config() -> str: """返回数学服务器的配置信息""" return "Math MCP Server v1.0" @mcp.prompt() def math_assistant_prompt(question: str) -> str: """数学助手的提示词模板""" return f"请解答：{question}" if __name__ == "__main__": mcp.run(transport="stdio") ``` ### Stage MCP-2：MCP 客户端集成（单服务器） 使用 LangGraph 的 `create_react_agent` 连接单个 MCP 服务器，调用其中的工具。 **知识要点：** - `StdioServerParameters`：配置 MCP 服务器连接参数（命令、参数、环境变量） - `stdio_client`：异步上下文管理器，自动启动 MCP 服务器进程并建立通信 - `ClientSession`：MCP 客户端会话，负责与服务器进行协议交互 - `session.initialize()`：初始化连接（握手），必须在使用前调用 - `load_mcp_tools()`：将 MCP 工具转换为 LangChain 工具格式，供 Agent 使用 - **资源与提示词的发现**： - `session.list_resources()`：列出服务器提供的所有资源 - `session.list_prompts()`：列出服务器提供的所有提示词模板 - `session.read_resource(uri)`：读取指定资源的内容 - **异步执行**：整个流程使用 `asyncio`，支持异步调用 ```python # stage_mcp_single.py 核心代码 from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools from langgraph.prebuilt import create_react_agent server_params = StdioServerParameters( command="python", args=["mcp_math_server.py"], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 加载 MCP 工具为 LangChain 格式 tools = await load_mcp_tools(session) # 创建 Agent agent = create_react_agent(model=llm, tools=tools, prompt=prompt) # 调用 Agent result = await agent.ainvoke({"messages": [("human", "3加5等于多少？")]}) ``` ### Stage MCP-3：多 MCP 服务器集成 使用 `MultiServerMCPClient` 同时连接多个 MCP 服务器，实现跨服务器的工具调用。 **知识要点：** - **MultiServerMCPClient**：管理多个 MCP 服务器连接的客户端 - **配置格式**：字典形式，每个服务器一个配置项 - `stdio` 模式：指定 `command` 和 `args`，客户端自动启动服务器 - `streamable_http` 模式：指定 `url`，需手动启动服务器 - **身份认证**：远程服务可通过 `headers` 传递 API Key - **client.get_tools()**：汇总所有服务器的工具，自动去重和合并 - **混合调用**：Agent 可根据需要调用任意服务器的工具，无需关心工具来源 - **多轮对话**：通过传递历史消息列表实现上下文连续的对话 ```python # stage_mcp_multi.py 核心代码 from langchain_mcp_adapters.client import MultiServerMCPClient client = MultiServerMCPClient({ "math": { "command": "python", "args": ["mcp_math_server.py"], "transport": "stdio", }, "weather": { "url": "http://127.0.0.1:8000/mcp", "transport": "streamable_http", }, "dream": { "url": "https://open.bigmodel.cn/api/mcp-broker/proxy/dream-interpretation/mcp", "transport": "streamable_http", "headers": {"Authorization": f"Bearer {API_KEY}"}, }, }) async with client: tools = await client.get_tools() agent = create_react_agent(model=llm, tools=tools, prompt=prompt) ``` ### Stage MCP 运行说明 **数学服务器（stdio，自动启动）：** ```bash # 单服务器测试 python stage_mcp_single.py # 多服务器测试（数学服务器会自动启动） python stage_mcp_multi.py ``` **天气服务器（streamable-http，需手动启动）：** ```bash # 终端1：启动天气服务器 python mcp_weather_server.py # 终端2：启动多服务器测试 python stage_mcp_multi.py ``` --- ## 综合实战：智能助手系统 整合所有知识点，构建一个完整的智能助手系统。 ### 架构设计 ``` 用户问题 ↓ 主管 Agent（分类） ├─ math ──────→ 数学 Agent（MCP 工具 + ReAct 循环） ├─ research ──→ 研究员 → 写手 → 人工审批 ─┬─ 通过 → 汇总 │ └─ 退回 → 修改 → 重新审批 └─ chat ──────→ 闲聊 Agent ↓ 汇总节点 ↓ 最终输出 ``` ### 整合的知识点 | 模块 | 使用的知识 | |------|-----------| | **State** | TypedDict + Annotated[list, add_messages] 追加模式 | | **Node** | 主管、数学Agent、研究员、写手、审稿人、审批、修改、汇总 | | **Edge** | 条件分支路由、循环边 | | **MCP** | MultiServerMCPClient 连接多个服务器 | | **Memory** | MemorySaver 检查点保存状态 | | **Human-in-the-loop** | interrupt() + Command(resume) 人工审批 | | **ReAct** | 数学Agent 手动实现思考→行动→观察循环 | | **多Agent** | 主管分配 + 专业Agent执行 + 汇总 | ### 核心代码结构 ```python # 1. MCP 服务器配置 mcp_client = MultiServerMCPClient({ "math": {"command": "python", "args": ["mcp_math_server.py"], "transport": "stdio"}, "weather": {"url": "http://127.0.0.1:8000/mcp", "transport": "streamable_http"}, }) # 2. State 定义 class State(TypedDict): messages: Annotated[list, add_messages] question: str task_type: str approved: bool retry_count: int final_output: str ... # 3. 节点定义 def supervisor_node(state): ... # 主管分类 def math_agent_node(state): ... # 数学Agent（MCP + ReAct） def researcher_node(state): ... # 研究员 def writer_node(state): ... # 写手 def human_review_node(state): ... # 人工审批（interrupt） def revise_node(state): ... # 根据反馈修改 def merge_node(state): ... # 汇总 # 4. 条件路由 graph_builder.add_conditional_edges("supervisor", route_task, {...}) graph_builder.add_conditional_edges("human_review", check_approval, {...}) # 5. 循环边 graph_builder.add_edge("revise", "human_review") # 修改后重新审批 # 6. 编译（带检查点） memory = MemorySaver() graph = graph_builder.compile(checkpointer=memory) ``` ### 运行说明 ```bash # 1. 启动天气服务器（可选） python ../langGraph_mcp/mcp_weather_server.py # 2. 运行智能助手 python main.py ``` --- ## LangChain 知识图谱 ### 整体架构 ```mermaid graph TB subgraph "应用层" Agent["Agent 智能体"] Chain["Chain 处理链"] end subgraph "核心模块" LLM["LLM/ChatModel
大语言模型"] Prompt["PromptTemplate
提示词模板"] Parser["OutputParser
输出解析器"] end subgraph "记忆模块" Memory["Memory
对话记忆"] History["ChatMessageHistory
消息历史"] end subgraph "数据模块" Loader["DocumentLoader
文档加载器"] Splitter["TextSplitter
文本分割器"] Embedding["Embeddings
向量嵌入"] VectorStore["VectorStore
向量数据库"] Retriever["Retriever
检索器"] end subgraph "工具模块" Tool["Tool
工具"] Toolkit["Toolkit
工具集"] end LLM --> Chain Prompt --> Chain Parser --> Chain Memory --> Chain Retriever --> Chain Tool --> Agent LLM --> Agent Loader --> Splitter Splitter --> Embedding Embedding --> VectorStore VectorStore --> Retriever ``` ### 模块关系图 ```mermaid graph LR subgraph "输入处理" A[用户输入] --> B[PromptTemplate] B --> C[格式化消息] end subgraph "模型调用" C --> D[ChatModel] D --> E[AIMessage] end subgraph "输出处理" E --> F[OutputParser] F --> G[结构化输出] end subgraph "记忆存储" D --> H[Memory] H --> I[历史记录] I --> B end subgraph "外部知识" J[文档] --> K[Loader] K --> L[Splitter] L --> M[Embedding] M --> N[VectorStore] N --> O[Retriever] O --> B end subgraph "工具调用" P[Tool] --> Q[Agent] Q --> D D --> Q end ``` ### 核心概念详解 #### 1. 模型层（Models） | 概念 | 说明 | 本项目示例 | |------|------|-----------| | **LLM** | 纯文本输入输出的基础大模型 | - | | **ChatModel** | 对话格式的聊天模型 | `ChatOpenAI` | | **Embedding** | 文本转向量的嵌入模型 | `embedding-3` | ```python # 模型调用流程 llm = ChatOpenAI(model="glm-4-flash") response = llm.invoke("Hello") # → AIMessage ``` #### 2. 提示词层（Prompts） | 概念 | 说明 | 关键类 | |------|------|--------| | **PromptTemplate** | 纯文本模板 | `PromptTemplate` | | **ChatPromptTemplate** | 对话消息模板 | `ChatPromptTemplate` | | **MessagesPlaceholder** | 历史消息占位符 | `MessagesPlaceholder` | ```python # 模板变量替换 prompt = ChatPromptTemplate.from_messages([ ("system", "你是{role}助手"), ("human", "{question}") ]) messages = prompt.invoke({"role": "幽默", "question": "为什么天是蓝的？"}) ``` #### 3. 输出解析层（Output Parsers） | 概念 | 说明 | 适用场景 | |------|------|---------| | **StrOutputParser** | 转为纯字符串 | 简单文本输出 | | **JsonOutputParser** | 转为JSON格式 | 结构化数据 | | **PydanticOutputParser** | 转为Pydantic对象 | 强类型校验 | #### 4. 记忆层（Memory） | 概念 | 说明 | 存储方式 | |------|------|---------| | **ChatMessageHistory** | 原始历史存储 | 内存/数据库 | | **RunnableWithMessageHistory** | 自动管理包装器 | 回调函数 | | **session_id** | 会话隔离标识 | 字符串 | ```python # 记忆管理流程 store = {} def get_history(session_id): if session_id not in store: store[session_id] = InMemoryChatMessageHistory() return store[session_id] chain_with_memory = RunnableWithMessageHistory(chain, get_history) ``` #### 5. 数据层（Data） ```mermaid graph LR A[DocumentLoader
加载文档] --> B[TextSplitter
分割文本] B --> C[Embeddings
向量化] C --> D[VectorStore
存储向量] D --> E[Retriever
检索相关文档] style A fill:#e1f5fe style B fill:#e1f5fe style C fill:#e1f5fe style D fill:#e1f5fe style E fill:#e1f5fe ``` | 概念 | 说明 | 常用实现 | |------|------|---------| | **DocumentLoader** | 加载各种格式文档 | `TextLoader`, `PyPDFLoader` | | **TextSplitter** | 文本分块处理 | `RecursiveCharacterTextSplitter` | | **Embeddings** | 文本转向量 | `OpenAIEmbeddings`, `HuggingFaceEmbeddings` | | **VectorStore** | 向量数据库 | `Chroma`, `FAISS`, `Pinecone` | | **Retriever** | 相似度检索 | `vectorstore.as_retriever()` | #### 6. 链式调用（LCEL） ```python # LCEL 核心语法 chain = prompt | llm | parser # 分支结构 from langchain_core.runnables import RunnableParallel branch = RunnableParallel( context=retriever | format_docs, question=RunnablePassthrough() ) # 带记忆的链 chain_with_memory = RunnableWithMessageHistory( chain, get_history, input_messages_key="input", history_messages_key="history" ) ``` #### 7. 工具与代理（Tools & Agents） ```mermaid graph TB A[用户请求] --> B[Agent] B --> C{需要工具?} C -->|是| D[选择Tool] D --> E[执行Tool] E --> F[获取结果] F --> B C -->|否| G[直接回答] subgraph "Tool类型" H[calculator] I[word_count] J[get_current_time] end D --> H D --> I D --> J style B fill:#fff3e0 style D fill:#e8f5e9 ``` | 概念 | 说明 | 关键组件 | |------|------|---------| | **@tool** | 函数转工具装饰器 | `langchain_core.tools` | | **Agent** | 自主决策的智能体 | `create_react_agent` | | **ReAct** | 思考-行动-观察循环 | Thought → Action → Observation | | **Tool Description** | 工具功能描述 | docstring | ### 完整数据流 ```mermaid sequenceDiagram participant U as 用户 participant P as PromptTemplate participant M as ChatModel participant O as OutputParser participant Mem as Memory participant R as Retriever U->>P: 发送问题 P->>P: 填充模板变量 Note over P: 插入历史消息 Mem-->>P: 提供历史记录 R-->>P: 提供相关文档 P->>M: 发送格式化消息 M->>M: 生成回答 M->>O: 返回AIMessage O->>O: 解析为字符串 O->>Mem: 保存对话历史 O->>U: 返回最终结果 ``` ### 包结构对照 | 包名 | 用途 | 本项目使用 | |------|------|-----------| | `langchain-core` | 核心抽象和基础组件 | PromptTemplate, StrOutputParser | | `langchain-openai` | OpenAI API封装 | ChatOpenAI, OpenAIEmbeddings | | `langchain-community` | 第三方集成 | TextLoader, Chroma | | `langchain-text-splitters` | 文本分割工具 | RecursiveCharacterTextSplitter | | `langgraph` | Agent编排框架 | create_react_agent | ## 技术栈 - **LangChain** - LLM 应用开发框架 - **LangGraph** - Agent 编排框架 - **MCP** - 模型上下文协议 - **ChromaDB** - 向量数据库 - **智谱 GLM-4-Flash** - 大语言模型 --- ## LangGraph 知识图谱 ### LangGraph 核心概念 ```mermaid graph TB subgraph "三要素" State["State
状态（共享数据）"] Node["Node
节点（处理函数）"] Edge["Edge
边（执行顺序）"] end subgraph "核心能力" Conditional["条件分支
add_conditional_edges"] Loop["循环
act → think"] Checkpoint["检查点
MemorySaver"] HumanLoop["Human-in-the-loop
interrupt + Command"] end subgraph "应用模式" SimpleChain["简单链
START → A → B → END"] Branch["分支
A → 条件 → B/C/D"] ReActAgent["ReAct Agent
think ⇄ act"] MultiAgent["多 Agent
主管 → 工人 → 汇总"] end State --> Node Node --> Edge Edge --> Conditional Edge --> Loop Checkpoint --> HumanLoop Conditional --> Branch Loop --> ReActAgent ``` ### LangGraph 与 LangChain 对比 | 特性 | LangChain Chain | LangGraph | |------|----------------|-----------| | **执行流程** | 线性（A→B→C） | 任意图结构 | | **循环** | ❌ 不支持 | ✅ 支持 | | **条件分支** | 有限支持 | 完整支持 | | **状态管理** | 外部管理 | 内置 State | | **Human-in-the-loop** | 不支持 | ✅ 支持 | | **检查点** | 不支持 | ✅ 支持 | ### LangGraph 执行流程 ```mermaid sequenceDiagram participant U as 用户 participant G as Graph participant S as State participant N1 as Node A participant N2 as Node B participant CP as Checkpoint U->>G: invoke(initial_state) G->>S: 加载 State G->>N1: 执行 Node A N1->>S: 更新 State G->>CP: 保存检查点 G->>N2: 执行 Node B N2->>S: 更新 State G->>CP: 保存检查点 G->>U: 返回最终 State ``` --- ## LangGraph MCP 知识图谱 ### MCP 架构概览 ```mermaid graph TB subgraph "MCP 客户端" Agent["LangGraph Agent"] Client["MCP Client"] end subgraph "MCP 服务器" S1["Math Server
(stdio)"] S2["Weather Server
(streamable-http)"] S3["Dream Server
(远程 HTTP)"] end subgraph "MCP 能力" Tools["Tools
可执行函数"] Resources["Resources
只读数据"] Prompts["Prompts
提示词模板"] end Agent --> Client Client --> S1 Client --> S2 Client --> S3 S1 --> Tools S1 --> Resources S1 --> Prompts S2 --> Tools S3 --> Tools ``` ### MCP 传输方式对比 | 传输方式 | 适用场景 | 启动方式 | 通信协议 | |---------|---------|---------|---------| | **stdio** | 本地服务 | 客户端自动启动 | 标准输入输出 | | **streamable-http** | 远程服务 | 手动启动 | HTTP | ### MCP 工具调用流程 ```mermaid sequenceDiagram participant U as 用户 participant A as Agent participant C as MCP Client participant S as MCP Server participant T as Tool U->>A: 发送请求 A->>A: 分析需要调用工具 A->>C: 调用工具 C->>S: 发送 MCP 请求 S->>T: 执行工具函数 T->>S: 返回结果 S->>C: 返回 MCP 响应 C->>A: 返回工具结果 A->>A: 整合结果生成回答 A->>U: 返回最终答案 ``` ### MCP 核心概念 | 概念 | 说明 | 用途 | |------|------|------| | **Tools** | 可执行的函数 | 计算、查询、操作 | | **Resources** | 只读数据端点 | 获取配置、读取数据 | | **Prompts** | 预定义的提示词模板 | 标准化对话流程 | | **Transport** | 通信方式 | stdio / streamable-http | ### MCP 与 LangChain 集成 ```python # 完整集成流程 from langchain_mcp_adapters.tools import load_mcp_tools from langchain_mcp_adapters.client import MultiServerMCPClient from langgraph.prebuilt import create_react_agent # 1. 配置 MCP 服务器 client = MultiServerMCPClient({...}) # 2. 加载工具 tools = await client.get_tools() # 3. 创建 Agent agent = create_react_agent(model=llm, tools=tools, prompt=prompt) # 4. 调用 result = await agent.ainvoke({"messages": [("human", "问题")]}) ```