数字员工智能平台 - 采购审核

git地址：https://github.com/zxliucn/digital_employee_platform

一、项目概述

数字员工智能平台是一个基于 LangGraph + LLM（DeepSeek）的采购订单智能审核系统。用户输入采购单编号，系统自动：

1. 从数据库（或模拟数据）获取采购单明细和商品档案
2. 计算价格溢价、动销状态、周转天数、退货率等业务指标
3. 自动判定风险（溢价过高、滞销品采购过多、库存积压、高退货率等）
4. 调用 LLM 生成专业审核报告，给出最终决策：通过 / 驳回 / 人工复核

前端使用 Gradio 提供交互界面，支持报告下载（MD / HTML）和历史报告归档查看。

二、技术架构

2.1 整体流程

用户输入采购单ID
    ↓
[Gradio UI] → app.py
    ↓
[LangGraph 状态图] → agents/procurement/graph_definition.py
    ↓
节点1: validate_input → 校验ID, 获取采购单+商品档案 (via MCPClient)
节点2: calculate_metrics → 计算指标, 生成风险标记
节点3: generate_report → LLM生成审核报告, 解析最终决策
    ↓
返回报告 + 自动归档到 reports/ 目录

2.2 技术栈

三、目录结构

digital_employee_platform1/
├── app.py                          # 主入口：Gradio UI + 审核调度 + 报告归档
├── .env                            # 全局配置（LLM、阈值、MCP开关）
│
├── config/                         # 配置层
│   ├── settings.py                 # Settings 类：加载 .env，提供所有配置常量
│   ├── logging_config.py           # JSON日志格式 + TraceID（ContextVar）
│   └── __init__.py
│
├── core/                           # 核心状态定义
│   ├── state.py                    # ProcurementState TypedDict（LangGraph 状态图数据结构）
│   └── __init__.py
│
├── models/                         # LLM 接入层
│   ├── llm_gateway.py              # LLMGateway 单例：ChatOpenAI 或 MockLLM
│   └── __init__.py
│
├── tools/                          # 数据查询工具层
│   ├── mcp_client.py               # MCPClient：真实 MCP 调用或模拟数据（_MockMCPClient）
│   └── __init__.py
│
├── agents/                         # 数字员工（Agent）层
│   ├── __init__.py                 # 注册中心（后续新增员工在此导入）
│   └ procurement/
│       ├── graph_definition.py     # LangGraph 状态图构建（三节点线性流水线）
│       ├── nodes.py                # 三个节点实现：validate_input / calculate_metrics / generate_report
│       ├── prompts.py              # LLM Prompt：系统提示 + 用户提示模板
│       └── __init__.py
│
├── mcp_server/                     # MCP Server（独立进程，JSON-RPC stdin/stdout）
│   ├── server.py                   # MCP Server 主程序：读取JSON请求 → 调用 db_queries → 返回JSON响应
│   ├── db_queries.py               # 数据库查询层：MySQLClient（asyncmy连接池）+ 模拟数据 + 业务SQL
│   ├── schemas.py                  # Pydantic 数据模型定义
│   ├── test_db_connection.py       # 数据库连接测试脚本
│   └── .env                        # MCP Server 专属配置（DATABASE_URL 等）
│
├── reports/                        # 审核报告归档目录（自动生成 .md 文件）
│
└── .idea/                          # IDE 配置（JetBrains）

四、核心模块详解

4.1 app.py — 主入口 & Gradio UI

职责：

• 初始化日志和配置
• 提供 Gradio Web UI（输入采购单ID → 显示审核报告）
• 异步执行审核流程（run_audit_async）
• 报告自动归档到 reports/ 目录（文件名格式：{order_id}_{timestamp}_{decision}.md）
• 支持下载报告为 .md 和 .html 格式
• 历史报告下拉列表查看

关键函数：

UI 特性：

• 科技感扁平化设计（0圆角按钮、直角输入框）
• 报告卡片式渲染（.report-box CSS）
• 支持回车提交（input_box.submit）

4.2 config/settings.py — 配置管理

从 .env 加载所有配置，提供 Settings 类单例 settings：

4.3 config/logging_config.py — 日志系统

• JsonFormatter：所有日志输出为 JSON 格式，包含 timestamp、trace_id、level、module、message 等字段
• trace_id_var：使用 ContextVar 在整个请求链路中传递 TraceID
• generate_trace_id()：生成 UUID 作为请求追踪标识

