当前位置： 首页 > news >正文

LangGraph系列：多智能体终极方案，ReAct+MCP工业级供应链系统

news 2025/11/15 6:38:54

在现代供应链管理中，管理人员每天要面对海量的库存数据、供应商信息、订单动态，传统的人工处理方式不仅效率低下，还容易因信息不对称导致库存积压或缺货、供应商选择失误等问题。随着AI技术的普及，智能体（Agent）系统逐渐成为解决这类复杂问题的有效方案，而ReAct（Reasoning + Acting）模式凭借“观察-思考-行动”的闭环逻辑，能让智能体像人类一样逐步拆解问题、调用工具解决问题，非常适合供应链这类多环节、强关联的场景。

我将带大家从零构建一个工业级的ReAct Agent供应链管理智能体系统，该系统基于LangGraph实现工作流编排，通过MCP协议标准化工具调用，支持本地化部署和双界面交互，既保证了数据安全，又具备极强的实用性和可扩展性。我们会深入技术细节，结合实际场景给出完整的代码实现，让你不仅能理解原理，还能直接落地使用。

在这里插入图片描述

一、系统设计思路与核心目标

做技术选型和架构设计前，我们首先要明确系统的核心目标：解决供应链管理中的实际痛点，比如库存动态监控、供应商智能评估、订单风险预警等，同时保证系统的稳定性、可扩展性和数据安全性。基于这些目标，我们确定了以下设计思路：

首先是本地化部署优先，供应链数据包含大量商业机密，因此系统必须支持本地存储和计算，避免数据外泄。我们选择SQLite作为数据库，它轻量级、零配置，无需额外部署服务，适合中小型企业的本地应用场景；大模型则采用Ollama，支持本地运行多种开源模型（如Llama 3、Mistral），无需依赖外部API，既保护隐私又降低使用成本。

其次是模块化与可扩展性，系统采用MCP（Model Context Protocol）工具协议，将所有业务功能封装为标准化工具，比如库存分析工具、供应商评估工具、订单查询工具等，后续新增功能只需按照协议开发新工具，无需修改核心逻辑。同时，核心的ReAct Agent引擎基于LangGraph实现，LangGraph的图结构特性天然适合描述“观察-思考-行动”的循环流程，支持灵活调整执行逻辑，比传统的线性流程更适应复杂场景。

最后是多端交互与数据可视化，考虑到不同用户的使用习惯，系统提供命令行（CLI）和React Web两种界面：CLI适合技术人员快速调试和自动化脚本集成，Web界面则面向业务人员，提供直观的仪表盘、数据图表和对话交互功能。数据可视化方面选用Recharts，它是React生态下的高性能图表库，能轻松实现库存趋势、供应商评分雷达图等可视化需求，帮助用户快速掌握关键信息。

整个系统的技术栈选型围绕“实用、高效、易落地”展开：后端用FastAPI提供高性能异步API，支持高并发处理；前端采用React + TypeScript保证类型安全和组件复用；异步处理依赖Python asyncio，提升多任务执行效率；工具协议遵循MCP标准，确保工具的可插拔性。这些技术的组合既兼顾了工业级应用的稳定性，又降低了开发和部署门槛。

二、核心组件深度实现：从Agent引擎到工具系统

（一）ReAct Agent核心引擎：实现“思考-行动”闭环

ReAct Agent是整个系统的大脑，其核心逻辑是“观察环境获取信息→分析问题形成思考→调用工具执行行动→根据结果迭代优化”的循环流程。要让这个循环高效运行，需要解决两个关键问题：状态管理和执行逻辑优化。

首先是状态管理，Agent在执行过程中需要记录观察结果、思考过程、已使用的工具、历史记忆等信息，这些状态数据是迭代优化的基础。我们设计了AgentState类来统一管理这些信息，其中memory字段存储历史交互记录，tools_used记录已调用的工具及结果，避免重复调用，observation和thought字段则记录当前循环的核心数据：

from dataclasses import dataclass, field
from typing import List, Dict, Any@dataclass
class AgentState:observation: str = ""  # 当前观察到的信息（如库存数据、供应商状态）thought: str = ""      # 思考过程（如“用户需要查询库存不足的商品，需调用库存分析工具”）action: str = ""       # 本次要执行的行动（如“调用库存分析工具”）memory: List[Dict[str, Any]] = field(default_factory=list)  # 历史记忆，存储过往交互tools_used: List[Dict[str, Any]] = field(default_factory=list)  # 已使用工具记录current_step: str = "observe"  # 当前所处阶段（observe/think/act）is_complete: bool = False  # 是否完成任务

