文章详细介绍如何使用LangGraph+MCP+ReactAgent技术组合构建智能代理系统。LangGraph提供智能体框架和执行机制，MCP实现外部服务接入，ReAct Agent支持推理与行动。文章展示了核心代码实现，包括环境配置、处理类和关键点，如checkpointer实现多轮对话记忆，running_tasks实现任务管理。项目提供多种MCP调用方式，支持不同业务场景，为大模型应用开发提供完整解决方案。

引言

LangGraph+MCP+ReactAgent技术组合，构建智能代理（Agent）系统。

LangGraph 提供了智能体的框架和执行机制。 MCP 提供了外部服务的接入能力，支持调用外部工具和数据源。 ReAct Agent 提供了推理和行动的机制，支持智能体的自主决策和任务执行。

---

实现效果

---

核心技术栈

1. LangGraph

---

LangGraph 是一个用于构建和运行智能代理的框架，支持并行、分支和循环执行任务，相比传统的线性执行方式更为高效和灵活 。LangGraph 提供了丰富的功能，如任务调度、记忆管理、工具调用等，支持构建复杂的智能体系统 。

2. MCP（Model Context Protocol）

---

MCP 是一种定义 AI 助手行为的标准协议，用于管理模型状态和交互逻辑。它支持多种服务端工具的集成，例如高德地图 MCP Server，允许开发者通过标准接口调用外部服务 。MCP 与 LangGraph 的集成使得开发者能够轻松地将外部服务集成到智能体中，提升系统的灵活性和可扩展性 。

3. ReAct Agent

---

ReAct Agent 是一种基于推理与行动结合的智能体模型，通过迭代推理和行动来解决复杂问题。ReAct Agent 的核心思想是通过“Thought→Action→Observation”循环迭代，逐步逼近问题答案 。ReAct Agent 与 LangGraph 的结合，使得智能体能够更高效地处理复杂任务，提升推理和决策能力 。

✨ 项目环境

• **Python 3.11+**：主要编程语言
• Sanic：Python WEB服务器框架
• McpHub：MCP集成管理工具
• MCP: MPCP Python客户端
• Langchain: 连接LLM与工具，链式编排任务
• LangGraph: 图结构控制复杂智能体流程

核心代码

• 核心处理类

❝

初始化方法构建大模型组件、注意通义官方api扩展参数不支持

classLangGraphReactAgent:
"""
    基于LangGraph的React智能体，支持多轮对话记忆
    """

def__init__(self):
# 校验并获取环境变量
        required_env_vars = [
"MODEL_NAME",
"MODEL_TEMPERATURE",
"MODEL_BASE_URL",
"MODEL_API_KEY",
"MCP_HUB_COMMON_QA_GROUP_URL",
        ]
for var in required_env_vars:
ifnot os.getenv(var):
raise ValueError(f"Missing required environment variable: {var}")

        self.llm = ChatOpenAI(
            model=os.getenv("MODEL_NAME", "qwen-plus"),
            temperature=float(os.getenv("MODEL_TEMPERATURE", 0.75)),
            base_url=os.getenv("MODEL_BASE_URL", "https://dashscope.aliyuncs.com"),
            api_key=os.getenv("MODEL_API_KEY"),
# max_tokens=int(os.getenv("MAX_TOKENS", 20000)),
            top_p=float(os.getenv("TOP_P", 0.8)),
            frequency_penalty=float(os.getenv("FREQUENCY_PENALTY", 0.0)),
            presence_penalty=float(os.getenv("PRESENCE_PENALTY", 0.0)),
            timeout=float(os.getenv("REQUEST_TIMEOUT", 30.0)),
            max_retries=int(os.getenv("MAX_RETRIES", 3)),
            streaming=os.getenv("STREAMING", "True").lower() == "true",
# 将额外参数通过 extra_body 传递
            extra_body={},
        )

# 构建MCP客户端
       self.client = MultiServerMCPClient(
            {
"mcp-hub": {
"url": os.getenv("MCP_HUB_COMMON_QA_GROUP_URL"),
"transport": "streamable_http",
            },
       })

# 全局checkpointer用于持久化所有用户的对话状态
        self.checkpointer = InMemorySaver()

# 存储运行中的任务
        self.running_tasks = {}



    @staticmethod
def_create_response(
        content: str, message_type: str = "continue", data_type: str = DataTypeEnum.ANSWER.value[0]
    ) -> str:
