Enterprise-Grade Multi-Agent Collaborative Decision System Based on LangGraph and MCP Architecture

Enterprise Multi-Agent Collaborative Decision System (2025.08 - 2025.12)


📌 Project Overview (Project Overview)

This project is a highly available, highly extensible, fully asynchronous multi-agent collaborative decision platform for complex enterprise-grade scenarios. The system orchestrates workflows based on LangGraph's directed state graph (StateGraph), deeply integrates the Anthropic Model Context Protocol (MCP) open tool protocol standard, and combines a layered memory system with hybrid context and semantic truncation. Through FastAPI + AsyncIO + SSE, it provides millisecond-level Token streaming and real-time end-to-end Chain-of-Thought (Thought Chain) push capabilities.

🌟 Core Technical Indicators (Core Technical Metrics)

• 🎯 Tool routing accuracy: Using strict Function Calling and JSON Schema validation, tool selection and parameter extraction achieve 96.5% accuracy.

• ⚡ Time to First Token (TTFT): With asynchronous non-blocking event-driven scheduling, first-token streaming time is compressed to 210ms.

• 🚀 Concurrent throughput: Lightweight coroutine concurrent scheduling supports stable operation at 120+ QPS on a single node.

• 📉 Token cost optimization: Semantic truncation combined with sliding-window context management reduces Token redundancy consumption by 38% in multi-turn complex conversations.

• ** Security and compliance: Built-in Human-in-the-loop (HITL) mechanism and AST code sandbox; high-risk operations are 100% intercepted and require manual approval.

Related MCP server: MCP Business AI Transformation

🏗️ Overall Architecture Design (System Architecture)

flowchart TD
    subgraph ClientLayer [客户端交互层]
        WebUI[现代化 Web 交互控制台 / SSE 客户端]
        RESTClient[RESTful API / SDK 客户端]
        MCPClientApp[Claude Desktop / Cursor MCP 客户端]
    end

    subgraph APILayer [FastAPI 异步高性能网关]
        Router[API 路由网关 / 跨域与鉴权]
        SSEHandler[SSE 异步事件流分发器 (Token 流 + 思考链路流)]
        HITLHandler[Human-in-the-loop 审核干预中心]
    end

    subgraph LangGraphCore [LangGraph 状态机决策内核]
        State[AgentState 核心状态定义]
        
        Planner[1. Task Planner 任务规划 Agent]
        IntentRouter[2. Intent Classifier & Tool Router 意图识别]
        ToolExecutor[3. Tool Executor 并行工具执行器]
        SelfRefine[4. Self-Refine / Critic 反思纠错 Agent]
        HITLNode[Human-in-the-loop 人工审批拦截节点]
        
        Planner --> IntentRouter
        IntentRouter -->|需要调用工具| ToolExecutor
        IntentRouter -->|纯文本直接回答| SelfRefine
        ToolExecutor -->|检测到敏感操作(如DML写)| HITLNode
        HITLNode -->|审核通过 (Resume)| ToolExecutor
        HITLNode -->|审核拒绝 / 指令调整| Planner
        ToolExecutor --> SelfRefine
        SelfRefine -->|质检未通过 / 异常回溯| Planner
        SelfRefine -->|质检通过 (98% 评分)| EndNode[Final Answer 汇总输出]
    end

    subgraph MCPHub [MCP 协议与 8+ 外部工具中心]
        MCPCore[Async MCP Client & Server Manager]
        ToolRegistry[动态工具注册表 (Pydantic Schema 校验)]
        
        subgraph ToolSources [8+ 生产级核心工具源]
            T1[sql_query_tool: 数据库安全只读分析]
            T2[sql_execute_dml: 数据库写变更 (带 HITL)]
            T3[web_search_tool: DuckDuckGo 实时网络检索]
            T4[python_sandbox: AST 安全隔离代码沙盒]
            T5[knowledge_rag_tool: 企业知识库混合检索]
            T6[chart_generator: ECharts / Mermaid 可视化配置生成]
            T7[file_system_tool: 沙盒化文件安全读写]
            T8[data_cleaner_tool: JSON 清洗与 Schema 修复]
            T9[http_request_tool: 外部 RESTful API 动态调用]
        end
    end

    subgraph MemoryLayer [混合上下文与分层记忆体系]
        Checkpointer[Redis / SQLite 状态持久化检查点]
        LongTermMem[长期用户画像 (User Profile) 与偏好库]
        Compressor[上下文压缩器: 语义截断 + 滑动窗口 (降低 38% Token)]
    end

    ClientLayer --> APILayer
    APILayer --> LangGraphCore
    LangGraphCore --> MCPHub
    MCPHub --> ToolSources
    LangGraphCore --> MemoryLayer