接下来是核心执行循环，传统的ReAct循环容易出现迭代次数过多、效率低下的问题，我们在实现时加入了三项关键优化：并行观察、智能停止、状态压缩。并行观察能同时从多个数据源获取信息（如数据库、外部API），减少等待时间；智能停止机制会在连续两次迭代没有新进展时自动终止，避免无限循环；状态压缩则定期将冗长的历史记忆总结为摘要，减少内存占用。

完整的优化后执行循环代码如下，我们用asyncio实现异步操作，确保高并发处理能力：

import asyncio
from langgraph.graph import StateGraph, END
from langchain_community.llms import Ollamaclass ReActAgent:def __init__(self, tools: List["BaseMCPTool"], model_name: str = "llama3"):self.tools = {tool.name: tool for tool in tools}  # 工具字典，便于快速查找self.llm = Ollama(model=model_name, base_url="http://localhost:11434")  # 本地Ollama服务self.max_iterations = 5  # 最大迭代次数self.state = AgentState()  # 初始化状态# 构建LangGraph工作流：observe → think → act → (loop/end)self.graph = StateGraph(AgentState)self.graph.add_node("observe", self.observe_stage)self.graph.add_node("think", self.think_stage)self.graph.add_node("act", self.act_stage)self.graph.add_edge("observe", "think")self.graph.add_edge("think", "act")self.graph.add_conditional_edges("act",self.should_continue,  # 判断是否继续循环{True: "observe", False: END})self.graph.set_entry_point("observe")self.app = self.graph.compile()async def observe_stage(self, state: AgentState) -> AgentState:"""观察阶段：并行获取多个数据源的信息"""# 并行执行数据库查询、外部API调用（如供应商状态接口）db_obs, api_obs = await asyncio.gather(self._get_database_observation(),self._get_external_api_observation())# 合并观察结果state.observation = f"数据库信息：{db_obs}\n外部API信息：{api_obs}"# 更新状态阶段state.current_step = "observe"return stateasync def think_stage(self, state: AgentState) -> AgentState:"""思考阶段：基于观察结果和历史记忆，决定下一步行动"""# 构建提示词，包含历史记忆、当前观察结果prompt = f"""你是供应链管理智能体，需要帮用户解决库存、供应商、订单相关问题。历史交互：{state.memory}当前观察到的信息：{state.observation}可用工具：{[f"{name}: {tool.description}" for name, tool in self.tools.items()]}请思考：1. 用户的核心需求是什么？2. 是否需要调用工具？3. 调用哪个工具？思考过程要简洁，最终输出行动指令（格式：工具名称+参数，如“inventory_analysis: {{}}”，无需调用工具则输出“无需调用工具”）"""# 调用本地大模型生成思考过程和行动指令response = await self.llm.ainvoke(prompt)state.thought = response.content# 提取行动指令（从思考过程中解析工具名称和参数）state.action = self._parse_action(response.content)state.current_step = "think"return stateasync def act_stage(self, state: AgentState) -> AgentState:"""行动阶段：执行工具调用或直接返回结果"""if state.action == "无需调用工具":# 无需调用工具，直接生成最终回答final_answer = await self.llm.ainvoke(f"基于以下信息生成最终回答：{state.observation}\n思考过程：{state.thought}")state.memory.append({"observation": state.observation,"thought": state.thought,"action": "无需调用工具","result": final_answer.content})else:# 解析工具名称和参数tool_name, tool_params = self._parse_tool_params(state.action)if tool_name not in self.tools:result = f"错误：不存在工具{tool_name}"else:# 调用工具（异步执行）tool = self.tools[tool_name]result = await tool.execute(**tool_params)# 记录工具使用情况state.tools_used.append({"tool_name": tool_name,"params": tool_params,"result": result})# 更新历史记忆state.memory.append({"observation": state.observation,"thought": state.thought,"action": state.action,"result": result})state.current_step = "act"state.is_complete = state.action == "无需调用工具"return statedef should_continue(self, state: AgentState) -> bool:"""判断是否继续循环：未完成且迭代次数未超限"""if len(state.memory) >= self.max_iterations:return False# 智能停止：连续两次迭代无新工具调用或结果无变化，则停止if len(state.memory) >= 2:prev_action = state.memory[-2]["action"]curr_action = state.memory[-1]["action"]if prev_action == curr_action == "无需调用工具":return Falsereturn not state.is_complete# 辅助方法：解析行动指令def _parse_action(self, thought: str) -> str:"""从思考过程中提取行动指令"""if "无需调用工具" in thought:return "无需调用工具"# 简单解析（实际项目可使用正则表达式或LLM解析）for tool_name in self.tools.keys():if tool_name in thought:return f"{tool_name}: {{}}"  # 此处简化，实际需提取参数return "无需调用工具"# 辅助方法：获取数据库观察信息async def _get_database_observation(self) -> str:"""从SQLite数据库获取库存和订单的最新状态"""import sqlite3conn = sqlite3.connect("supply_chain.db")cursor = conn.cursor()# 查询库存不足的商品（低于最小库存阈值）cursor.execute("SELECT name, current_stock, min_stock FROM inventory WHERE current_stock < min_stock")low_stock = cursor.fetchall()conn.close()if low_stock:return f"库存不足商品：{[f'{name}（当前：{curr}，最小：{min_}）' for name, curr, min_ in low_stock]}"return "无库存不足商品"# 辅助方法：获取外部API观察信息async def _get_external_api_observation(self) -> str:"""调用外部API获取供应商状态（模拟）"""import aiohttpasync with aiohttp.ClientSession() as session:try:async with session.get("http://localhost:8000/api/suppliers/status") as resp:if resp.status == 200:data = await resp.json()return f"供应商状态：{data}"except Exception as e:return f"外部API调用失败：{str(e)}"return "未获取到供应商状态信息"