"""封装响应结构"""
        res = {
"data": {"messageType": message_type, "content": content},
"dataType": data_type,
        }
return"data:" + json.dumps(res, ensure_ascii=False) + "\n\n"

    @staticmethod
defshort_trim_messages(state):
"""
        模型调用前的消息清理的钩子函数
        短期记忆：限制模型调用前的消息数量，只保留最近的若干条消息
        :param state: 状态对象，包含对话消息
        :return: 修剪后的消息列表
        """
        trimmed_messages = trim_messages(
            messages=state["messages"],
            max_tokens=20000,  # 设置更合理的token限制（根据模型上下文窗口调整）
            token_counter=lambda messages: sum(len(msg.content or"") for msg in messages),  # 更准确的token计算方式
            strategy="last",  # 保留最新的消息
            allow_partial=False,
            start_on="human",  # 确保从人类消息开始
            include_system=True,  # 包含系统消息
            text_splitter=None,  # 不使用文本分割器
        )
return {"llm_input_messages": trimmed_messages}

asyncdefrun_agent(
        self, query: str, response, session_id: Optional[str] = None, uuid_str: str = None, user_token=None
    ):
"""
        运行智能体，支持多轮对话记忆
        :param query: 用户输入
        :param response: 响应对象
        :param session_id: 会话ID，用于区分同一轮对话
        :param uuid_str: 自定义ID，用于唯一标识一次问答
        :param user_token:
        :return:
        """

# 获取用户信息 标识对话状态
        user_dict = await decode_jwt_token(user_token)
        task_id = user_dict["id"]
        task_context = {"cancelled": False}
        self.running_tasks[task_id] = task_context

try:
            t02_answer_data = []

            tools = await self.client.get_tools()

# 使用用户会话ID作为thread_id，如果未提供则使用默认值
            thread_id = session_id if session_id else"default_thread"
            config = {"configurable": {"thread_id": thread_id}}

            system_message = SystemMessage(
                content="""
            # Role: 高级AI助手

            ## Profile
            - language: 中文
            - description: 一位具备多领域知识、高度专业性与结构化输出能力的智能助手，专注于提供精准、高效、可信赖的信息服务。
            - background: 基于大规模语言模型训练，融合技术、学术、生活等多维度知识体系，能够适应多种场景下的信息查询与任务处理需求。
            - personality: 严谨、专业、逻辑清晰，注重细节与用户体验，追求信息传递的准确性与表达的简洁性。
            - expertise: 多领域知识整合、结构化内容生成、技术说明、数据分析、编程辅助、语言表达优化等。
            - target_audience: 技术人员、研究人员、学生、内容创作者及各类需要精准信息支持的用户。

            ## Skills

            1. 信息处理与表达
               - 精准应答：确保输出内容准确无误，对不确定信息明确标注「暂未掌握该信息」
               - 结构化输出：根据内容类型采用文本、代码块、列表等多种形式进行清晰表达
               - 语言适配：始终使用用户提问语言进行回应，确保语义一致与文化适配
               - 技术说明：对专业术语、技术原理提供背景信息与详细解释，便于理解

            2. 工具协作与交互
               - 工具调用提示：在需要调用外部工具时明确标注「工具调用」并说明调用目的
               - 操作透明化：在涉及流程性任务时说明步骤与逻辑，增强用户信任与理解
               - 多模态支持：支持文本、代码、数据等多种信息类型的识别与响应
               - 用户反馈整合：根据用户反馈优化输出策略，提升交互质量

            ## Rules

            1. 基本原则：
               - 准确性优先：所有输出内容必须基于可靠知识，不臆测、不虚构
               - 用户导向：围绕用户需求组织内容，避免无关信息干扰
               - 透明性：在涉及工具调用、逻辑推理或数据处理时保持过程透明
               - 可读性：结构清晰、层级分明、排版整洁，便于快速阅读与理解

            2. 行为准则：
               - 语言一致性：始终使用用户提问语言进行回应
               - 技术细节补充：对复杂或专业内容提供背景信息与解释
               - 信息边界明确：对未知或超出能力范围的内容如实说明
               - 风格统一：保持段落、层级、图标风格一致，避免杂乱

            3. 限制条件：
               - 不生成违法、有害或误导性内容
               - 不模拟人类情感或主观判断
               - 不提供医疗、法律等专业建议（除非明确授权）
               - 不处理包含隐私、敏感或机密信息的请求

            ## Workflows

            - 目标: 提供准确、结构清晰、风格统一的高质量回答
            - 步骤 1: 理解用户意图，识别问题类型与需求层次
            - 步骤 2: 检索知识库，组织相关信息，判断是否需要调用工具
            - 步骤 3: 按照格式规范生成内容，进行语言与结构优化
            - 预期结果: 用户获得结构清晰、语言准确、风格统一的专业级回答

            ## OutputFormat

            1. 输出格式类型：
               - format: markdown
               - structure: 分节说明，层级清晰，模块分明
               - style: 专业、简洁、结构化，强调信息密度与可读性
               - special_requirements: 使用Unicode图标增强视觉引导，图标与内容匹配，风格统一

            2. 格式规范：
               - indentation: 使用两个空格缩进
               - sections: 按模块划分，使用标题、列表、加粗等方式增强可读性
               - highlighting: 关键信息使用**加粗**或代码块```
               - icons: 每个主要模块前添加1个相关图标，与文字保留1个空格

            3. 验证规则：
               - validation: 所有输出需符合markdown语法规范
               - constraints: 图标风格统一，层级结构清晰，内容与格式分离
               - error_handling: 若格式错误，自动尝试恢复结构并提示用户

            4. 示例说明：

               1. 示例1：
                  - 标题: 简单问答示例
                  - 格式类型: markdown
                  - 说明: 展示基本问答格式与图标使用规范
                  - 示例内容: |
                      📌 **问题：** 什么是AI？
                      ✅ **回答：** AI（Artificial Intelligence，人工智能）是指由人创造的能够感知环境、学习知识、逻辑推理并执行任务的智能体。

               2. 示例2：
                  - 标题: 代码输出示例
                  - 格式类型: markdown
                  - 说明: 展示代码类输出格式与图标使用
                  - 示例内容: |
                      💻 **Python示例：**
                      ```python
                      def greet(name):
                          print(f"Hello, {name}!")
                      greet("World")
                      ```
                      📌 说明：这是一个简单的Python函数，用于打印问候语。

            ## Initialization
            作为高级AI助手，你必须遵守上述Rules，按照Workflows执行任务，并按照[输出格式]输出。
            """
            )

            agent = create_react_agent(
                model=self.llm,
                tools=tools,
                prompt=system_message,
                checkpointer=self.checkpointer,  # 使用全局checkpointer
                pre_model_hook=self.short_trim_messages,
            )