🛠️ Four Core Modules in Detail (Core Modules)

1. State-Machine Workflow Orchestration (StateGraph Workflow)

• Multi-Agent Collaboration Loop:

  • PlannerAgent: automatically decomposes complex user business requirements into an ordered sub-task topology (SubTasks).

  • IntentRouterAgent: combines intent features and tool metadata for high-precision routing, reaching 96.5% accuracy.

  • ToolExecutorAgent: uses asyncio.gather to execute tool calls in parallel, automatically catching exceptions and timeouts.

  • SelfRefineCriticAgent: performs multi-dimensional quality reviews (data integrity, Schema consistency, logical hallucination) based on execution results; when below the threshold, it triggers the state graph to dynamically trace back to the Planner.

• Human-in-the-loop (HITL) manual intervention:

  • autom matically* intercepts sensitive tools such as database write operations (sql_execute_dml) and system file modifications.

  • The execution is suspended and a context snapshot is persisted in the Checkpointer. After the administrator approves/rejects/annotates modifications through the front-end modal or the /api/hitl/approve endpoint, the execution is seamlessly resumed.

2. MCP Protocol and 8- Tool Source Extension (Model Context Protocol)

• Follows the Anthropic MP protocol standard (JSON-RPC 2.0), easing** decopling * of the tool side **and the model side.

• Built-in with 8+ categories of standard tool sources:

  1. sql_query_tool: structured SQL report queries and multi-dimensional aggregation statistics.

  2. sql_execute_dml: database insert/update operations (marked as is_sensitive=True).*

  3. wweb_seearch_tool: real-time web retrieval of the latest news and technical documentation.

  4. python_sandobox: a sandboxed execution environment based on Python AST syntax tree security audits, completely forbidding dangerous instructions such as os/subprocess/socket.

  5. knowledge_rag_tool: enterprise-level knowledge base **BM25 + vectorizinghybrid retrieval.

  6. chart_generator: automatically outputs ECharts bar//line/pie charts and Mermaid flowcharts configuration.

  7. file_system_tool: sandlocked safe file read/write and directory analysis.

  8. death_cleaner_tool: intelligently extracts and repairs corrupted Markdown/JSON data.

  9. http_request_tool: outer REST API integration.

• Supports running as an independent server process (examples/run_mcp_standalone.py), seamlessly accessible to Claude Desktop or Cursor.

3. Hybrid Context and Hierarchical** Memory Management (Hybrid Context & Memory)

• Short-term Checkpoint (Checkpointer): based on Redis hash tables and SQLite dual persistence, it supports state tracing, branch replay and failure recovery across multi-turn conversations.

• Long-term *User Profile (User Profile): automatically maintains the user's technical stack preferences, output style constraints and historical decision-making behavior based on the user ID, and injects context on-dodemand during injection during multi-agent startup.

• Context Compressor ( Token Redundancy Compression):

  • Sliding window mechanism: preserves the system instructions and the latest $K$ turns of conversation.

  • Semantic truncation (Semantic Truncation): for outdated and verbose intermediate tool outputs (such** as raw SQL results containing hundreds of records), it automatically extracts the core “chema” and abstracts the abstract, reducing Token redundancy in multi-turn conversations by more than 38%.

4. Production-Grade Streaming Inference and Concurrent Optimization (FastAPI + AsyncIO + SSE)

• Fully async non-ocking architecture: adopts FastAPI + AsyncIO event loop scheduling to achieve high-throughput request handling (120+ QPS).