4.4 core/state.py — 状态定义

ProcurementState 是 LangGraph 的状态数据结构（TypedDict）：

4.5 agents/procurement/ — 采购审核 Agent

4.5.1 graph_definition.py — 状态图

三节点线性流水线：

validate → calculate_metrics → generate_report → END

使用 StateGraph(ProcurementState) 构建，编译后返回可执行的 graph。

4.5.2 nodes.py — 三个节点

节点1：validate_input

• 校验 order_id 是否存在
• 调用 MCPClient.get_purchase_order(order_id) 获取采购单数据
• 调用 MCPClient.query_product_archive(product_ids) 批量获取商品档案
• 提取采购单主信息（order_id、supplier、order_date、total_amount）
• 返回 purchase_order、order_items、product_archives

节点2：calculate_metrics_and_risk

对每个商品项计算：

风险判定规则：

节点3：generate_report

• 使用 LLMGateway.get_llm() 获取 LLM 实例
• 组装 system prompt + human prompt（含采购单信息、分析结果、风险标记）
• 调用 llm.ainvoke() 生成审核报告
• 简单解析最终决策：报告中含"通过"且不含"驳回"→ 通过；含"驳回"→ 驳回；否则→ 人工复核
• LLM 调用失败时降级生成基础报告

4.5.3 prompts.py — LLM Prompt

SYSTEM_PROMPT：定义 LLM 角色（资深供应链采购审核专家），明确说明：

• 分析指标已由系统计算完成，LLM 只需引用不需重算
• 输出必须包含：概览、商品明细审核表、风险汇总、最终结论
• 审核表必须包含：商品ID、当前库存、补货后总库存、溢价率、退货率等字段

HUMAN_PROMPT_TEMPLATE：将 order_info、analysis_results、risk_flags 格式化传入

4.6 tools/mcp_client.py — 数据查询层

提供两个核心查询方法：

两种模式：

• 模拟模式 (USE_MOCK_MCP=true)：使用 _MockMCPClient 内存数据（5个商品 + 2个采购单）
• 真实模式 (USE_MOCK_MCP=false)：使用 _MCPClient，每次调用启动独立 MCP Server 进程，通过 stdin/stdout JSON-RPC 通信，调用后立即清理进程

4.7 mcp_server/ — MCP Server

4.7.1 server.py — MCP Server 主程序

• 基于 JSON-RPC 2.0 协议，通过 stdin/stdout 通信
• 支持 tools/call 方法，注册两个工具：get_purchase_order、query_product_archive
• 使用 asyncio Stream Protocol 读写

4.7.2 db_queries.py — 数据库查询层

MySQLClient 类：

• 使用 SQLAlchemy asyncmy 驱动
• 连接池配置：pool_size=10、max_overflow=20、pool_recycle=3600
• 支持重试（3次，指数退避）
• 提供 execute / execute_one / get_session 方法

业务 SQL：

• get_purchase_order(order_id)：

  • 主表 purchase_order + 供应商表 purchase_supplier（LEFT JOIN）
  • 明细表 purchase_order_detail（获取 goods_id、order_count、real_price、purchase_last_price）

• query_product_archive(product_ids)：

  • 查询 goods 表（包含：name、category_name、spec、stock、rec_daily_sales、last_daily_sales、purchase_avg_price、purchase_last_price、total_sale_count、total_refund_count、total_service_count）

模拟数据：

当 USE_MOCK=true 时使用内存数据，包含 5 个商品（P001-P005）和 2 个采购单（PO-2026-001、PO-2026-002）

4.7.3 schemas.py — 数据模型

定义三个 Pydantic 模型：

• PurchaseItem：product_id、quantity、unit_price、purchase_last_price
• PurchaseOrder：order_id、supplier、order_date、total_amount、items
• ProductArchive：商品档案完整字段（13个字段）

4.7.4 test_db_connection.py — 数据库测试

• 独立脚本，执行 SELECT 1 测试连接
• 可选测试业务查询（查询 purchase_order 表）

4.7.5 mcp_server/.env

• USE_MOCK=false — 使用真实数据库
• DATABASE_URL=mysql+asyncmy://dp_read_user:xxx@aliyuncs.com:3306/dongpin_db
• 连接池参数配置

4.8 models/llm_gateway.py — LLM 接入

LLMGateway 类（单例模式）：

