Unternehmensweites Multi-Agenten-Kollaborationsentscheidungssystem auf Basis von LangGraph- und MCP-Architektur

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

📌 Projektübersicht

Dieses Projekt ist eine hochverfügbare, hochskalierbare und vollständig asynchrone Multi-Agenten-Plattform für kollaborative Entscheidungen in anspruchsvollen Unternehmensszenarien. Das System orchestriert Workflows über den LangGraph StateGraph (StateGraph-Zustandsgraph), integriert den offenen Standard Anthropic Model Context Protocol (MCP) und kombiniert ein hybrides Kontextsystem mit hierarchischer Speicherverwaltung durch semantische Truncation. Über FastAPI + AsyncIO + SSE werden Token-Streaming im Millisekundenbereich und die vollständige Thought Chain in Echtzeit ausgeliefert.

🌟 Zentrale technische Kennzahlen

• 🎯 Tool-Routing-Genauigkeit: Dank strikter Function-Calling- und JSON-Schema-Validierung erreichen Tool-Auswahl und Argument-Extraktion eine Genauigkeit von 96,5 %.

• ⚡ TTFT (Time to First Token): Asynchrones, nicht blockierendes ereignisgesteuertes Scheduling komprimiert die Latenz bis zum ersten Token auf 210 ms.

• 🚀 Nebenläufige Durchsatzleistung: Leichtgewichtiges Coroutine-Scheduling; ein einzelner Knoten unterstützt stabil 120+ QPS.

• 📉 Token-Verbrauchsoptimierung: Semantische Truncation kombiniert mit Sliding-Window-Kontextmanagement reduziert den redundanten Token-Verbrauch in komplexen Multi-Turn-Dialogen um 38 %.

• 🛡️ Sicherheit und Compliance: Integrierter Human-in-the-loop (HITL)-Mechanismus und AST-Code-Sandbox mit 100 % Blockierung und manueller Freigabe für risikoreiche Vorgänge.

Related MCP server: MCP Business AI Transformation

🏗️ Gesamtarchitektur

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

🛠️ Detaillierte Beschreibung der vier Kernmodule

1. StateGraph-Workflow-Orchestrierung

• Multi-Agenten-Kollaborationsschleife:

  • PlannerAgent: Zerlegt komplexe Nutzeranfragen automatisch in eine geordnete Sub-Task-Topologie (SubTasks).

  • IntentRouterAgent: Führt anhand von Intent-Merkmalen und Tool-Metadaten ein hochpräzises Routing durch – mit einer Genauigkeit von 96,5 %.

  • ToolExecutorAgent: Führt Tool-Aufrufe parallel über asyncio.gather aus und fängt dabei Ausnahmen und Timeouts automatisch ab.

  • SelfRefineCriticAgent: Goats mehrdimensionale Qualitätsprüfungen der Ausführungsergebnisse (Datenvollständigkeit, Schema-Konsistenz, logische Halluzinationen) und stößt bei Unterwertung die dynamische Rücksetzung zum Planner im StateGraph an.

• Human-in-the-loop (HITL) – manuelle Intervention:

  • Automatische Unterbrechung für sensible Tools wie Datenbankschreiboperationen (sql_execute_dml) oder Dateisystemänderungen.

  • Mitigiert den Zustandsgraphen und speichert den Kontext-Snapshot im Checkpointer. Nach der Freigabe/Ablehnung/annotierten Anpassung durch einen Administrator über das Frontend-Modabcam oder die /api/hitl/approve-Schnittstelle wird die Ausführung nahtlos fortgesetzt.

2. MCP-Protokoll und Erweiterung um 8+ Tool-Quellen

• Einhaltung des Anthropic-MCP-Protokollstandards (JSON-RPC 2.0) zur Entkopplung von Tool- und Modellseite.

• Integriert 8+ Kategorien an Standard-Tool-Quellen:

  1. sql_query_tool: Strukturierte SQL-Berichtsabfragen und multidimensionale Aggregationsstatistiken.

  2. sql_execute_dml: Datenbank-Insert-/Update-Operationen (als is_sensitive=True gekennzeichnet).

  3. web_search_tool: Echtzeit-Websuche nach aktuellen Informationen und technischen Spezifikationen.

  4. python_sandbox: Isolierte Ausführungsumgebung mit sicherheitsorientierter Abscharfung auf Basis des Python-AST; kritische Operationen wie os/subprocess/socket sind vollständig verboten.

  5. knowledge_rag_tool: Enterprise-Wissensbasis mit BM25+ und vektorierter Hybrid-Retrieval.

  6. chart_generator: Automatische Erzeugung von ECharts Balken-/Linien-/Kreisdiagrammen sowie Mermaid-Fluxdiagrammkonfigurationen.

  7. file_system_tool: Sandboxige, sichere Datei-Lese- und Schreibzugriffe sowie Verzeichnisanalysen.

  8. data_cleaner_tool: Intelligente Rekonstruktion und Reparatur beschädigter Markdown-/JSON-Daten.

  9. http_request_tool: Down Connection zu externen REST-APIs.