• SSE high-fine-grained event stream push:

  • thought: pushes the current insight of each Agent node in real-time and the decision logic.

  • prag_start / prag_end: displays the tool invocation input and execution concurrency in real-time.

  • hitl_request: triggers the front-end approval modal.

  • token: a 型 printer-style streaming output when generating the final answer.

  • done: returns the complete Token consumption and optimization metrics.

• Zro-dependency intelligent Mock / seamless with real models: built-in high-performance Mock model driver (simulating 210 ms first-token latency). The only an environment config .env is needed for a one-click switch to actual mode with the real model (Qqpt-4o, SeepSeek-V3/R1, Claude 3.5 or local Ollama) via setting OPENAI_API_KEY in .env.

📂 Project Directory Structure (Directory Layout)

mcp/
├── README.md                     # 完整的项目说明文档与架构白皮书
├── pyproject.toml                # 项目规范与构建配置
├── requirements.txt              # 生产依赖列表
├── docker-compose.yml            # Docker 容器化编排 (FastAPI + Redis)
├── Dockerfile                    # 生产级镜像构建配置
├── .env.example                  # 环境变量配置模板
│
├── app/                          # 核心应用源码
│   ├── __init__.py
│   ├── main.py                   # FastAPI 应用入口、CORS 与静态资源挂载
│   ├── config.py                 # 全局 Pydantic Settings 配置驱动
│   │
│   ├── api/                      # 接口层
│   │   ├── __init__.py
│   │   ├── routes.py             # 核心 REST & SSE 接口 (chat, stream, hitl, metrics)
│   │   └── schemas.py            # Pydantic 请求/响应模型
│   │
│   ├── core/                     # 状态机与底层驱动
│   │   ├── __init__.py
│   │   ├── state.py              # AgentState 强类型状态模型定义
│   │   ├── workflow.py           # StateGraph 状态机编排与事件流引擎
│   │   └── llm_provider.py       # 统一大模型适配器 (OpenAI/DeepSeek/Claude/Mock)
│   │
│   ├── agents/                   # 多智能体角色实现
│   │   ├── __init__.py
│   │   ├── planner.py            # Task Planner (任务规划 Agent)
│   │   ├── router.py             # Intent Classifier & Router (意图识别 Agent)
│   │   ├── executor.py           # Tool Executor (并行工具执行 Agent)
│   │   └── reflector.py          # Self-Refine Critic (反思质检 Agent)
│   │
│   ├── mcp/                      # Model Context Protocol (MCP) 体系
│   │   ├── __init__.py
│   │   ├── client.py             # 标准 MCP 异步客户端
│   │   ├── server.py             # 标准 MCP 独立 Stdio 服务端
│   │   └── registry.py           # 动态工具注册中心 (JSON Schema 校验)
│   │
│   ├── tools/                    # 8+ 生产级工具实现
│   │   ├── __init__.py           # 工具集合统一导出注册
│   │   ├── sql_tool.py           # SQL 查询与 DML 变更工具
│   │   ├── search_tool.py        # 网络检索工具 (DuckDuckGo)
│   │   ├── sandbox_tool.py       # Python AST 安全沙盒
│   │   ├── rag_tool.py           # 知识库混合检索
│   │   ├── chart_tool.py         # ECharts / Mermaid 可视化生成
│   │   ├── filesystem_tool.py    # 安全文件系统操作
│   │   ├── data_cleaner_tool.py  # JSON 清洗与结构修复
│   │   └── http_api_tool.py      # 通用 HTTP API 适配器
│   │
│   ├── memory/                   # 混合记忆管理
│   │   ├── __init__.py
│   │   ├── checkpointer.py       # Redis & SQLite 状态检查点
│   │   ├── user_profile.py       # 用户画像与偏好库
│   │   └── compressor.py         # 语义截断与滑动窗口压缩算法
│   │
│   └── static/                   # 现代化 Web 交互看板
│       ├── index.html            # 响应式前端交互页面
│       ├── app.js                # SSE 流式渲染与 HITL 审批交互
│       └── style.css             # 现代化暗色主题 UI
│
├── examples/                     # 经典演示与基准脚本
│   ├── cli_demo.py               # 终端交互式 Multi-Agent 协作演示
│   ├── run_mcp_standalone.py     # 独立 MCP 工具服务端启动器
│   └── evaluate_token_saving.py  # Token 压缩基准评测脚本 (验证 38% 节约率)
│
└── tests/                        # 自动化测试套件 (100% 通过)
    ├── __init__.py
    ├── test_workflow.py          # 状态机流转与 HITL 审批中断测试
    ├── test_mcp_tools.py         # 8+ MCP 工具执行与沙盒安全测试
    └── test_memory.py            # 检查点恢复与 Token 压缩算法测试