asyncfor message_chunk, metadata in agent.astream(
                input={"messages": [HumanMessage(content=query)]},
                config=config,
                stream_mode="messages",
            ):
# 检查是否已取消
if self.running_tasks[task_id]["cancelled"]:
await response.write(
                        self._create_response("\n> 这条消息已停止", "info", DataTypeEnum.ANSWER.value[0])
                    )
# 发送最终停止确认消息
await response.write(self._create_response("", "end", DataTypeEnum.STREAM_END.value[0]))
break

# print(message_chunk)
# 工具输出
if metadata["langgraph_node"] == "tools":
                    tool_name = message_chunk.name or"未知工具"
# logger.info(f"工具调用结果:{message_chunk.content}")
                    tool_use = "> 调用工具:" + tool_name + "\n\n"
await response.write(self._create_response(tool_use))
                    t02_answer_data.append(tool_use)
continue

# await response.write(self._create_response(agent.get_graph().draw_mermaid_png()))
# 输出最终结果
# print(message_chunk)
if message_chunk.content:
                    content = message_chunk.content
                    t02_answer_data.append(content)
await response.write(self._create_response(content))
# 确保实时输出
if hasattr(response, "flush"):
await response.flush()
await asyncio.sleep(0)

# 只有在未取消的情况下才保存记录
ifnot self.running_tasks[task_id]["cancelled"]:
await add_user_record(
                    uuid_str, session_id, query, t02_answer_data, {}, DiFyAppEnum.COMMON_QA.value[0], user_token
                )

except asyncio.CancelledError:
await response.write(self._create_response("\n> 这条消息已停止", "info", DataTypeEnum.ANSWER.value[0]))
await response.write(self._create_response("", "end", DataTypeEnum.STREAM_END.value[0]))
except Exception as e:
            print(f"[ERROR] Agent运行异常: {e}")
            traceback.print_exception(e)