这个Agent引擎的核心优势在于LangGraph的工作流编排能力，将“观察-思考-行动”拆分为三个节点，通过条件边控制循环逻辑，比传统的线性代码更清晰、更易维护。同时，异步处理的设计让多个数据源查询可以并行执行，大幅提升了执行效率，这在供应链管理中尤为重要，比如同时查询库存数据和供应商交付状态，无需等待一个完成再执行另一个。

（二）MCP工具系统：标准化可插拔的业务功能封装

MCP协议的核心思想是将工具的调用接口标准化，让Agent可以无需关注工具的具体实现，只需按照统一的协议调用即可。我们设计了BaseMCPTool抽象基类，所有工具都必须实现execute方法和name、description属性，这样Agent在选择工具时，只需根据description判断工具用途，调用时直接传入参数即可。

首先是工具基类的实现，使用Python的ABC（抽象基类）模块确保子类必须实现指定方法：

from abc import ABC, abstractmethod
from typing import Dict, Anyclass BaseMCPTool(ABC):"""MCP工具基类，定义标准化接口"""@abstractmethodasync def execute(self, **kwargs) -> Dict[str, Any]:"""执行工具操作，返回结果字典"""pass@property@abstractmethoddef name(self) -> str:"""工具名称（唯一标识）"""pass@property@abstractmethoddef description(self) -> str:"""工具描述（供Agent理解工具用途）"""pass

基于这个基类，我们实现供应链管理中最常用的三个工具：库存分析工具、供应商评估工具、订单风险预警工具。每个工具都封装了具体的业务逻辑，并且可以独立测试和扩展。

以库存分析工具为例，它的核心功能是查询库存数据，分析哪些商品库存不足需要补货、哪些商品库存积压需要清理，并给出补货建议（基于历史销量数据）：