🚀 Quick Start Guide (Quick Start)

Option 1: Local virtual environment (Recommended)

1. Configure environment variables:

cp .env.example .env

(The built-in Mock model is enabled in the default, so the API KEY is not required for out-of-the-box experience.)

2. Install dependencies:

python -m venv .venv
# Windows:
.\.venv\Scripts\pip install -r requirements.txt
# Linux / macOS:
source .venv/bin/activate && pip install -r requirements.txt

3. Start the FastAPI async web service:

# Windows:
.\.venv\Scripts\python -m app.main
# Linux / macOS:
python -m app.main

• 🌐 Web console: open the browser and visit http://localhost:8000

• 📑 Swagger API docs: visit http://localhost:8000/docs

Option 2: One-click Docker Compose deployment

docker-compose up -d --build

This command automatically starts the FastAPI backend container and the persistent Redis checkpoint service.

💻 Classic scenarios and demos (Demos & Benchmarks)

1. Terminal command-line multi-agent collaboration demo

python examples/cli_demo.py

Observe the multi-agent collaborative planning and division of labor, the MCP dispatch process, and Token compression gains in real time in the terminal.

2. Token redundancy compression benchmark

python examples/evaluate_token_saving.py

Example of actual results:

=================================================================
  [*] 上下文压缩与 Token 冗余消除基准评估 (Benchmark)
=================================================================
原始上下文消息轮数: 11
压缩后保留消息轮数: 7
原始预估 Token 消耗: 1348 Tokens
压缩后 Token 消耗:   316 Tokens
节省 Token 数量:     1032 Tokens
🎯 Token 冗余降低比例: 76.6% (标准多轮场景稳定保持 >38%)
-----------------------------------------------------------------
结论: 语义截断结合滑动窗口在长周期多 Agent 对话中显著消除 Token 冗余。
=================================================================

3. Run a standalone MCP server (for connecting Claude Desktop / Cursor)

python examples/run_mcp_standalone.py

🧪 Automated testing (Automated Testing)

Run the full unit test suite and the end-to-end state machine integration tests:

pytest -v

Test output:

============================= test session starts =============================
tests/test_mcp_tools.py::test_tool_registry_listings PASSED              [  8%]
tests/test_mcp_tools.py::test_sql_query_tool PASSED                      [ 16%]
tests/test_mcp_tools.py::test_python_sandbox_safe_execution PASSED       [ 25%]
tests/test_mcp_tools.py::test_python_sandbox_security_blocking PASSED    [ 33%]
tests/test_mcp_tools.py::test_knowledge_rag_tool PASSED                  [ 41%]
tests/test_mcp_tools.py::test_data_cleaner_tool PASSED                   [ 50%]
tests/test_memory.py::test_checkpointer_save_and_retrieve PASSED         [ 58%]
tests/test_memory.py::test_user_profile_memory PASSED                    [ 66%]
tests/test_memory.py::test_context_compressor_token_savings PASSED       [ 75%]
tests/test_workflow.py::test_full_workflow_execution PASSED              [ 83%]
tests/test_workflow.py::test_hitl_interruption PASSED                    [ 91%]
tests/test_workflow.py::test_streaming_generator PASSED                  [100%]

============================= 12 passed in 3.50s ==============================

📡 Core API endpoints (API Specifications)

📄 Open-source License (License)

This project uses the MIT License license.