Корпоративная многоагентная система совместного принятия решений на основе LangGraph и MCP

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

📌 Обзор проекта (Project Overview)

Данный проект — это высокодоступная, высокомасштабируемая, полностью асинхронная многоагентная платформа совместного принятия решений для сложных корпоративных сценариев. Система оркестрирует рабочие процессы на основе направленного графа состояний LangGraph (StateGraph), глубоко интегрирует открытый стандарт протокола инструментов Anthropic Model Context Protocol (MCP), объединяет гибридный контекст и многоуровневую систему памяти с семантическим усечением, а через FastAPI + AsyncIO + SSE обеспечивает потоковую передачу токенов с задержкой на уровне миллисекунд и отправку полной цепочки рассуждений (Thought Chain) в реальном времени.

🌟 Ключевые технические показатели

• 🎯 Точность маршрутизации инструментов: благодаря строгой проверке Function Calling и JSON Schema точность выбора инструментов и извлечения параметров достигает 96.5%.

• ⚡ Время до первого токена (TTFT): асинхронная неблокирующая событийно-ориентированная диспетчеризация сокращает время потоковой выдачи первого токена до 210ms.

• 🚀 Пропускная способность при параллельной работе: лёгкое планирование на корутинах обеспечивает стабильную работу одного узла на уровне 120+ QPS.

• 📊 Оптимизация расхода токенов: семантическое усечение в сочетании с управлением контекстом через скользящее окно снижает избыточный расход токенов в сложных многораундовых диалогах на 38%.

• 🛡️ Безопасность и соответствие требованиям: встроены механизм Human-in-the-loop (HITL) и песочница кода на основе AST; высокоопасные операции блокируются в 100% случаев и отправляются на ручное утверждение.

Related MCP server: MCP Business AI Transformation

