Sistema empresarial de toma de decisiones colaborativa con múltiples agentes basado en LangGraph y arquitectura MCP

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

📌 Resumen del proyecto (Project Overview)

Este proyecto es una plataforma de toma de decisiones colaborativa basada en múltiples agentes, altamente disponible, altamente extensible y totalmente asíncrona, orientada a escenarios empresariales complejos. El sistema organiza flujos de trabajo mediante grafos de estados dirigidos de LangGraph (StateGraph), se integra profundamente con el estándar de protocolo abierto de herramientas Model Context Protocol (MCP) de Anthropic, combina un sistema de memoria jerárquica con contexto mixto y truncamiento semántico, y ofrece flujo de tokens a nivel de milisegundos y notificación en tiempo real de la cadena de pensamiento completa (Thought Chain) a través de FastAPI + AsyncIO + SSE.

🌟 Indicadores técnicos clave

• 🎯 Precisión de enrutamiento de herramientas: mediante la validación estricta de Function Calling y JSON Schema, la selección de herramientas y la extracción de parámetros alcanzan una precisión del 96.5%.

• ⚡ Tiempo hasta el primer token (TTFT): con una programación de eventos asíncrona y no bloqueante, la salida en streaming del primer token se reduce a 210ms.

• 🚀 Rendimiento concurrente: con programación concurrente ligera de corrutinas, cada nodo es capaz de mantener de forma estable 120+ QPS de concurrencia.

• 📉 Optimización del consumo de tokens: mediante el truncamiento semántico combinado con la gestión de contexto de ventana deslizante, reduce el desperdicio de tokens redundantes en 38% en conversaciones complejas de varias vueltas.

• 🛡️ Seguridad y cumplimiento: incorpora el mecanismo Human-in-the-loop (HITL) y un sandbox de código AST, que garantiza la interceptación y aprobación humana del 100% de las operaciones de alto riesgo.