import sqlite3
from typing import Dict, List, Anyclass InventoryAnalysisTool(BaseMCPTool):@propertydef name(self) -> str:return "inventory_analysis"@propertydef description(self) -> str:return "库存分析工具：查询所有商品的库存状态，识别库存不足（低于最小库存）和库存积压（高于最大库存）的商品，提供补货建议"async def execute(self, **kwargs) -> Dict[str, Any]:"""执行库存分析参数：product_name: 可选，指定商品名称（不指定则分析所有商品）返回：low_stock_items: 库存不足商品列表overstock_items: 库存积压商品列表reorder_recommendations: 补货建议列表"""product_name = kwargs.get("product_name")# 连接SQLite数据库查询库存数据conn = sqlite3.connect("supply_chain.db")conn.row_factory = sqlite3.Row  # 支持按列名访问cursor = conn.cursor()# 构建查询SQLif product_name:cursor.execute("""SELECT id, name, current_stock, min_stock, max_stock, avg_daily_sales FROM inventory WHERE name LIKE ?""", (f"%{product_name}%",))else:cursor.execute("""SELECT id, name, current_stock, min_stock, max_stock, avg_daily_sales FROM inventory""")inventory_data = cursor.fetchall()conn.close()# 分析库存状态low_stock = []overstock = []reorder_recommendations = []for item in inventory_data:item_dict = dict(item)if item["current_stock"] < item["min_stock"]:low_stock.append(item_dict)# 计算补货量：建议补到最大库存，考虑日均销量（补货周期按7天计算）reorder_qty = item["max_stock"] - item["current_stock"] + item["avg_daily_sales"] * 7reorder_recommendations.append({"product_name": item["name"],"current_stock": item["current_stock"],"reorder_qty": reorder_qty,"reason": f"库存低于最小阈值（最小：{item['min_stock']}）"})elif item["current_stock"] > item["max_stock"]:overstock.append(item_dict)return {"low_stock_items": low_stock,"overstock_items": overstock,"reorder_recommendations": reorder_recommendations,"total_products": len(inventory_data)}

供应商评估工具则用于分析供应商的综合表现，基于质量、交付准时率、价格竞争力三个维度进行评分，并生成雷达图数据，帮助用户选择最优供应商：

class SupplierEvaluationTool(BaseMCPTool):@propertydef name(self) -> str:return "supplier_evaluation"@propertydef description(self) -> str:return "供应商评估工具：分析供应商的质量、交付准时率、价格竞争力，生成综合评分和雷达图数据"async def execute(self, **kwargs) -> Dict[str, Any]:"""执行供应商评估参数：supplier_id: 可选，指定供应商ID（不指定则评估所有供应商）返回：suppliers: 供应商评估列表（包含综合评分和各维度得分）"""supplier_id = kwargs.get("supplier_id")conn = sqlite3.connect("supply_chain.db")conn.row_factory = sqlite3.Rowcursor = conn.cursor()if supplier_id:cursor.execute("""SELECT id, name, quality_score, delivery_score, price_score FROM suppliers WHERE id = ?""", (supplier_id,))else:cursor.execute("""SELECT id, name, quality_score, delivery_score, price_score FROM suppliers""")suppliers_data = cursor.fetchall()conn.close()result = []for supplier in suppliers_data:# 计算综合评分（加权平均：质量40%，交付30%，价格30%）overall_score = (supplier["quality_score"] * 0.4 +supplier["delivery_score"] * 0.3 +supplier["price_score"] * 0.3)# 生成雷达图数据（适配Recharts的Radar组件）radar_data = [{"subject": "质量", "score": supplier["quality_score"], "fullMark": 100},{"subject": "交付准时率", "score": supplier["delivery_score"], "fullMark": 100},{"subject": "价格竞争力", "score": supplier["price_score"], "fullMark": 100}]result.append({"id": supplier["id"],"name": supplier["name"],"overall_score": round(overall_score, 2),"radar_data": radar_data,"dimensions": {"quality": supplier["quality_score"],"delivery": supplier["delivery_score"],"price": supplier["price_score"]}})return {"suppliers": result}

这种工具化的设计带来了极强的可扩展性，比如后续需要新增“订单跟踪工具”，只需新建一个类继承BaseMCPTool，实现execute方法和相关属性，然后在Agent初始化时加入工具列表即可，无需修改Agent的核心逻辑。在实际项目中，我们还可以为工具添加权限控制、缓存机制等功能，进一步提升系统的实用性。

（三）数据库设计：支撑供应链核心数据存储

数据库是系统的数据基础，我们需要设计合理的表结构来存储库存、供应商、订单等核心数据。考虑到系统是本地化部署且面向中小型应用，SQLite完全能满足需求，其文件型数据库的特性也便于备份和迁移。

核心表结构设计如下（我们用SQL语句直接创建，无需ORM框架，简化部署）：