• 有 API Key → 创建 ChatOpenAI（temperature=0.1）
• 无 API Key → 创建 MockLLM（返回固定文本，用于测试）

五、数据流向

┌─────────────┐
│  Gradio UI   │  用户输入 order_id
└──────┬───────┘
       │
       ▼
┌──────────────────────────────────────────┐
│  LangGraph StateGraph                     │
│                                           │
│  ┌──────────────┐                         │
│  │ validate_input│ → MCPClient.get_xxx    │
│  └──────┬───────┘                         │
│         │                                 │
│         ▼                                 │
│  ┌────────────────────────┐               │
│  │ calculate_metrics_and_risk│ → 规则计算  │
│  └──────┬─────────────────┘               │
│         │                                 │
│         ▼                                 │
│  ┌──────────────┐                         │
│  │generate_report│ → LLM.ainvoke          │
│  └──────┬───────┘                         │
│         │                                 │
└─────────┼─────────────────────────────────┘
          │
          ▼
┌──────────────────┐
│  审核报告 + 归档   │  reports/{order_id}_{timestamp}_{decision}.md
└──────────────────┘

MCP 数据查询路径（真实模式）：

MCPClient._call_tool(tool_name, args)
    ↓
启动独立进程: python mcp_server/server.py
    ↓
stdin 写入 JSON-RPC request
    ↓
server.py → handle_request → db_queries.py → MySQLClient → asyncmy → MySQL
    ↓
stdout 返回 JSON-RPC response
    ↓
进程立即终止

六、配置说明

6.1 根目录 .env

# LLM
DS_KEY=sk-xxx
DS_BASE_URL=https://api.deepseek.com
DS_MODEL=deepseek-v4-flash

# 业务阈值
PRICE_PREMIUM_THRESHOLD=0.1      # 溢价阈值 10%
LOW_SALES_THRESHOLD=10           # 滞销品阈值（月销量）
HIGH_SALES_THRESHOLD=1000        # 热销品阈值（月销量）
MIN_TURNOVER_DAYS=15             # 最少周转天数
MAX_TURNOVER_DAYS=90             # 最多周转天数

# MCP
USE_MOCK_MCP=false               # false=真实数据库
MCP_SERVER_COMMAND=["python","mcp_server/server.py"]
MCP_REQUEST_TIMEOUT=30

6.2 mcp_server/.env

USE_MOCK=false
DATABASE_URL=mysql+asyncmy://dp_read_user:xxx@aliyuncs.com:3306/dongpin_db
DB_POOL_SIZE=10
DB_MAX_OVERFLOW=20

⚠️ 注意：根目录 .env 的 USE_MOCK_MCP 控制 MCPClient 是否启动真实 MCP Server；mcp_server/.env 的 USE_MOCK 控制 MCP Server 内部是否使用模拟数据。两者需协调：USE_MOCK_MCP=false + USE_MOCK=false = 真实数据库模式。

七、运行方式

# 安装依赖
pip install gradio langgraph langchain-openai sqlalchemy asyncmy dotenv markdown

# 启动服务
python app.py
# → http://0.0.0.0:7860

# 测试数据库连接
python mcp_server/test_db_connection.py

八、扩展设计

项目在 agents/__init__.py 中预留了注册中心模式：

# 数字员工注册中心（后续新增员工在此导入）
from .procurement import build_procurement_graph

__all__ = ["build_procurement_graph"]

新增数字员工（如：库存预警员、供应商评级员等）只需：

1. 在 agents/ 下新建子目录（如 agents/inventory_alert/）
2. 实现 graph_definition.py、nodes.py、prompts.py
3. 在 agents/__init__.py 中导入并注册
4. 在 app.py 中添加对应的 Gradio UI 入口

九、报告归档示例

报告自动归档到 reports/ 目录，文件名格式：

{order_id}_{YYYYMMDD}_{HHMMSS}_{决策}.md

示例：260701081805831_20260701_092906_通过.md

报告内容包含：

1. 最终决策标记（## 🏷️ 最终决策：**通过/驳回/人工复核**）
2. 概览（采购单号、供应商、总金额）
3. 商品明细审核表（14列完整数据）
4. 风险汇总
5. 最终结论

十、依赖清单

项目截图