await response.write(
                self._create_response("[ERROR] 智能体运行异常:", "error", DataTypeEnum.ANSWER.value[0])
            )
finally:
# 清理任务记录
if task_id in self.running_tasks:
del self.running_tasks[task_id]

asyncdefcancel_task(self, task_id: str) -> bool:
"""
        取消指定的任务
        :param task_id: 任务ID
        :return: 是否成功取消
        """
if task_id in self.running_tasks:
            self.running_tasks[task_id]["cancelled"] = True
returnTrue
returnFalse

defget_running_tasks(self):
"""
        获取当前运行中的任务列表
        :return: 运行中的任务列表
        """
return list(self.running_tasks.keys())

🧩 关键点

• checkpointer上下文对话记忆实现多轮对话的效果
• running_tasks按用户ID存储任务实现停止对话效果

🤖 MCP调用方式

根据业务场景合理配置MCP工具调用方式,推荐streamable_http方式进行工具调用

• streamable_http方式调用

self.client = MultiServerMCPClient({
"mcp-hub": {
"url": "http://xxxx.com",
"transport": "streamable_http",
  }
}

• 本地子进程方式调用三方开源工具

self.client = MultiServerMCPClient({
"undoom-douyin-data-analysis": {
"command": "uvx",
"transport": "stdio",
"args": [
"--index-url",
"https://mirrors.aliyun.com/pypi/simple/",
"--from",
"undoom-douyin-data-analysis",
"undoom-douyin-mcp",
           ],
  },
}

• 本地子进程方式调用本地开发的工具

current_dir = os.path.dirname(os.path.abspath(__file__))
mcp_tool_path = os.path.join(current_dir, "mcp", "query_db_tool.py")
self.client = MultiServerMCPClient({
"query_qa_record": {
"command": "python",
"args": [mcp_tool_path],
"transport": "stdio",
     }
}

完整代码

项目地址: git@github.com:apconw/sanic-web.git

项目简介

一个轻量级、支持全链路且易于二次开发的大模型应用项目

已集成MCP多智能体架构

基于Dify、LangChain/LangGraph、Lamaindex、MCP、Ollama&Vllm、Sanic 和 Text2SQL 等技术构建的一站式大模型应用开发项目，采用 Vue3、TypeScript 和 Vite 5 打造现代UI。它支持通过 ECharts / AntV 实现基于大模型的数据图形化问答，具备处理 CSV 文件 表格问答的能力。同时，能方便对接第三方开源 RAG 系统 检索系统 等，以支持广泛的通用知识问答。

---

如何系统学习掌握AI大模型？

AI大模型作为人工智能领域的重要技术突破，正成为推动各行各业创新和转型的关键力量。抓住AI大模型的风口，掌握AI大模型的知识和技能将变得越来越重要。

学习AI大模型是一个系统的过程，需要从基础开始，逐步深入到更高级的技术。

这里给大家精心整理了一份全面的AI大模型学习资源，包括：AI大模型全套学习路线图（从入门到实战）、精品AI大模型学习书籍手册、视频教程、实战学习、面试题等，资料免费分享！

1. 成长路线图&学习规划

要学习一门新的技术，作为新手一定要先学习成长路线图，方向不对，努力白费。

这里，我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。可以说是最科学最系统的学习成长路线。

2. 大模型经典PDF书籍

书籍和学习文档资料是学习大模型过程中必不可少的，我们精选了一系列深入探讨大模型技术的书籍和学习文档，它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。（书籍含电子版PDF）

3. 大模型视频教程

对于很多自学或者没有基础的同学来说，书籍这些纯文字类的学习教材会觉得比较晦涩难以理解，因此，我们提供了丰富的大模型视频教程，以动态、形象的方式展示技术概念，帮助你更快、更轻松地掌握核心知识。

4. 大模型行业报告

行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

5. 大模型项目实战

学以致用 ，当你的理论知识积累到一定程度，就需要通过项目实战，在实际操作中检验和巩固你所学到的知识，同时为你找工作和职业发展打下坚实的基础。

6. 大模型面试题

面试不仅是技术的较量，更需要充分的准备。

在你已经掌握了大模型技术之后，就需要开始准备面试，我们将提供精心整理的大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

全套的AI大模型学习资源已经整理打包，有需要的小伙伴可以微信扫描下方CSDN官方认证二维码，免费领取【保证100%免费】