---

---

1 摘要

长期记忆 + RAG + MCP + LangGraph = 强大代理式AI的关键通过LangGraph、MCP、RAG与长期记忆构建多代理聊天机器人；主管协调网络搜索与文件操作，MCP充当工具执行桥梁，并配合配置与会话持久化，实现更稳定、可扩展的端到端自动化。

---

在这篇文章中，我提供了一个超快速教程，展示如何使用LangGraph、MCP、RAG和长期记忆创建多代理聊天机器人，为您的企业或个人使用构建强大的代理聊天机器人。

这个AI代理是我构建过的最强大的代理。您可以使用RAG通过查找字典和其他文档中的信息来回答问题。

就像我们通过查阅书籍或互联网来回答困难问题一样，MCP服务器充当AI的"手脚"。它使用人类类比，即使大脑（AI代理）思考"给我拿那本书"，如果手（MCP）不实际移动，书就无法被取回。MCP服务器充当将AI的"思想"转化为实际"行动"的桥梁。

代理的一个大问题是通信。一开始运行良好，但我使用得越多，情况就越糟。它没有从过去的错误中学习，而是不断重复同样的错误。但通过这个强大的AI代理，我们解决了AI社区的所有主要痛点。

如果这是您第一次看我的内容，我强烈建议查看我之前的文章。我制作了一个关于最新AI技术的视频，在AI社区引起了广泛关注。

让我通过一个实时聊天机器人的快速演示来说明我的意思。

在我提问之前，我将加载一个记忆，其中包含过去的对话，然后我会问聊天机器人一个问题：“查找关于大型语言模型的最新信息”

如果您查看代理如何生成输出，您会看到我们正在构建的多AI代理系统使用Google的生成式AI模型（Gemini），是一个基础的多AI代理系统，其中网络搜索代理和文件操作代理在与用户交互并向专业代理发出指令的管理代理的自主协作下工作。

正如人类在团队中工作一样，AI代理也一起工作，利用各自的专业领域。

游戏中的三个代理是：

• 主管（管理者）：团队的大脑。他们理解用户的指令，规划整个任务，决定哪个工作者应该做什么，何时做，并给出精确的指令。
• 网络冲浪者（工作者）：专业信息收集者。按照主管的指示使用关键词搜索网络，收集必要信息，并报告结果。
• 文件操作员（工作者）：组织和记录保存的大师。按照主管的指示将信息写入文件并从现有文件中读取。

通过让这些代理一起工作，结合网络搜索和文件操作的复杂任务，如"查找有关任何产品的信息并将其编译成CSV文件"，可以通过单个用户命令自动执行。

在这个例子中，专业代理执行的任务仅限于网络搜索和文件操作。但是，通过增加专业代理的数量并为它们分配角色和工具，可以根据用例灵活地扩展功能。

例如，通过利用MCP，可以实现以下作为工作者的代理，从而实现更复杂和实用任务的自动化。

2 让我们开始编码：

现在让我们一步步探索并揭示如何创建LangGraph、RAG、MCP和长期记忆的答案。我们将安装支持该模型的库。为此，我们将执行pip安装需求。

我想告诉您，我在这里分享的代码只是我代码的一部分。如果您想要完整的文件夹，可以在我的Patreon上找到。这段代码花费了我相当多的时间，这个代理是我构建的最强大和最先进的代理。所有技术都在我的文件夹中。

pip install -r requirements,txt

下一步是常规步骤：我们将导入相关库，其重要性将随着我们的进行而变得明显，并执行一些基本配置。

import streamlit as st
import json
import os
import logging
import uuid
import asyncio
import warnings
from dotenv import load_dotenv, find_dotenv
from typing import List, TypedDict


from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage, ToolMessage, messages_to_dict, messages_from_dict
from langchain_core.utils.function_calling import convert_to_openai_function
from langchain_google_genai import ChatGoogleGenerativeAI
from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_mcp_adapters.client import MultiServerMCPClient

**mcp_config.json**

代理使用"工具"执行特定"操作"，如网络搜索或文件操作。该系统通过称为Model-Context-Protocol（MCP）的机制使用工具。mcp_config.json文件是定义要启动哪些工具以及如何启动的配置文件。