-- 库存表：存储商品库存信息
CREATE TABLE IF NOT EXISTS inventory (id INTEGER PRIMARY KEY AUTOINCREMENT,name TEXT NOT NULL,  -- 商品名称sku TEXT UNIQUE NOT NULL,  -- 商品SKU编码current_stock INTEGER NOT NULL DEFAULT 0,  -- 当前库存数量min_stock INTEGER NOT NULL DEFAULT 10,  -- 最小库存阈值（低于则预警）max_stock INTEGER NOT NULL DEFAULT 100,  -- 最大库存阈值（高于则积压）avg_daily_sales REAL NOT NULL DEFAULT 5,  -- 日均销量（用于补货建议）category TEXT,  -- 商品分类created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);-- 供应商表：存储供应商信息和评分
CREATE TABLE IF NOT EXISTS suppliers (id INTEGER PRIMARY KEY AUTOINCREMENT,name TEXT NOT NULL,  -- 供应商名称contact_person TEXT,  -- 联系人phone TEXT,  -- 联系电话email TEXT,  -- 邮箱quality_score INTEGER NOT NULL DEFAULT 80,  -- 质量评分（0-100）delivery_score INTEGER NOT NULL DEFAULT 80,  -- 交付准时率评分（0-100）price_score INTEGER NOT NULL DEFAULT 80,  -- 价格竞争力评分（0-100）is_active BOOLEAN DEFAULT 1,  -- 是否启用created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);-- 订单表：存储订单信息
CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY AUTOINCREMENT,order_no TEXT UNIQUE NOT NULL,  -- 订单编号product_id INTEGER NOT NULL,  -- 关联商品IDsupplier_id INTEGER NOT NULL,  -- 关联供应商IDquantity INTEGER NOT NULL,  -- 订单数量order_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,  -- 下单日期expected_delivery_date TIMESTAMP,  -- 预计交付日期actual_delivery_date TIMESTAMP,  -- 实际交付日期status TEXT NOT NULL DEFAULT 'pending',  -- 订单状态：pending/paid/delivered/canceledFOREIGN KEY (product_id) REFERENCES inventory (id),FOREIGN KEY (supplier_id) REFERENCES suppliers (id)
);-- 初始化测试数据
INSERT INTO inventory (name, sku, current_stock, min_stock, max_stock, avg_daily_sales, category)
VALUES 
('笔记本电脑', 'LT-001', 15, 10, 30, 2.5, '电子产品'),
('无线鼠标', 'MS-001', 8, 10, 50, 4.2, '电子产品'),
('办公桌椅套装', 'FS-001', 30, 15, 50, 1.8, '办公家具'),
('打印纸（A4）', 'PP-001', 120, 50, 200, 10.5, '办公用品');INSERT INTO suppliers (name, contact_person, phone, email, quality_score, delivery_score, price_score)
VALUES 
('科技设备有限公司', '张三', '13800138000', 'zhangsan@tech.com', 92, 88, 85),
('办公耗材供应商', '李四', '13900139000', 'lisi@office.com', 85, 90, 92),
('家具制造工厂', '王五', '13700137000', 'wangwu@furniture.com', 88, 82, 80);

这个表结构覆盖了供应链管理的核心数据实体，并且字段设计简洁实用，没有冗余信息。比如库存表中的avg_daily_sales字段用于计算补货建议，供应商表中的三个评分字段用于综合评估，订单表中的状态字段用于跟踪订单进度。在实际使用中，我们可以根据业务需求新增字段，比如库存表中的供应商ID（关联默认供应商）、订单表中的金额字段等。

三、前后端交互与可视化实现

后端API设计：基于FastAPI的异步接口

后端采用FastAPI框架，它天生支持异步操作，性能优秀，并且能自动生成API文档（访问/docs即可查看），非常适合快速开发和调试。我们设计了三类核心API：Agent交互API、数据查询API、系统状态API，分别对应智能交互、数据获取、系统监控三大功能。