🏗️ Общая архитектура системы (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

🛠️ Детальный разбор четырёх ключевых модулей (Core Modules)

1. Оркестрация рабочих процессов на StateGraph (StateGraph Workflow)

• Замкнутый многоагентный цикл совместной работы:

  • PlannerAgent: автоматически декомпозирует сложный бизнес-запрос пользователя в упорядоченную топологию подзадач (SubTasks).

  • IntentRouterAgent: выполняет высокоточную маршрутизацию на основе признаков намерений и метаданных инструментов с точностью 96.5%.

  • ToolExecutorAgent: параллельно выполняет вызовы инструментов через asyncio.gather, автоматически перехватывая исключения и таймауты.

  • SelfRefineCriticAgent: проводит многомерную проверку качества по результатам выполнения (целостность данных, согласованность Schema, логические галлюцинации); при результате ниже порогового срабатывает динамический возврат графа состояний к Planner.

• Ручное вмешательство через Human-in-the-loop (HITL):

  • Автоматическая блокировка «чувствительных» инструментов: операций записи в БД (sql_execute_dml), изменения системных файлов и т.п.

  • Граф состояний приостанавливается, а снимок контекста сохраняется в Checkpointer; после выбора администратором «одобрить / отклонить / добавить примечание» во всплывающем окне фронтенда или через API /api/hitl/approve выполнение бесстыково возобновляется.

2. Протокол MCP и расширение 8+ источников инструментов (Model Context Protocol)

• Соблюдается стандарт протокола Anthropic MCP (JSON-RPC 2.0), разделяющий уровень инструментов и уровень моделей.

• Встроено более 8 категорий типовых источников инструментов:

  1. sql_query_tool: структурированные SQL-запросы для отчётов и многомерная агрегированная статистика.

  2. sql_execute_dml: операции вставки/обновления в БД (помечены как is_sensitive=True).

  3. web_search_tool: поиск в сети свежих новостей и технических спецификаций в реальном времени.

  4. python_sandbox: изолированная среда выполнения с проверкой безопасности на основе синтаксического дерева Python AST; полностью запрещены опасные команды os, subprocess, socket.

  5. knowledge_rag_tool: корпоративный гибридный поиск по базе знаний — BM25 + векторный поиск.

  6. chart_generator: автоматическая генерация конфигураций диаграмм ECharts (столбчатых/линейных/круговых) и Mermaid-блок-схем.

  7. file_system_tool: безопасное чтение/запись файлов и анализ каталогов в песочнице.

  8. data_cleaner_tool: интеллектуальное извлечение и исправление повреждённых данных Markdown/JSON.

  9. http_request_tool: динамическое подключение к внешним REST API.

• Поддерживается запуск в качестве отдельного серверного процесса (examples/run_mcp_standalone.py) с бесшовным подключением к Claude Desktop или Cursor.

3. Гибридный контекст и многоуровневое управление памятью (Hybrid Context & Memory)

• Краткосрочные контрольные точки (Checkpointer): двойное персистентное хранение на основе хэш-таблиц Redis и SQLite, поддержка трассировки состояний многораундовых диалогов, воспроизведения веток и восстановления после сбоев.

• Долгосрочный пользовательский профиль (User Profile): по ID пользователя автоматически поддерживаются предпочтения по технологическому стеку, ограничения по стилю вывода и история принятия решений; контекст при необходимости внедряется при запуске множества агентов.

• Алгоритм сжатия избыточных токенов (Context Compressor):

  • Механизм скользящего окна: постоянное сохращение системных инструкций и последних $K$ раундов диалога.

  • Семантическое усечение (Semantic Truncation): для устаревших и объёмных промежуточных результатов работы инструментов (например, исходных результа-выводов SQL, содержащих сотни записей) автоматически извлекаются ключевые Schema и резюме, снижая избыточный расход токенов многораундового диалога более чем на 38%.

4. Продакшн-потоковый вывод и оптимизация параллелизма (FastAPI + AsyncIO + SSE)

• Полностью асинхронная неблокирующая архитектура: на базе FastAPI и цикла событий AsyncIO обеспечивается высокопропускная обработка запросов (120+ QPS).

• Детализированная отправка событий SSE:

  • thought: мгновенная отправка процесса рассуждений и логики принятия решений каждого узла Agent.

  • tool_start / tool_end: наглядное отображение параметров вызова инструмента и времени выполнения.

  • hitl_request: активация всплывающего диалога утверждения во фронтенде.

  • token: посимвольная потоковая выдача в стиле «печатающей машинки» при формировании конечного ответа.

  • done: завершение с полной статистикой расхода токенов и эффективности оптимизации.

• Бесшовное переключение между «умным» Mock с нулевой зависимостью и реальными моделями: по умолчанию встроена высокопроизводительная Mock-модель (симуляция задержки первого токена 210ms); достаточно прописать OPENAI_API_KEY в конфигурации .env, чтобы одним движением переключиться на GPT-4o, DeepSeek-V3/R1, Claude 3.5 или локальную Ollama.

📂 Структура каталогов проекта (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)

Способ 1: запуск в локальной виртуальной среде (рекомендуется)

1. Настройте переменные окружения:

   cp .env.example .env

   (По умолчанию используется встроенная высокопроизводительная Mock-модель — можно работать «из коробки» без настройки API Key.)

2. Установите зависимости:

   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:

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

   • 🌐 Веб-консоль взаимодействия: откройте в браузере адрес http://localhost:8000

   • 📑 Swagger-документация API: перейдите по адресу http://localhost:8000/docs

Способ 2: развёртывание одним нажатием через Docker Compose

docker-compose up -d --build

Эта команда автоматически запустит контейнер бэкенда FastAPI и сервис персистентных контрольных точек Redis.

💻 Демонстрация классических сценариев и скриптов (Demos & Benchmarks)

1. Демонстрация многоагентного взаимодействия в терминальной командной строке

python examples/cli_demo.py

В реальном времени можно наблюдать планирование и разделение задач многоагентной системой в консоли терминала, процесс диспетчеризации MCP и эффект сжатия токенов.

2. Бенчмарк-оценка сжатия избыточных токенов

python examples/evaluate_token_saving.py

Пример реальных результатов:

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

3. Запуск отдельного MCP-сервера (для подключения Claude Desktop / Cursor)

python examples/run_mcp_standalone.py

🧪 Автоматизированное тестирование (Automated Testing)

Запуск полного набора модульных тестов и сквозных интеграционных тестов конечного автомата:

pytest -v

Результаты выполнения тестов:

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

📡 Описание основных API-интерфейсов (API Specifications)

📄 Лицензия с открытым исходным кодом (License)

Проект выпущен под лицензией с открытым исходным кодом MIT License.