在项目文件夹下直接创建一个mcp_config.json文件，并在其中写入以下内容。

• **web-search**：用于执行网络搜索的工具服务器设置。npx启动Playwright MCP服务器。
• **file-system**：用于读取和写入文件的工具服务器设置。
• args/path/to/your/project/multi-agent-system/output请将部分更改为适合您环境的路径。这是文件操作代理允许读取和写入文件的文件夹的绝对路径。例如，在项目中创建一个output文件夹并指定其路径。请注意，如果指定路径不正确，将无法执行文件操作。

{
    "mcpServers": {
      "web-search": {
        "command": "npx",
        "args": [
          "@playwright/mcp@latest"
        ],
        "transport": "stdio"
      },
      "file-system": {
        "command": "npx",
        "args": [
          "-y",
          "@modelcontextprotocol/server-filesystem",
          "/path/to/your/project/multi-agent-system/output"
        ],
        "transport": "stdio"
      }
    }
}

让我们使用sanitize_schema递归遍历字典和列表，删除不需要的键如additionalProperties和$schema，并通过选择第一个非NULL值并将其大写来规范化可以是列表的type字段，对每个嵌套值应用相同的清理；我添加了save_conversation(session_id, messages)，它防止空输入，在CONVERSATION_HISTORY_DIR下构建路径，使用messages_to_dict将消息对象转换为普通字典，并使用ensure_ascii=False和indent=2将其作为UTF-8 JSON写入；

我实现了load_conversation(session_id)，如果文件丢失则返回空列表，否则加载JSON并使用messages_from_dict将其转回消息对象，在JSONDecodeError或TypeError时返回空列表以优雅地失败；

我构建了list_conversations()来扫描目录中的.json文件，提取每个文件的修改时间，加载其消息，选择第一个不是内部指令的人类消息作为简短标题（在40个字符处用省略号截断），并收集{id, title, mtime}条目，同时跳过有错误的文件，最后按修改时间降序排序，并添加了delete_conversation(session_id)以安全地删除相应的JSON文件（如果存在）。

def sanitize_schema(item):
    """Sanitize MCP tool schema for LangChain compatibility"""
    if isinstance(item, dict):
        item.pop('additionalProperties', None)
        item.pop('$schema', None)
        if 'type' in item and isinstance(item['type'], list):
            non_null_types = [t for t in item['type'] if str(t).upper() != 'NULL']
            item['type'] = str(non_null_types[0]).upper() if non_null_types else None
        for key, value in item.items():
            item[key] = sanitize_schema(value)
    elif isinstance(item, list):
        return [sanitize_schema(i) for i in item]
    return item

def save_conversation(session_id: str, messages: List[BaseMessage]):
    """Save conversation to JSON file"""
    if not session_id or not messages:
        return
    file_path = os.path.join(CONVERSATION_HISTORY_DIR, f"{session_id}.json")
    with open(file_path, "w", encoding="utf-8") as f:
        json.dump(messages_to_dict(messages), f, ensure_ascii=False, indent=2)

def load_conversation(session_id: str) -> List[BaseMessage]:
    """Load conversation from JSON file"""
    file_path = os.path.join(CONVERSATION_HISTORY_DIR, f"{session_id}.json")
    if not os.path.exists(file_path):
        return []
    with open(file_path, "r", encoding="utf-8") as f:
        try:
            data = json.load(f)
            return messages_from_dict(data)
        except (json.JSONDecodeError, TypeError):
            return []

def list_conversations() -> List[dict]:
    """Get list of saved conversations"""
    conversations = []
    for filename in os.listdir(CONVERSATION_HISTORY_DIR):
        if filename.endswith(".json"):
            session_id = filename[:-5]
            file_path = os.path.join(CONVERSATION_HISTORY_DIR, filename)
            try:
                mtime = os.path.getmtime(file_path)
                messages = load_conversation(session_id)
                # Get first user message as conversation title
                first_user_message = next(
                    (m.content for m in messages 
                     if isinstance(m, HumanMessage) and m.additional_kwargs.get("role") != "internal_instruction"), 
                    "New conversation"
                )
                title = first_user_message[:40] + "..." if len(first_user_message) > 40 else first_user_message
                conversations.append({"id": session_id, "title": title, "mtime": mtime})
            except Exception:
                continue
    conversations.sort(key=lambda x: x["mtime"], reverse=True)
    return conversations

