news 2026/4/17 22:53:42

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

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
LangGraph+MCP+ReactAgent技术组合，构建智能代理（Agent）系统

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

实现效果

核心技术栈

  1. LangGraph

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

  1. MCP（Model Context Protocol）

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

  1. 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方式调用*** ​```plaintext 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 系统 检索系统 🌐等，以支持广泛的通用知识问答。

作为轻量级的大模型应用开发项目，Sanic-Web 🛠️ 支持快速迭代与扩展，助力大模型项目快速落地。🚀

想入门 AI 大模型却找不到清晰方向？备考大厂 AI 岗还在四处搜集零散资料？别再浪费时间啦！2025 年AI 大模型全套学习资料已整理完毕，从学习路线到面试真题，从工具教程到行业报告，一站式覆盖你的所有需求，现在全部免费分享！

👇👇扫码免费领取全部内容👇👇

​

一、学习必备：100+本大模型电子书+26 份行业报告 + 600+ 套技术PPT，帮你看透 AI 趋势

想了解大模型的行业动态、商业落地案例？大模型电子书？这份资料帮你站在 “行业高度” 学 AI：

1. 100+本大模型方向电子书

2. 26 份行业研究报告：覆盖多领域实践与趋势

报告包含阿里、DeepSeek 等权威机构发布的核心内容，涵盖：

  • 职业趋势：《AI + 职业趋势报告》《中国 AI 人才粮仓模型解析》；
  • 商业落地：《生成式 AI 商业落地白皮书》《AI Agent 应用落地技术白皮书》；
  • 领域细分：《AGI 在金融领域的应用报告》《AI GC 实践案例集》；
  • 行业监测：《2024 年中国大模型季度监测报告》《2025 年中国技术市场发展趋势》。

3. 600+套技术大会 PPT：听行业大咖讲实战

PPT 整理自 2024-2025 年热门技术大会，包含百度、腾讯、字节等企业的一线实践：

  • 安全方向：《端侧大模型的安全建设》《大模型驱动安全升级（腾讯代码安全实践）》；
  • 产品与创新：《大模型产品如何创新与创收》《AI 时代的新范式：构建 AI 产品》；
  • 多模态与 Agent：《Step-Video 开源模型（视频生成进展）》《Agentic RAG 的现在与未来》；
  • 工程落地：《从原型到生产：AgentOps 加速字节 AI 应用落地》《智能代码助手 CodeFuse 的架构设计》。

二、求职必看：大厂 AI 岗面试 “弹药库”，300 + 真题 + 107 道面经直接抱走

想冲字节、腾讯、阿里、蔚来等大厂 AI 岗？这份面试资料帮你提前 “押题”，拒绝临场慌！

1. 107 道大厂面经：覆盖 Prompt、RAG、大模型应用工程师等热门岗位

面经整理自 2021-2025 年真实面试场景，包含 TPlink、字节、腾讯、蔚来、虾皮、中兴、科大讯飞、京东等企业的高频考题，每道题都附带思路解析：

2. 102 道 AI 大模型真题：直击大模型核心考点

针对大模型专属考题，从概念到实践全面覆盖，帮你理清底层逻辑：

3. 97 道 LLMs 真题：聚焦大型语言模型高频问题

专门拆解 LLMs 的核心痛点与解决方案，比如让很多人头疼的 “复读机问题”：


三、路线必明： AI 大模型学习路线图，1 张图理清核心内容

刚接触 AI 大模型，不知道该从哪学起？这份「AI大模型 学习路线图」直接帮你划重点，不用再盲目摸索！

路线图涵盖 5 大核心板块，从基础到进阶层层递进：一步步带你从入门到进阶，从理论到实战。

L1阶段:启航篇丨极速破界AI新时代

L1阶段：了解大模型的基础知识，以及大模型在各个行业的应用和分析，学习理解大模型的核心原理、关键技术以及大模型应用场景。

L2阶段：攻坚篇丨RAG开发实战工坊

L2阶段：AI大模型RAG应用开发工程，主要学习RAG检索增强生成：包括Naive RAG、Advanced-RAG以及RAG性能评估，还有GraphRAG在内的多个RAG热门项目的分析。

L3阶段：跃迁篇丨Agent智能体架构设计

L3阶段：大模型Agent应用架构进阶实现，主要学习LangChain、 LIamaIndex框架，也会学习到AutoGPT、 MetaGPT等多Agent系统，打造Agent智能体。

L4阶段：精进篇丨模型微调与私有化部署

L4阶段：大模型的微调和私有化部署，更加深入的探讨Transformer架构，学习大模型的微调技术，利用DeepSpeed、Lamam Factory等工具快速进行模型微调，并通过Ollama、vLLM等推理部署框架，实现模型的快速部署。

L5阶段：专题集丨特训篇 【录播课】


四、资料领取：全套内容免费抱走，学 AI 不用再找第二份

不管你是 0 基础想入门 AI 大模型，还是有基础想冲刺大厂、了解行业趋势，这份资料都能满足你！
现在只需按照提示操作，就能免费领取：

👇👇扫码免费领取全部内容👇👇

​

2025 年想抓住 AI 大模型的风口？别犹豫，这份免费资料就是你的 “起跑线”！

版权声明: 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！
网站建设 2026/4/13 14:14:33

Z-Image-Turbo指令遵循性差？Prompt优化实战教程提升效果

Z-Image-Turbo指令遵循性差？Prompt优化实战教程提升效果 你是不是也遇到过这种情况：满怀期待地输入一段精心设计的提示词，点击生成，结果Z-Image-Turbo却“自由发挥”，画面和你想的根本不是一回事？文字错乱…

作者头像 李华
网站建设 2026/4/18 5:37:40

告别git checkout：现代Git工作流效率提升指南

快速体验 打开 InsCode(快马)平台 https://www.inscode.net输入框内输入如下内容： 构建一个Git性能分析工具，功能：1. 统计不同分支切换方式的耗时对比 2. 检测仓库状态对切换速度的影响 3. 提供索引优化建议 4. 集成git switch/restore命令…

作者头像 李华
网站建设 2026/4/4 20:50:14

5分钟搞懂什么是‘一级一级保一级‘管理原则

快速体验 打开 InsCode(快马)平台 https://www.inscode.net输入框内输入如下内容： 创建一个交互式学习应用，通过动画和简单案例帮助用户理解一级一级保一级概念。应用需包含概念讲解、互动示例、小测验和常见问题解答。采用递进式教学设计，…

作者头像 李华
网站建设 2026/4/17 20:12:20

ENSP实战：从下载到搭建完整企业网络实验

快速体验 打开 InsCode(快马)平台 https://www.inscode.net输入框内输入如下内容： 开发一个基于ENSP的企业网络实验教学模块。包含从官网获取ENSP的步骤说明，自动生成典型企业网络拓扑图（总部-分支结构），预配置VLAN、…

作者头像 李华
网站建设 2026/4/16 1:32:11

零基础搞定TOMCAT7下载安装

快速体验 打开 InsCode(快马)平台 https://www.inscode.net输入框内输入如下内容： 制作一个面向新手的TOMCAT7入门指南，包含：1) 官网下载页面的截图标注 2) 解压目录结构图解 3) 配置JAVA_HOME的动画演示 4) 部署WAR包的步骤截图 5) 常见错…

作者头像 李华
网站建设 2026/3/31 13:29:26

本地部署资产跟踪器 DumbAssets 并实现外网访问

DumbAssets 是一款简单好用的资产跟踪器，这款软件可以帮助用户管理和记录他们的物理资产，包括资产的组成部分、保修信息和例行维护，有了这款资产跟踪器，用户可以随时了解自己的资产情况。本文将详细的介绍如何利用 Docker 在本地部…

作者头像 李华