Multiagent Langgraph MCP

基于LangGraph和MCP的多Agent客户服务系统

一个多代理客户服务系统，其中专业代理使用 LangGraph的共享状态和条件边 并通过模型上下文协议（MCP）访问客户数据。

建筑

多智能体协调方法

该项目实施 LangGraph消息传递 代理协调：

• 共享状态结构：代理对共享进行读写操作 AgentState 类型冲突
• 每个代理的节点：路由器、CustomerDataAgent和SupportAgent都有专用节点
• 条件边：基于状态的动态路由（路由器的分析决定下一个代理）
• 状态迁移：通过LangGraph的图执行进行显式处理

代理

• 路由器代理（编排器）：接收客户查询，分析意图，将路线决策写入状态
• 客户数据代理（专家）：通过MCP访问客户数据库，将数据写入状态
• 支持代理（专家）：处理支持查询，使用来自共享状态的客户上下文

LangGraph协调流程

User Query → Router Node → [Analyze Intent, Write to State]
                ↓
         Conditional Edge (reads state.requires_customer_data)
                ↓
         Customer Data Node → [Get Info via MCP, Write to State]
                ↓
         Conditional Edge (reads state.requires_support)
                ↓
         Support Node → [Handle Query, Read Customer Context from State]
                ↓
         Synthesize Node → [Combine All State Data] → User

项目结构

multiagent-a2a-mcp/
├── src/
│   ├── __init__.py
│   ├── main.py                 # Entry point (terminal or Gradio mode)
│   ├── agents/
│   │   ├── __init__.py
│   │   ├── base_agent.py       # Base class for LangGraph agents
│   │   ├── router_agent.py     # Orchestrator agent
│   │   ├── customer_data_agent.py  # Customer data specialist
│   │   ├── support_agent.py    # Support specialist
│   │   └── tools.py            # LangGraph tool wrappers (use MCP Client)
│   ├── mcp/
│   │   ├── __init__.py
│   │   ├── mcp_server.py       # MCP server (FastMCP SDK)
│   │   └── mcp_client.py       # MCP client for protocol-compliant access
│   ├── a2a/
│   │   ├── __init__.py
│   │   ├── protocol.py         # Data classes for logging (NOT A2A protocol)
│   │   ├── registry.py         # Agent registry for discovery
│   │   └── a2a_server.py       # Optional: A2A SDK server for external exposure
│   ├── graph/
│   │   ├── __init__.py
│   │   └── workflow.py         # LangGraph workflow with shared state
│   └── ui/
│       ├── __init__.py
│       └── gradio_app.py       # Gradio web interface
├── tests/
│   ├── __init__.py
│   ├── test_agents.py
│   ├── test_mcp.py
│   ├── test_a2a.py             # Tests for data classes
│   └── test_scenarios.py
├── data/
│   ├── database_setup.py       # Database initialization script
│   └── customers.db            # SQLite database (generated)
├── run_servers.py              # CLI to run MCP server
├── demo.ipynb                  # Interactive demo notebook
├── requirements.txt
├── .env.example
└── README.md

______________________________________________________________________

LangGraph共享状态架构

代理协调的核心是 AgentState 类型冲突：

class AgentState(TypedDict):
    query: str                      # Original user query
    analysis: Dict[str, Any]        # Router's analysis (routing decisions)
    customer_data: Optional[Dict]   # Data from CustomerDataAgent
    support_response: Optional[Dict] # Response from SupportAgent
    completed_agents: List[str]     # Track which agents processed
    agent_logs: List[str]           # Coordination logs
    message_history: List[Dict]     # Message tracking for visualization
    response: str                   # Final synthesized response
    iteration: int                  # Safety counter

条件路由

def should_continue(state: AgentState) -> str:
    """Conditional edge function - reads state to determine routing."""
    analysis = state.get("analysis", {})
    completed = state.get("completed_agents", [])
    
    if analysis.get("requires_customer_data") and "customer_data" not in completed:
        return "customer_data"
    if analysis.get("requires_support") and "support" not in completed:
        return "support"
    return "synthesize"

______________________________________________________________________

架构：正确的MCP协议合规性

此项目使用 正确遵守MCP协议 -代理通过HTTP/JSON-RPC 2.0与MCP服务器通信

┌─────────────┐        ┌─────────────┐          ┌─────────────┐
│   Agents    │ ──────▶│ MCP Client  │ ───────▶│ MCP Server  │
│ (LangGraph) │ Tools  │ (HTTP/JSON) │ Protocol │  (FastMCP)  │
└─────────────┘        └─────────────┘          └─────────────┘
                                                       │
                                                 ┌─────▼─────┐
                                                 │  SQLite   │
                                                 │ Database  │
                                                 └───────────┘

为什么是这种架构？

1. 通用兼容性 -任何MCP客户端都可以使用这些工具
2. 清洁分离 -服务器和客户端可以独立部署
3. 可测试性 -组件可以单独测试
4. 生产就绪 -匹配真实MCP集成的工作方式

用法

MCP服务器必须正在运行，代理才能工作：

# Start MCP Server (required)
python run_servers.py mcp --transport http

# Then run the demo notebook

代码示例

# Agents use MCP Client to call tools via protocol
from src.mcp import MCPClient

client = MCPClient()  # Default: http://localhost:8080
result = client.call_tool("get_customer", {"customer_id": 5})

______________________________________________________________________

安装说明

1.克隆并导航到项目