首先是API的整体实现，我们将Agent、工具、数据库操作封装为独立的服务，然后通过API接口暴露给前端：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Dict, Any
import sqlite3
import asyncio# 初始化FastAPI应用
app = FastAPI(title="供应链管理智能体API", version="1.0")# 导入之前实现的Agent和工具
from agent import ReActAgent
from tools import InventoryAnalysisTool, SupplierEvaluationTool# 初始化工具列表
tools = [InventoryAnalysisTool(),SupplierEvaluationTool()
]# 初始化ReAct Agent
react_agent = ReActAgent(tools=tools, model_name="llama3")# 数据模型（Pydantic）：定义请求和响应格式
class AgentQueryRequest(BaseModel):query: str  # 用户查询语句class AgentQueryResponse(BaseModel):response: str  # 最终回答steps: List[Dict[str, Any]]  # 执行步骤（观察、思考、行动）tools_used: List[Dict[str, Any]]  # 已使用的工具class SystemStatusResponse(BaseModel):database: bool  # 数据库连接状态ollama: bool    # Ollama服务状态agent: str      # Agent状态# 核心API接口
@app.post("/api/agent/query", response_model=AgentQueryResponse)
async def query_agent(request: AgentQueryRequest):"""Agent交互接口：处理用户查询，返回ReAct执行结果"""try:# 执行Agent工作流result = await react_agent.app.ainvoke(react_agent.state)# 提取最终回答（从历史记忆的最后一条结果中获取）final_response = result.memory[-1]["result"]# 构造响应数据return {"response": final_response,"steps": [{"step": item["observation"],"thought": item["thought"],"action": item["action"],"result": item["result"]} for item in result.memory],"tools_used": result.tools_used}except Exception as e:raise HTTPException(status_code=500, detail=f"Agent处理失败：{str(e)}")@app.get("/api/inventory", response_model=Dict[str, Any])
async def get_inventory(product_name: str = None):"""库存查询API：获取库存数据，支持按商品名称筛选"""conn = sqlite3.connect("supply_chain.db")conn.row_factory = sqlite3.Rowcursor = conn.cursor()try:if product_name:cursor.execute("""SELECT id, name, sku, current_stock, min_stock, max_stock, avg_daily_sales, category FROM inventory WHERE name LIKE ?""", (f"%{product_name}%",))else:cursor.execute("""SELECT id, name, sku, current_stock, min_stock, max_stock, avg_daily_sales, category FROM inventory""")data = [dict(row) for row in cursor.fetchall()]return {"data": data,"total": len(data),"low_stock_count": len([item for item in data if item["current_stock"] < item["min_stock"]]),"overstock_count": len([item for item in data if item["current_stock"] > item["max_stock"]])}finally:conn.close()@app.get("/api/suppliers", response_model=Dict[str, Any])
async def get_suppliers(supplier_id: int = None):"""供应商查询API：获取供应商评估数据"""conn = sqlite3.connect("supply_chain.db")conn.row_factory = sqlite3.Rowcursor = conn.cursor()try:if supplier_id:cursor.execute("""SELECT id, name, contact_person, phone, email, quality_score, delivery_score, price_score FROM suppliers WHERE id = ?""", (supplier_id,))data = [dict(row) for row in cursor.fetchall()]else:cursor.execute("""SELECT id, name, contact_person, phone, email, quality_score, delivery_score, price_score FROM suppliers""")data = [dict(row) for row in cursor.fetchall()]# 计算综合评分for item in data:item["overall_score"] = round(item["quality_score"] * 0.4 +item["delivery_score"] * 0.3 +item["price_score"] * 0.3,2)return {"data": data, "total": len(data)}finally:conn.close()@app.get("/api/system/status", response_model=SystemStatusResponse)
async def get_system_status():"""系统状态API：检查数据库、Ollama服务、Agent状态"""# 检查数据库连接db_ok = Falsetry:conn = sqlite3.connect("supply_chain.db")conn.execute("SELECT 1")db_ok = Trueconn.close()except:pass# 检查Ollama服务ollama_ok = Falsetry:async with aiohttp.ClientSession() as session:async with session.get("http://localhost:11434/api/tags") as resp:ollama_ok = resp.status == 200except:pass# Agent状态agent_status = "running" if react_agent.state.current_step else "stopped"return {"database": db_ok,"ollama": ollama_ok,"agent": agent_status}# 启动服务（本地开发用）
if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

这些API接口设计遵循RESTful规范，使用GET请求获取数据，POST请求处理交互，参数通过查询字符串或请求体传递，响应格式统一为JSON，便于前端处理。异步接口的设计确保了系统在高并发场景下的性能，比如同时有多个用户查询Agent或获取库存数据时，不会出现阻塞。

四、系统部署与实际应用场景

本地化部署步骤