🏗️ Diseño general de la arquitectura (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

🛠️ Detalle de los cuatro módulos principales (Core Modules)

1. Orquestación de flujos de trabajo en máquina de estados (StateGraph Workflow)

• Ciclo cerrado de colaboración multiagente:

  • PlannerAgent: descompone automáticamente las necesidades empresariales complejas del usuario en una topología ordenada de subtareas (SubTasks).

  • IntentRouterAgent: combina las características de la intención y los metadatos de las herramientas para el enrutamiento de alta precisión, con una exactitud del 96.5%.

  • ToolExecutorAgent: implementa las llamadas a las herramientas en paralelo usando asyncio.gather y captura automáticamente excepciones y tiempos de espera.

  • SelfRefineCriticAgent: realiza una revisión de calidad multidimensional (integridad de datos, consistencia de Schema, alucinaciones lógicas) basada en los resultados de ejecución y, si el resultado se encuentra por debajo del umbral, activa en la red de redes del grafo de estados una reversión dinámica a Planner.

• Intervención humana Human-in-the-loop (HITL):

  • Intercepta automáticamente las herramientas sensibles como las operaciones de escritura en bases de datos (sql_execute_dml) y la modificación de archivos del sistema.

  • Pausa el grafo de estados y persiste una instantánea del contexto en el Checkpointer, esperando que el administrador apruebe, rechace o modifique con comentarios posteriores.

2. Protocol MCP y extensión de fuentes de herramientas 8+ (Model Context Protocol)

• Sigue la norma de protocol abierto de Anthropic MCP (JSON-RPC 2.0) y acopla de forma desacoplada la capa de herramientas y el modelo.

• Incluye 8+ tipos de fuentes de herramientas estándar integradas:

  1. sql_query_tool: consulta de informes SQL estructurados y agregación estadística multidimensional.

  2. sql_execute_dml: operaciones de inserción/actualizacifra en la base de datos (marcada como is_sensitive=True).

  3. web_search_tool: búsqueda en tiempo real de noticias y especificaciones técnicas de la web.

  4. python_sandbox: entorno de ejecución aislado con revisión de seguridad basada en el árbol de sintaxis AST de Python, con bloqueo total de directivas peligrosas como os, y subprocess y socket.

  5. knowledge_rag_tool: recuperación mixta de contexto empresarial con BM25 + vectores.

  6. chart_generator: genera automáticamente configuraciones de gráficos de barras/líneas/tartas de ECharts y diagramas de flujo Mermaid.

  7. file_system_tool: lectura/escritura segura de archivos en sandbox y análisis de directorios.

  8. data_cleaner_tool: extrae y repara automáticamente datos Markdown/JSON dañados.

  9. http_request_tool: integración dinámica con API REST externas.

• Soporta ejecución como proceso de servidor independiente (examples/run_mcp_standalone.py) y puede montarse sin problemas en Claude Desktop o Cursor.

3. Contexto mixto y gestión de memoria jerárquica (Hybrid Context & Memory)

• Checkpoint a corto plazo (Checkpointer servicios, es una herramienta de: basado en la persistencia dual de hash tables de Redis y SQLite, soporta trazabilidad de estados, reproducción de ramas y recuperación de fallos en todas las conversaciones multi-turno.

• Perfil de usuario a largo plazo (User Profile): mantiene automáticamente las preferencias de pila tecnológica, las restricciones de estilo de salida y el historial de decisiones del usuario según el identificador del usuario, e inyecta contexto bajo demanda al inicio de los múltiples agentes.

• Algoritmo de compresión de redundancia de tokens (Context Compressor):

  • Mecanismo de ventana deslizante: mantiene fijas las instruciones del sistema y las últimas $K$ réplicas de la conversación.

  • Trucamiento semántico (Semantic Truncación): extrae automáticamente el esquema núcleo y resúmenes de las resultas formes de herramientas que han caducado o son extensos (como los resultos crudos de SQL con cientos de registros), reduciendo más de un 38% los tokens redundantes en conversaciones multi-turn.

4. Razamiénto en streaming a nivel de producción y optimización de concurrencia (FastAPI + AsyncIO + SSE)

• Arquitectura totalmente asíncrona y no bloqueante: usa el bucle de eventos de FastAPI y AsyncIO para lograr un alto rendimiento (120+QPS).

• Granularidad fina en eventos SSE:

  • thought: envía en tiempo real el proceso de razonamiento y la lógica de decisión de cada agente.

  • tool_start/tool_end: muestra en tiempo real los parámetros de invocación de las herramientas y su duración.

  • hitl_request: dispara modales de aprobación frontales.

  • token: muestra con streaming carácter por carácter la salida final.

  • done: adjunta las estadísticas completas de consumo y optimización de tokens.

• Cambio de modelo sin fricción con Mock de alta funcionabilidad y modelos reales: incluye un modelo Mock de alta calidad integrado por defecto (simulación de 210 ms de latencia del primer token). Simplemente configure OPENAI_API_KEY en .env para cambiar directamente a GPT-4o, DeepSeek-V3/R1, Claude 3.5 o un conección a una instancia local de Ollama.

📂 Estructura del directorio del proyecto (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 压缩算法测试

🚀 Guía de inicio rápido (Quick Start)

Modo uno: Ejecución local en virtualenv (recomendada)

1. Configure las variables de entorno:

   cp .env.example .env

   (Por moto de uso, el modelo Mock de alto rendimiento integrado por defecto no requiere clave API y permite ejecución / experiencia inmediata).

2. Instalar las dependencias:

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

3. Iniciar el servicio web asíncrono FastAPI:

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

   • 🌐 Consola web interactiva: Abra un navegador y vaya a http://localhost:8000.

   • 📑 Documentación de la interfaz Swagger API: Zur Ver http://localhost:8000/docs.

Modo 2: despliegue en contenedores con Docker Compose (un toque)

docker-compose up -d --build

Este comando inicia automáticamente el contenedor FastAPI del backend y el servicio de checkpoint persistente Redis.

💻 Escenarios clásics y demostración de scripts (Demos & Benchmarks)

1. Demostración de colaboración multiagente en la línea de comandos

python examples/cli_demo.py

Ver en tiempo real el plan de división del trabajo, el proceso de unicership simulado MCP y los beneficios en tokens generados por los múltiples agentes en la consola de terminal.

2. Benchmark de Compresión de Redundancia de Tokens

python examples/evaluate_token_saving.py

Ejemplo de resultados de medición:

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

3. Procesar un servidor MCP independiente (para conectar con Claude Desktop / Cursor)

python examples/run_mcp_standalone.py

🧪 Pruebas automatizadas (Automated Testing)

Ejecute la suite completa de pruebas unitarias e integración generada de pruebas e2e de la máquina de estados:

pytest -v

Resultado de la salida de las pruebas:

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

📡 Especificación de las API principales (API Specifications)

📄 Licencia de código abierto (License)

Este proyecto está distribuido bajo la licencia MIT License.