git clone 
cd multiagent-a2a-mcp

2.创建虚拟环境

# Create virtual environment
python -m venv venv

# Activate (Windows)
venv\Scripts\activate

# Activate (macOS/Linux)
source venv/bin/activate

3.安装依赖项

pip install uv
uv pip install -r requirements.txt

4.配置环境变量

# Copy example environment file
copy .env.example .env    # Windows
cp .env.example .env      # macOS/Linux

# Edit .env and add your OpenAI API key
# OPENAI_API_KEY=your-api-key-here

5.初始化数据库

python data/database_setup.py

这创造了 data/customers.db 带有客户和票务数据样本。

6.运行应用程序

Gradio Web界面（默认）：

python -m src.main --mode gradio

然后打开http://localhost:7860在您的浏览器中。

终端模式：

python -m src.main --mode terminal

数据库模式

客户表

门票桌

MCP工具

测试场景

场景1：简单查询（任务分配）

Query: "Get customer information for ID 5"
Flow: Router → CustomerDataAgent (MCP call) → Router → Response

场景2：协同查询

Query: "I'm customer 12345 and need help upgrading my account"
Flow: Router → CustomerDataAgent → SupportAgent → Router → Response

场景3：复杂查询（协商）

Query: "Show me all active customers who have open tickets"
Flow: Router → CustomerDataAgent (complex query) → SupportAgent (format report) → Router

情景4：升级

Query: "I've been charged twice, please refund immediately!"
Flow: Router (detect urgency) → CustomerDataAgent → SupportAgent (escalation) → Create Ticket → Response

场景5：多意图

Query: "Update my email to new@email.com and show my ticket history"
Flow: Router → CustomerDataAgent (update + history) → SupportAgent → Router → Response

运行测试

pytest tests/ -v --cov=src

代理协调日志

系统提供详细的日志，显示基于LangGraph状态的协调：

[Router] Analyzing query: 'Get customer information for ID 5'
[Router] Identified intents: ['get_info']. Requires customer data: True.
[CustomerDataAgent] Processing request from shared state...
[CustomerDataAgent] Found 1 task request(s) in message history
[CustomerDataAgent] Tool call: get_customer_tool({'customer_id': 5})
[CustomerDataAgent] Result: Success - Request processed successfully
[SupportAgent] Processing request from shared state...
[SupportAgent] Using customer data from shared state (context sharing)
[SupportAgent] Request classified as: general
[Router] Synthesizing final response from agent outputs...
[Router] Agents that contributed: ['customer_data', 'support']

依赖项

• 兰格拉夫：多代理工作流编排
• 语言链：LLM框架
• 兰开夏：OpenAI集成
• 格拉迪奥：Web界面
• python dotenv：环境变量管理
• pytest：测试框架

展示学习目标

1. 使用LangGraph进行代理协调 -共享状态、条件边、消息传递
2. 通过MCP进行外部工具集成 -通过协议访问SQLite数据库
3. 多代理任务分配 -路由器确定哪些代理处理每个查询
4. 基于状态的上下文共享 -客户数据通过共享状态流向支持代理
5. 实用的客户服务自动化 -现实世界中的多步骤工作流程

运行端到端演示Jupyter Notebook

打开并运行 demo.ipynb -这提供了所有6个场景的交互式演练，其中包含详细的A2A协调日志。

这将运行所有6个必需的场景并显示：

• 显示A2A通信的代理协调日志
• 查询分析和路由决策
• 系统的最终响应
• 所有情景结果摘要

结论

我学到了什么

构建这个多代理客户服务系统为设计协调的人工智能系统提供了深刻的见解。最有价值的学习是了解如何使用以下方法构建代理协调 LangGraph的共享状态和条件边 （选项B:LangGraph消息传递）。这种方法使用共享的A2A协议，而不是实现外部A2A协议 AgentState TypedDict，其中每个代理从以前的代理读取上下文并贡献其专门的输出。条件边模式(should_continue)支持基于路由器分析的动态路由，在保持灵活性的同时创建干净的关注点分离。

实现用于数据库访问的模型上下文协议（MCP）加强了定义良好的工具接口的重要性——MCP服务器充当AI代理和外部数据源之间的桥梁，提供代理可以可靠地解释和采取行动的结构化响应。

关键设计决策

1. 消息队列上的共享状态：使用LangGraph的 TypedDict 状态允许在代理之间传递类型安全的数据
2. 路由的条件边：The should_continue 函数读取状态以确定下一个执行哪个代理
3. 通过状态共享上下文：客户数据通过以下方式流向支持代理 state["customer_data"]
4. 日志记录的消息历史记录：跟踪消息以进行可视化，而不将其用于实际路由

面临的挑战

主要的挑战是设计路由器代理的查询分析，以正确识别用户意图并确定调用哪些专业代理。早期迭代要么路由过度（将每个查询发送给所有代理），要么路由不足（缺少多意图查询）。该解决方案是实现一个强大的回退分析，结合关键字匹配和基于LLM的意图检测。

另一个重大挑战是管理响应延迟——每个查询有多个LLM调用（路由器分析、数据解释、支持响应和最终合成），响应时间可能会超过可接受的阈值。通过仔细的提示工程来减少令牌的使用，并通过为代理实例实现单例模式来避免重复的初始化开销，从而解决了这个问题。

调试多代理交互最初也很困难；在每个协调点添加全面的日志记录并跟踪消息历史对于理解流程和确定问题发生的位置至关重要。