{"timestamp": "2026-07-02T01:43:15.297447Z", "trace_id": "1fba887a-36cd-4417-86bf-ec129fa799a3", "level": "INFO", "module": "app", "func": "run_audit_async", "line": 35, "message": "收到采购审核请求, 采购单ID: 260629110324484, TraceID: 1fba887a-36cd-4417-86bf-ec129fa799a3"}
{"timestamp": "2026-07-02T01:43:15.321561Z", "trace_id": "1fba887a-36cd-4417-86bf-ec129fa799a3", "level": "INFO", "module": "nodes", "func": "validate_input", "line": 16, "message": "执行节点: 输入校验与数据获取"}
{"timestamp": "2026-07-02T01:43:15.321649Z", "trace_id": "1fba887a-36cd-4417-86bf-ec129fa799a3", "level": "INFO", "module": "mcp_client", "func": "_call_tool", "line": 20, "message": "启动 MCP Server (临时): python mcp_server/server.py"}
{"timestamp": "2026-07-02T01:43:15.792673Z", "trace_id": "1fba887a-36cd-4417-86bf-ec129fa799a3", "level": "ERROR", "module": "mcp_client", "func": "read_stderr", "line": 35, "message": "MCP Server stderr: 2026-07-02 09:43:15,792 - INFO - MCP Server started"}
{"timestamp": "2026-07-02T01:43:15.833645Z", "trace_id": "1fba887a-36cd-4417-86bf-ec129fa799a3", "level": "ERROR", "module": "mcp_client", "func": "read_stderr", "line": 35, "message": "MCP Server stderr: 2026-07-02 09:43:15,833 - INFO - Initializing MySQL connection pool"}
{"timestamp": "2026-07-02T01:43:16.293008Z", "trace_id": "1fba887a-36cd-4417-86bf-ec129fa799a3", "level": "INFO", "module": "mcp_client", "func": "_call_tool", "line": 20, "message": "启动 MCP Server (临时): python mcp_server/server.py"}
{"timestamp": "2026-07-02T01:43:16.553488Z", "trace_id": "1fba887a-36cd-4417-86bf-ec129fa799a3", "level": "ERROR", "module": "mcp_client", "func": "read_stderr", "line": 35, "message": "MCP Server stderr: 2026-07-02 09:43:16,553 - INFO - MCP Server started"}
{"timestamp": "2026-07-02T01:43:16.802567Z", "trace_id": "1fba887a-36cd-4417-86bf-ec129fa799a3", "level": "ERROR", "module": "mcp_client", "func": "read_stderr", "line": 35, "message": "MCP Server stderr: 2026-07-02 09:43:16,802 - INFO - Initializing MySQL connection pool"}
{"timestamp": "2026-07-02T01:43:17.119473Z", "trace_id": "1fba887a-36cd-4417-86bf-ec129fa799a3", "level": "INFO", "module": "nodes", "func": "calculate_metrics_and_risk", "line": 57, "message": "执行节点: 计算指标与风险判定"}
{"timestamp": "2026-07-02T01:43:17.120029Z", "trace_id": "1fba887a-36cd-4417-86bf-ec129fa799a3", "level": "INFO", "module": "nodes", "func": "generate_report", "line": 185, "message": "执行节点: LLM生成报告"}
{"timestamp": "2026-07-02T01:43:17.768366Z", "trace_id": "1fba887a-36cd-4417-86bf-ec129fa799a3", "level": "INFO", "module": "_client", "func": "_send_single_request", "line": 1740, "message": "HTTP Request: POST https://api.deepseek.com/chat/completions \"HTTP/1.1 200 OK\""}
{"timestamp": "2026-07-02T01:43:38.360949Z", "trace_id": "1fba887a-36cd-4417-86bf-ec129fa799a3", "level": "INFO", "module": "app", "func": "run_audit_async", "line": 57, "message": "审核完成, 决策: 驳回"}
{"timestamp": "2026-07-02T01:43:38.363435Z", "trace_id": "no-trace", "level": "INFO", "module": "app", "func": "save_report_to_local", "line": 87, "message": "报告已归档: /Users/t-mac/workspace/AI/digital_employee_platform1/reports/260629110324484_20260702_094338_驳回.md"}
{"timestamp": "2026-07-02T01:43:38.363544Z", "trace_id": "no-trace", "level": "INFO", "module": "app", "func": "run_and_archive_audit", "line": 179, "message": "报告已保存到本地: /Users/t-mac/workspace/AI/digital_employee_platform1/reports/260629110324484_20260702_094338_驳回.md"}

文档生成完毕。如有更新需求，可基于此文档持续迭代。