• Unterstützt den Betrieb als eigenständiger Serverprozess (examples/run_mcp_standalone.py) und nahtloses Einbinden in Claude Desktop oder Cursor.

3. Hybrides Kontextsystem und hierarchische Speicherverwaltung

• Kurzzeitige Checkponts (Checkpointer): Haltige Redis-Hash-Tabelle und SQLite-Doppelpersistenz; unterstützt Rückverfolgbarkeit mehrerer Gesprächsrufzustände, Branch-Relay und Fehlerwiederherstellung.

• Langzeitliches Benutzerprofil: Automatische Pflege von Technologie-Stack-Präferenzen, Output-Stil-Beschränkungen und historischem Entscheidungsverhalten basierend auf der Benutzer-ID; Kontext wird nach Bedarf beim Start der Agents injiziert.

• Token-Redundanz-Kompressionsalgorithmus (Context Compressor):

  • Sliding-Window-Mechanismus: Festes Behalten der Systembefehle und der letzten $K$ Gesprächsrunden.

  • Multi-Turn-Cutting (Semantic Truncation): Automatisches Extrahieren des Kern-Schemas und von Zusammenfassungen aus veralteten, langen Tool-Ausgaben (z. B. SQL-Ausgabe mit hunderte Sätzen) und Reduktion des Token-Overheads um über 38 % in mehrläufigen Dialogdaten.

4. Produktionsreifes Streaming-Inference und Nebenläufigkeitsoptimierung (FastAPI + AsyncIO + SSE)

• Vollständig asynchrone, nicht blockierende Architektur: FastAPI in Kombination mit der AsyncIO-Event-Loop für hohen Durchsatz (120+ QPS).

• Detailliertes SSE-Ereignistreaming:

  • thought: Echtzeit-Ausgabe der Überlegungen und Entscheidungslogik jedes Agent-Knotens.

  • tool_start / tool_end: Live-Anzeige, der Eingabeparameter der Tool-Aufrufe und der Ausführungszeiten.

  • hitl_request: Auslösen des Frontend-Freigabe-Dialogfelds.

  • token: Wiedergabe-Modus für die abschließende Antwort (Token für Token).

  • done: Inklusive vollständiger Token-Nutzung und Optimierungsstatistiken.

• Zero-Dependency Intelligent Mock und nahtloser Wechsel zum realen Modell: Enthält einen hochwertigen Mock-Modell-Treiber (simulierte Latenz von 210 ms bis zum ersten Token). Durch Konfigurieren von OPENAI_API_KEY in der .env-Datei kann per Ein-Klick zu GPT-4o. DeepSeek-V3/R1, Claude 3.5 oder lokalem Ollama gewechselt werden.

📂 Projektverzeichnisstruktur

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 压缩算法测试

🚀 Schnellstart-Anleitung

Variante 1: Lokale Ausführung in virtueller Umgebung (empfohlen)

1. Umgebungsvariablen konfigurieren:

   cp .env.example .env

   (Standard: eingebautes Mock-Modell, sofort einsatzbereit ohne API-Key...)

2. Abhängigkeiten installieren:

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

3. FastAPI-Webdienst im Async-Modus starten:

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

   • 🌐 Web-Interaktionskonsole: Öffnen Sie im Browser http://localhost:8000

   • 📑 Swagger-API-Dokumentation: Besundes sie http://localhost:8000/docs

Variante 2: Ein-Klick-Containerbbereitung mit Docker Compose

docker-compose up -d --build

Dieser Befehl startet automatisch den FastAPI-Backend-Container sowie den persistenten Redis-Checkpoint-Dienst.

💻 Klassische Szenarien und Skript-Demos

1. Terminal-Kommandozeilen-Multi-Agenten-Kollaborationsdemo

python examples/cli_demo.py

Live-Anzeige der Planung und Aufgabenverteilung, MCP-Verteilungsprozesse sowie des Token-Einsparungseffekts der Agents im Terminalkonsolen-Output.

2. Benchmark der Token-Redundanz-Kompression

python examples/evaluate_token_saving.py

Beispiel mit echten Ergebnissen:

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

3. Ausführen eines eigenständigen MCP-Servers (z. B. für Claude Desktop / Cursor)

python examples/run_mcp_standalone.py

🧪 Automatisierte Tests

Führe die vollständige Testsuite mit Unit- und End-to-End-Integrationstests der Zustandsmaschine aus:

pytest -v

Testergebnisse:

============================= 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 ==============================

📡 Kern-API-Spezifikationen

📄 Lizenz

Dieses Projekt steht unter der MIT License und ist Open Source lizenziert.