def delete_conversation(session_id: str):
    """Delete conversation file"""
    file_path = os.path.join(CONVERSATION_HISTORY_DIR, f"{session_id}.json")
    if os.path.exists(file_path):
        os.remove(file_path)

我创建了一对辅助函数来启动工作代理和主管代理，它们在多代理设置中协调任务：我编写了create_worker，它接受一个语言模型、工具列表和系统提示，然后构建一个ChatPromptTemplate，包含系统角色消息和过去对话历史的占位符，最后返回一个将此提示与绑定到其工具的LLM连接的管道；

然后我构建了create_supervisor来协调工作者，定义了一个长系统提示，解释了管理者的责任——分析用户请求，将其分解为子任务，决定下一个工作者的行动，传递先前结果以保持连续性，完成时结束，工作者失败时重试——同时还动态列出可用的工作者；

我创建了一个output_schema，强制主管以结构化对象响应，包含next字段（工作者名称或FINISH）和content字段（指令或最终用户响应）；最后，我为主管构建了一个ChatPromptTemplate，然后使用bind_tools和tool_choice="supervisor_decision"将LLM与此模式绑定，返回提示和配置的LLM，以便它们可以一起驱动代理循环。

def create_worker(llm: ChatGoogleGenerativeAI, tools: list, system_prompt: str):
    """Create a worker agent with specific role"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        MessagesPlaceholder(variable_name="messages"),
    ])
    return prompt | llm.bind_tools(tools)

def create_supervisor(llm: ChatGoogleGenerativeAI, worker_names: List[str]):
    """Create supervisor that manages tasks and directs workers"""
    system_prompt = (
        "You are the manager of an AI team. Your job is to supervise your worker team to achieve user requests.\n"
        "Carefully review the entire conversation history (user requests, workers' previous results, etc.).\n\n"
        "Follow these steps:\n"
        "1. **Task Analysis**: Consider the steps needed to fulfill the user's request. Multiple workers may need to collaborate. "
        "For example, 'WebSurfer' collects information that 'FileOperator' writes to a file.\n"
        "2. **Decide Next Action**: Based on analysis, determine the next action:\n"
        "   - **Worker Instructions**: When assigning a task to a worker, specify the worker name in `next` and detailed instructions in `content`. "
        "**Important: Include previous workers' output results in the next worker's instructions.** This enables information flow between workers.\n"
        "   - **Direct User Response**: When all tasks are complete or for simple responses not requiring workers, "
        "set `next` to 'FINISH' and provide the final response in `content`.\n"
        "   - **Recovery from Failure**: If a worker fails, review conversation history, modify instructions and retry, or try a different approach.\n\n"
        f"Available workers:\n{chr(10).join(f'- {name}' for name in worker_names)}"
    )
    
    output_schema = {
        "title": "supervisor_decision",
        "type": "object",
        "properties": {
            "next": {"type": "string", "description": f"Next worker name ({', '.join(worker_names)} or FINISH)"},
            "content": {"type": "string", "description": "Instructions for worker or final response to user"}
        },
        "required": ["next", "content"]
    }
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        MessagesPlaceholder(variable_name="messages"),
    ])
    
    llm_with_tool = llm.bind_tools(tools=[output_schema], tool_choice="supervisor_decision")
    return prompt, llm_with_tool

我开发了一个supervisor_node函数，作为主管代理的大脑，指导工作流程并记录其推理：我首先记录主管节点正在运行，然后通过将supervisor_prompt管道连接到supervisor_llm构建链，并使用当前对话历史（state["messages"]）调用它；

我从响应中提取usage_metadata来计算并记录运行主管模型的成本；然后我从第一个工具调用中提取主管的结构化决策，捕获content（指令或最终消息）和next动作（工作者名称或FINISH），并打印带有这些值的调试语句；我创建了一个反映主管推理的AIMessage，在指导工作者时将其格式化为指令，在完成时作为普通内容。

如果决策不是FINISH，我生成了一个标记为role="internal_instruction"的内部HumanMessage传递给工作者，返回一个更新的状态，包含主管的评论和工作者的指令，以及next动作，但如果决策是FINISH，我只附加主管的评论并返回状态，其中next="FINISH"

最后，我用try/except块包装所有内容，以捕获错误，记录它们，并优雅地返回带有next="FINISH"的错误AIMessage，这样流程不会中断。

def supervisor_node(state: AgentState):
        """Supervisor node that decides what to do next and records its thinking"""
        logger.info("--- Supervisor Node ---")
        
        try:
            chain = supervisor_prompt | supervisor_llm
            response_message = chain.invoke({"messages": state["messages"]})
            
            
            usage_metadata = response_message.response_metadata.get("usage_metadata", {})
            costs = calculate_cost(usage_metadata, supervisor_model_name)
            logger.info(f"Cost (Supervisor): ${costs['total']:.6f}")
            
            
            tool_call = response_message.tool_calls[0]
            supervisor_output = tool_call['args']
            logger.info(f"Supervisor Decision: {supervisor_output}")
            
            content = supervisor_output.get("content", "")
            next_action = supervisor_output.get("next", "FINISH")

            print(f"DEBUG Supervisor Decision: next='{next_action}', content='{content}'")
            
            
            supervisor_comment_content = content if next_action == "FINISH" else f"【Instruction to {next_action}】\n{content}"
            supervisor_comment = AIMessage(content=supervisor_comment_content, name="Supervisor")
            
            if next_action != "FINISH":
                
                instruction_for_worker = HumanMessage(
                    content=content, 
                    additional_kwargs={"role": "internal_instruction"}
                )
                return {
                    "messages": state["messages"] + [supervisor_comment, instruction_for_worker], 
                    "next": next_action
                }
            else:
                return {
                    "messages": state["messages"] + [supervisor_comment], 
                    "next": next_action
                }
        except Exception as e:
            logger.error(f"Supervisor error: {e}")
            error_response = AIMessage(content=f"I encountered an error while processing your request: {str(e)}", name="Supervisor")
            return {"messages": state["messages"] + [error_response], "next": "FINISH"}

我创建了一个worker_node及其支持路由逻辑，让工作者执行任务，调用工具，并将结果反馈到多代理循环中，具有强大的错误处理：我从state["next"]中查找分配的工作者名称，记录哪个工作者正在运行，并使用对话历史调用工作者，同时强制执行10的递归限制；我添加了调试打印，以显示工作者响应是否包含工具调用以及触发了哪些工具；我在安全的try/except中包装了成本计算，在usage_metadata可用时记录模型的成本，否则发出警告；

我检查响应是否携带有意义的content或tool_calls，如果两者都不存在，我用一个道歉的后备AIMessage替换它，这样系统永远不会返回空输出；我还确保响应携带正确的工作者名称，并将其附加到返回状态中的消息历史；我用try/except包围整个块，这样任何异常都会被捕获并转化为工作者的错误消息，而不是崩溃。

然后我创建了_tool_node作为ToolNode(tools)实例，并将其包装在一个异步custom_tool_node中，通过ainvoke执行工具调用，并将结果附加回状态；最后，我定义了路由辅助函数：after_worker_router，它检查工作者的最后一条消息是否包含工具调用，并路由到"tools"或返回到"supervisor"，以及supervisor_router，它检查主管的next决策，并路由到指定的工作者或如果不需要进一步操作则路由到END。

def worker_node(state: AgentState):
        """Worker node that executes assigned tasks with error handling"""
        worker_name = state["next"]
        worker = workers[worker_name]
        logger.info(f"--- Worker Node: {worker_name} ---")
        
        try:

            response = worker.invoke({"messages": state['messages']}, {"recursion_limit":10})
            
            print(f"DEBUG Worker {worker_name} response has tool_calls: {hasattr(response, 'tool_calls') and bool(response.tool_calls)}")
            if hasattr(response, 'tool_calls') and response.tool_calls:
                print(f"DEBUG Tool calls: {[tc['name'] for tc in response.tool_calls]}")
            
            try:
                usage_metadata = response.response_metadata.get("usage_metadata", {})
                costs = calculate_cost(usage_metadata, worker_model_name)
                logger.info(f"Cost ({worker_name}): ${costs['total']:.6f}")
            except:
                logger.warning("Could not calculate costs")
            
            
            has_content = bool(response.content)
            has_tool_calls = hasattr(response, 'tool_calls') and bool(response.tool_calls)
            
            if not has_content and not has_tool_calls:
                error_message = "I apologize, but I encountered a technical issue and couldn't complete the task. Please try rephrasing your request."
                response = AIMessage(content=error_message, name=worker_name)
            
            
            response.name = worker_name
            return {"messages": state["messages"] + [response]}
            
        except Exception as e:
            logger.error(f"Worker {worker_name} exception: {e}")
            error_message = "I encountered an error while processing your request. Please try again or rephrase your question."
            error_response = AIMessage(content=error_message, name=worker_name)
            return {"messages": state["messages"] + [error_response]}
    
    
    _tool_node = ToolNode(tools)
    
    async def custom_tool_node(state: AgentState):
        """Node that executes tools called by workers"""
        tool_results = await _tool_node.ainvoke(state)
        return {"messages": state["messages"] + tool_results["messages"]}
    
    
    
    def after_worker_router(state: AgentState) -> str:
        """Router that decides where to go after worker execution"""
        last_message = state["messages"][-1]
        if hasattr(last_message, "tool_calls") and last_message.tool_calls:
            return "tools"
        return "supervisor"
    
    def supervisor_router(state: AgentState) -> str:
        """Router that decides where to go after supervisor decision"""
        next_val = state.get("next")
        if not next_val or next_val == "FINISH":
            return END
        return next_val

我创建了一个工作流编排图，将主管、工作者和工具连接到一个状态机中：我首先使用AgentState类型初始化一个StateGraph，然后通过循环workers字典为主管、工具和每个工作者动态添加节点；

我为每个工作者使用after_worker_router设置条件边，这样在完成任务后，流程要么路由到"tools"（如果存在工具调用），要么返回到"supervisor"；我定义了从"tools"到"supervisor"的直接边，以确保工具结果始终被审查；然后我使用supervisor_router配置主管的路由，

这样它的决策可以分支到特定的工作者或在任务完成时结束工作流；我通过添加从START到"supervisor"的边将主管标记为入口点，确保所有请求都在其控制下开始；最后，我使用MemorySaver检查点编译工作流，以在步骤之间保持对话状态，返回结果app，并记录图初始化已完成。

workflow = StateGraph(AgentState)
    
    
    workflow.add_node("supervisor", supervisor_node)
    workflow.add_node("tools", custom_tool_node)
    for name in workers:
        workflow.add_node(name, worker_node)
    
    
    for name in workers:
        workflow.add_conditional_edges(
            name,
            after_worker_router,
            {"tools": "tools", "supervisor": "supervisor"}
        )
    
    
    workflow.add_edge("tools", "supervisor")
    
    
    workflow.add_conditional_edges(
        "supervisor",
        supervisor_router,
        {**{name: name for name in workers}, END: END}
    )
    
    
    workflow.add_edge(START, "supervisor")
    
    
    memory = MemorySaver()
    app = workflow.compile(checkpointer=memory)
    
    logger.info("Graph initialization completed.")
    return app

3 结论

AI代理的组合预计将极大地改变我们的工作和商业方式。AI的角色将从当前的"教师"角色戏剧性地转变为"可靠的伙伴"，甚至是"代表我们行动的人"。

以下能力被认为对有效利用AI代理特别重要：

• “提问的力量”：定义问题并向AI提供明确指令和要求的能力。
• “确认和决策的能力”：评估AI生成的结果并做出最终决策的能力。
• “通过多任务分配工作的能力”：适当使用多个AI代理并有效分配任务的能力。