由于系统采用本地化部署方案，部署过程非常简单，无需复杂的服务器配置，只需按照以下步骤操作即可：

  1. 环境准备：安装Python 3.8+、Node.js 16+、Ollama。

    • Python用于运行后端FastAPI服务和Agent引擎；
    • Node.js用于运行前端React应用；
    • Ollama用于本地大模型服务，安装后需拉取模型（如ollama pull llama3）。

  2. 后端部署：

    • 克隆项目代码，进入后端目录，安装依赖：pip install fastapi uvicorn aiohttp langgraph langchain-community sqlite3；
    • 初始化数据库：运行python init_db.py（执行之前的SQL表结构创建脚本）；
    • 启动后端服务：uvicorn main:app --reload --host 0.0.0.0 --port 8000，服务启动后可访问http://localhost:8000/docs查看API文档。

  3. 前端部署：

    • 进入前端目录，安装依赖：npm install；
    • 启动前端服务：npm run start，默认端口3000，访问http://localhost:3000即可进入Web界面。

  4. CLI界面使用：

    • 后端目录中创建cli.py文件，实现简单的命令行交互：
      import asyncio
from agent import ReActAgent
from tools import InventoryAnalysisTool, SupplierEvaluationToolasync def main():tools = [InventoryAnalysisTool(), SupplierEvaluationTool()]agent = ReActAgent(tools=tools)print("供应链管理智能体（CLI模式）- 输入'quit'退出")while True:query = input("请输入查询：")if query.lower() == 'quit':breakresult = await agent.app.ainvoke(agent.state)print(f"智能体回答：{result.memory[-1]['result']}")if __name__ == "__main__":asyncio.run(main())
    • 运行CLI：python cli.py，即可在命令行中与Agent交互。

整个部署过程无需依赖云服务，所有数据都存储在本地SQLite数据库中，适合中小型企业或对数据隐私有严格要求的场景。

五、总结

本文详细介绍了基于LangGraph与MCP的工业级ReAct Agent供应链管理系统的设计与实现，从核心组件的技术细节到完整的部署步骤，提供了一套可直接落地的解决方案。该系统通过ReAct模式实现了“观察-思考-行动”的智能闭环，MCP协议保证了工具的可插拔性，本地化部署确保了数据安全，双界面交互满足了不同用户的使用需求。

与传统的供应链管理系统相比，这个智能体系统的优势在于它能理解自然语言查询，自动调用工具解决问题，无需用户手动操作复杂的查询界面，大幅提升了工作效率。同时，模块化的设计让系统具备极强的可扩展性，后续可以根据业务需求不断新增功能，而无需重构核心逻辑。

http://www.dtcms.com/a/609444.html

相关文章：

  • 移动端 HTTPS 抓包实战，多工具组合分析与高效排查指南
  • 自动化测试工具Parasoft C/C++test如何导入IAR项目
  • 蓝桥杯嵌入式赛道—-软件篇（GPIO输出模式配置）
  • 仿第四城地方门户网站模板python可以做网站吗
  • ToDesk如何充当分屏显示器？扩展屏、多屏结合虚拟屏都可行！
  • qKnow 知识平台开源版 v1.0.3 发布：Docker Compose 部署 多项稳定性优化和关键问题修复
  • 【Prompt学习技能树地图】检索增强生成（RAG）核心技术剖析与实践指南
  • Zookeeper在Kafka中的作用
  • MySQL数据表操作
  • 访问外国网站速度慢怎样优化标题关键词
  • 《VScode搭建教程（附安装包）--- 开启你的编程之旅》
  • MFC Edit Control控件完全指南：从基础使用到高级定制
  • C语言编译器安装教程 | 快速上手C语言编译环境的配置与调试
  • C库OpenSSL安装与VisualStudio配置
  • ppt模板免费下载网站不需要登录网页设计的三大基本技术
  • 什么是营销型网站?h5可以连接别的网站吗
  • Android13修改系统支持gps同步时间
  • Webpack到Vite：构建工具迁移实战经验总结
  • QT音乐播放器18----新歌速递播放、隐藏顶部和底部工具栏、自定义ToolTips
  • 工控一体机在超声波清洗机中的应用
  • Qt5配置MSVC2017
  • 【AI】基于YOLO11-SEG的PCB缺陷检测与分类系统实现_3
  • 2. Qt深入 线程
  • [Godot] C#基于噪声的简单TileMap地图生成
  • Linux（4）—— 基础开发工具
  • 广州好的网站设计公司如何建单位网站
  • 无损改造+智能防控，安科瑞ASCB3为古建筑与历史街区量身打造安全用电解决方案
  • 解决“Move to iOS 卡在准备中”的 9 种有效方法
  • iOS IPA 上传工具全面解析，从 Transporter 到开心上架（Appuploader）命令行的高效上架实践
  • iOS性能调优的系统化实践，从架构分层到多工具协同的全流程优化指南（开发者深度版）