LangGraph多智能体系统实战：MCP与RAG技术融合的医疗项目开发

如果你正在学习大模型 Agent 开发，可能已经发现了一个尴尬的现实：网上教程要么停留在单智能体的基础概念，要么直接跳到复杂的工业级架构，中间缺少一个平滑的过渡。更让人困惑的是，当你看到 LangGraph、MCP、RAG 这些热门技术时，很难理解它们在实际项目中到底如何协同工作。

本文要解决的核心问题就是：如何用 LangGraph 构建真正可落地的多智能体系统，特别是结合 MCP（Model Context Protocol）和 RAG（Retrieval-Augmented Generation）技术，实现从理论到工业级项目的跨越。我将通过一个完整的医疗项目实战，带你掌握多智能体架构的设计思路和工程化细节。

1. 这篇文章真正要解决的问题

大多数 LangGraph 教程只教你如何创建简单的对话流程，但现实中的多智能体系统需要解决三个关键问题：

第一，智能体间的协作机制。单个智能体能力有限，多个智能体如何分工协作？比如在医疗场景中，需要诊断智能体、用药建议智能体、病历分析智能体协同工作。

第二，外部工具的有效集成。智能体需要调用各种外部工具和知识库，传统集成方式复杂且难以维护。MCP 协议的出现正是为了解决这个痛点。

第三，专业知识的准确获取。在医疗等专业领域，单纯依赖大模型的内部知识远远不够，必须结合 RAG 技术从权威知识库中检索信息。

本文的医疗项目实战将完整演示如何用 LangGraph 构建多智能体工作流，用 MCP 协议标准化工具调用，用 RAG 系统提供专业知识支持。学完后你将掌握工业级多智能体系统的设计方法和实现细节。

2. LangGraph 多智能体架构基础概念

2.1 LangGraph 与 LangChain 的本质区别

很多人容易混淆 LangGraph 和 LangChain，认为它们只是同一框架的不同版本。实际上，两者的设计理念有根本区别：

• LangChain更注重链式调用（Chains），适合构建线性的、确定性的任务流程
• LangGraph专门为有状态、多分支的图结构工作流设计，特别适合多智能体协作场景

用通俗的话说：LangChain 像是一条生产线，每个环节按固定顺序执行；LangGraph 更像是一个团队，不同成员（智能体）可以并行工作、相互协作、根据情况动态调整分工。

2.2 多智能体系统的核心组件

一个完整的 LangGraph 多智能体系统包含以下核心概念：

节点（Nodes）：每个节点代表一个智能体或处理步骤。在医疗项目中，可能有"症状分析节点"、"诊断建议节点"、"用药审查节点"等。

边（Edges）：定义节点之间的流转条件。比如"如果症状严重程度高，则跳转到紧急处理节点"。

状态（State）：共享的上下文信息。这是 LangGraph 的关键优势——所有智能体都能访问和更新统一的状态对象。

检查点（Checkpoints）：保存执行状态，支持故障恢复和长时间运行的工作流。

2.3 MCP（Model Context Protocol）的核心价值

MCP 是一种标准化协议，让大模型能够安全、一致地使用外部工具和数据源。它的三大优势：

1. 工具标准化：不同来源的工具都通过统一的 MCP Server 暴露给智能体
2. 安全控制：可以精细控制每个工具的使用权限和资源访问
3. 开发效率：工具开发者只需实现一次 MCP 接口，就能被各种支持 MCP 的框架使用

在医疗项目中，我们将用 MCP 集成药品数据库、医疗指南、患者档案等外部系统。

2.4 Agentic RAG 与传统 RAG 的区别

传统 RAG 只是被动地检索和生成答案，而 Agentic RAG 让 RAG 系统成为主动的智能体：

• 传统 RAG：用户提问 → 检索文档 → 生成答案
• Agentic RAG：智能体分析问题 → 决定需要哪些信息 → 主动检索多个来源 → 综合判断 → 生成答案

在我们的医疗系统中，Agentic RAG 智能体能够主动查询药品相互作用、检查治疗禁忌症、验证诊断依据等。

3. 环境准备与项目架构设计

3.1 技术栈与版本要求

# 核心依赖 python>=3.9 langgraph>=0.0.40 langchain>=0.1.0 mcp>=1.0.0 # 可选组件，根据实际需求安装 chromadb>=0.4.0 # 向量数据库 fastapi>=0.100.0 # API服务 uvicorn>=0.20.0 # ASGI服务器

建议使用虚拟环境管理依赖：

python -m venv langgraph-medical source langgraph-medical/bin/activate # Linux/Mac # 或 langgraph-medical\Scripts\activate # Windows pip install langgraph langchain-core mcp-client chromadb

3.2 项目目录结构

medical_agent_system/ ├── src/ │ ├── agents/ # 智能体定义 │ │ ├── diagnostic_agent.py │ │ ├── medication_agent.py │ │ └── rag_agent.py │ ├── mcp_servers/ # MCP工具服务器 │ │ ├── drug_database.py │ │ └── medical_guidelines.py │ ├── graphs/ # LangGraph工作流定义 │ │ └── medical_workflow.py │ └── knowledge_base/ # RAG知识库 │ ├── vector_store.py │ └── documents/ # 医疗文档 ├── tests/ # 测试用例 └── config/ # 配置文件 └── settings.py

3.3 医疗项目业务架构设计

我们的多智能体医疗系统包含以下核心智能体：

1. 分诊智能体：初步评估症状紧急程度
2. 诊断智能体：基于症状和病史进行诊断分析
3. 用药智能体：检查药物相互作用和剂量建议
4. RAG 智能体：从医疗知识库检索最新指南
5. 报告智能体：生成患者友好的诊断报告

这些智能体通过 LangGraph 工作流协同工作，通过 MCP 调用外部医疗工具，通过 RAG 获取专业知识支持。

4. 核心流程拆解：从零构建多智能体系统

4.1 步骤一：定义共享状态模型

状态模型是多智能体系统的核心，它定义了智能体之间共享的数据结构：

# src/graphs/state.py from typing import Annotated, List, Dict, Any, Optional from typing_extensions import TypedDict from langgraph.graph import add_messages class MedicalState(TypedDict): # 患者信息 patient_id: str symptoms: List[str] medical_history: Dict[str, Any] # 诊断过程 triage_level: Annotated[str, add_messages()] # 紧急程度 preliminary_diagnosis: Annotated[List[str], add_messages()] diagnostic_confidence: float # 治疗建议 medication_suggestions: Annotated[List[Dict], add_messages()] drug_interactions: Annotated[List[str], add_messages()] # RAG检索结果 relevant_guidelines: Annotated[List[Dict], add_messages()] evidence_sources: Annotated[List[str], add_messages()] # 最终输出 final_report: Annotated[str, add_messages()] recommendations: Annotated[List[str], add_messages()]

这个状态对象会在整个工作流中传递，每个智能体都可以读取和更新自己负责的字段。

4.2 步骤二：实现 MCP 工具服务器

MCP 服务器负责提供标准化的工具接口。以下是药品数据库的 MCP 实现：

# src/mcp_servers/drug_database.py import json from mcp import MCPServer from mcp.types import Tool, TextContent class DrugDatabaseServer(MCPServer): def __init__(self): super().__init__("drug-database") # 模拟药品数据库 self.drug_db = { "aspirin": { "indications": ["pain", "fever", "inflammation"], "contraindications": ["bleeding disorders", "asthma"], "interactions": ["warfarin", "ibuprofen"] }, "metformin": { "indications": ["type 2 diabetes"], "contraindications": ["renal impairment"], "interactions": ["contrast media"] } } async def get_tools(self) -> list[Tool]: return [ Tool( name="check_drug_info", description="查询药品基本信息", inputSchema={ "type": "object", "properties": { "drug_name": {"type": "string"} }, "required": ["drug_name"] } ), Tool( name="check_interactions", description="检查药物相互作用", inputSchema={ "type": "object", "properties": { "drugs": {"type": "array", "items": {"type": "string"}} }, "required": ["drugs"] } ) ] async def call_tool(self, name: str, arguments: dict) -> list[TextContent]: if name == "check_drug_info": drug_name = arguments["drug_name"].lower() drug_info = self.drug_db.get(drug_name) if drug_info: return [TextContent(type="text", text=json.dumps(drug_info, ensure_ascii=False))] else: return [TextContent(type="text", text="药品未找到")] elif name == "check_interactions": drugs = [d.lower() for d in arguments["drugs"]] interactions = [] for drug in drugs: if drug in self.drug_db: interactions.extend(self.drug_db[drug].get("interactions", [])) return [TextContent(type="text", text=json.dumps(list(set(interactions)), ensure_ascii=False))]

4.3 步骤三：构建 RAG 知识库系统

RAG 系统为智能体提供专业知识支持：

# src/knowledge_base/vector_store.py import chromadb from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import OpenAIEmbeddings import os class MedicalKnowledgeBase: def __init__(self, persist_directory="./data/chroma"): self.embeddings = OpenAIEmbeddings() self.client = chromadb.PersistentClient(path=persist_directory) self.collection = self.client.get_or_create_collection("medical_guidelines") def add_documents(self, documents: list): """添加医疗文档到知识库""" text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200 ) chunks = text_splitter.split_documents(documents) # 为每个块生成嵌入并存储 for i, chunk in enumerate(chunks): self.collection.add( documents=[chunk.page_content], metadatas=[chunk.metadata], ids=[f"chunk_{i}"] ) def search(self, query: str, n_results: int = 3): """检索相关医疗指南""" results = self.collection.query( query_texts=[query], n_results=n_results ) return results

4.4 步骤四：实现诊断智能体

诊断智能体负责分析症状并提出初步诊断：

# src/agents/diagnostic_agent.py from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI from .base_agent import BaseAgent class DiagnosticAgent(BaseAgent): def __init__(self): super().__init__() self.llm = ChatOpenAI(model="gpt-4", temperature=0.1) self.system_prompt = """你是一名经验丰富的医疗诊断专家。根据患者的症状和病史，提供专业的诊断建议。 请遵循以下原则： 1. 基于症状进行逻辑推理 2. 考虑常见的鉴别诊断 3. 指出需要进一步检查的项目 4. 评估紧急程度 5. 引用最新的医疗指南""" async def invoke(self, state: dict) -> dict: symptoms = state.get("symptoms", []) history = state.get("medical_history", {}) prompt = f""" 患者症状：{', '.join(symptoms)} 病史信息：{history} 请进行诊断分析： 1. 可能的诊断方向 2. 紧急程度评估 3. 需要排除的危险情况 4. 建议的检查项目 """ messages = [ SystemMessage(content=self.system_prompt), HumanMessage(content=prompt) ] response = self.llm.invoke(messages) # 更新状态 return { "preliminary_diagnosis": self._extract_diagnosis(response.content), "triage_level": self._assess_urgency(response.content), "diagnostic_confidence": 0.85 # 模拟置信度 } def _extract_diagnosis(self, text: str) -> list: # 简化实现，实际应该用更复杂的方法解析LLM输出 return ["初步诊断待确认"] def _assess_urgency(self, text: str) -> str: if "紧急" in text or "立即" in text: return "high" elif "尽快" in text: return "medium" else: return "low"

5. LangGraph 工作流集成与编排

5.1 构建完整的医疗工作流

现在我们将所有组件集成到 LangGraph 工作流中：

# src/graphs/medical_workflow.py from langgraph.graph import StateGraph, END from langgraph.prebuilt import create_react_agent from ..agents import DiagnosticAgent, MedicationAgent, RAGAgent from ..mcp_servers import DrugDatabaseServer from .state import MedicalState class MedicalWorkflow: def __init__(self): self.graph_builder = StateGraph(MedicalState) # 初始化各个智能体 self.diagnostic_agent = DiagnosticAgent() self.medication_agent = MedicationAgent() self.rag_agent = RAGAgent() # 添加节点 self.graph_builder.add_node("triage", self.triage_node) self.graph_builder.add_node("diagnose", self.diagnostic_agent.invoke) self.graph_builder.add_node("check_medication", self.medication_agent.invoke) self.graph_builder.add_node("consult_guidelines", self.rag_agent.invoke) self.graph_builder.add_node("generate_report", self.report_node) # 定义工作流路径 self.graph_builder.set_entry_point("triage") self.graph_builder.add_edge("triage", "diagnose") self.graph_builder.add_conditional_edges( "diagnose", self.route_after_diagnosis, { "need_medication": "check_medication", "need_guidelines": "consult_guidelines", "direct_report": "generate_report" } ) self.graph_builder.add_edge("check_medication", "consult_guidelines") self.graph_builder.add_edge("consult_guidelines", "generate_report") self.graph_builder.add_edge("generate_report", END) self.graph = self.graph_builder.compile() def triage_node(self, state: MedicalState) -> dict: """分诊节点：评估紧急程度""" symptoms = state.get("symptoms", []) # 简单的紧急程度评估逻辑 emergency_keywords = ["胸痛", "呼吸困难", "意识模糊", "大出血"] urgency = "low" for symptom in symptoms: if any(keyword in symptom for keyword in emergency_keywords): urgency = "high" break return {"triage_level": urgency} def route_after_diagnosis(self, state: MedicalState) -> str: """根据诊断结果路由到不同路径""" diagnosis = state.get("preliminary_diagnosis", []) confidence = state.get("diagnostic_confidence", 0) if confidence < 0.7: return "need_guidelines" # 需要查阅指南 elif any("用药" in d or "药物" in d for d in diagnosis): return "need_medication" # 需要药物检查 else: return "direct_report" # 直接生成报告 def report_node(self, state: MedicalState) -> dict: """生成最终报告""" diagnosis = state.get("preliminary_diagnosis", []) medications = state.get("medication_suggestions", []) guidelines = state.get("relevant_guidelines", []) report = f""" 医疗诊断报告 ============ 初步诊断：{', '.join(diagnosis)} 治疗建议： {self._format_medications(medications)} 参考指南： {self._format_guidelines(guidelines)} 注意事项：本报告仅供参考，具体诊疗请遵医嘱。 """ return { "final_report": report, "recommendations": [ "定期复查", "注意症状变化", "如有不适及时就医" ] } def _format_medications(self, meds: list) -> str: return "\n".join([f"- {med.get('name', '未知')}: {med.get('dose', '')}" for med in meds]) def _format_guidelines(self, guidelines: list) -> str: return "\n".join([f"- {guideline.get('title', '未知指南')}" for guideline in guidelines])

5.2 工作流执行与状态管理

# src/main.py from graphs.medical_workflow import MedicalWorkflow from graphs.state import MedicalState def run_medical_consultation(patient_data: dict): """运行医疗咨询工作流""" workflow = MedicalWorkflow() # 初始化状态 initial_state: MedicalState = { "patient_id": patient_data["id"], "symptoms": patient_data["symptoms"], "medical_history": patient_data.get("history", {}), "triage_level": "", "preliminary_diagnosis": [], "diagnostic_confidence": 0.0, "medication_suggestions": [], "drug_interactions": [], "relevant_guidelines": [], "evidence_sources": [], "final_report": "", "recommendations": [] } # 执行工作流 final_state = workflow.graph.invoke(initial_state) return final_state # 使用示例 if __name__ == "__main__": patient_data = { "id": "patient_001", "symptoms": ["头痛", "发热", "咳嗽"], "history": {"allergies": ["青霉素"], "chronic_conditions": []} } result = run_medical_consultation(patient_data) print("诊断报告：", result["final_report"])

6. 运行结果与效果验证

6.1 测试用例设计与执行

为了验证系统的可靠性，我们需要设计全面的测试用例：

# tests/test_medical_workflow.py import pytest from src.main import run_medical_consultation class TestMedicalWorkflow: def test_common_cold_symptoms(self): """测试普通感冒症状""" patient_data = { "id": "test_001", "symptoms": ["鼻塞", "流涕", "喉咙痛"], "history": {} } result = run_medical_consultation(patient_data) assert "final_report" in result assert len(result["preliminary_diagnosis"]) > 0 assert result["triage_level"] == "low" # 普通感冒应该是低紧急度 def test_emergency_symptoms(self): """测试紧急症状""" patient_data = { "id": "test_002", "symptoms": ["胸痛", "呼吸困难"], "history": {"heart_disease": True} } result = run_medical_consultation(patient_data) assert result["triage_level"] == "high" def test_medication_interaction(self): """测试药物相互作用检查""" patient_data = { "id": "test_003", "symptoms": ["关节疼痛"], "history": {"current_medications": ["华法林"]} } result = run_medical_consultation(patient_data) # 应该检测到阿司匹林和华法林的相互作用 assert len(result["drug_interactions"]) >= 0 if __name__ == "__main__": pytest.main([__file__])

6.2 性能与准确性评估

在实际运行中，我们需要关注以下指标：

1. 响应时间：整个工作流应在合理时间内完成（通常<30秒）
2. 诊断准确性：与专业医生诊断的一致性
3. 指南相关性：RAG 系统检索结果的相关度
4. 药物安全：药物相互作用检查的完备性

可以通过以下命令监控系统运行：

# 运行性能测试 python -m cProfile -o profile.stats tests/performance_test.py # 查看详细报告 python -c "import pstats; p = pstats.Stats('profile.stats'); p.sort_stats('time').print_stats(10)"

7. 常见问题与排查思路

7.1 LangGraph 工作流问题

7.2 MCP 集成问题

# MCP 连接调试工具 async def debug_mcp_connection(): from mcp_client import MCPClient try: async with MCPClient("http://localhost:8000") as client: tools = await client.list_tools() print("可用工具:", tools) except Exception as e: print(f"MCP 连接失败: {e}") # 常见的 MCP 问题排查步骤： # 1. 检查 MCP 服务器是否启动 # 2. 验证端口和地址配置 # 3. 检查工具定义是否符合规范 # 4. 查看服务器日志中的错误信息

7.3 RAG 知识库问题

检索结果不相关：

• 检查文档分块策略是否合适
• 验证嵌入模型的质量
• 调整检索参数（top_k、相似度阈值）

知识库更新问题：

# 知识库更新监控 def monitor_knowledge_freshness(): """检查知识库的新鲜度""" from datetime import datetime, timedelta # 获取最新文档的更新时间 latest_update = get_latest_document_time() if datetime.now() - latest_update > timedelta(days=30): print("警告：知识库需要更新") # 触发自动更新流程 update_knowledge_base()

8. 最佳实践与工程建议

8.1 多智能体系统设计原则

职责单一原则：每个智能体应该专注于一个明确的职责范围。比如诊断智能体只负责诊断，用药智能体只负责药物相关判断。

状态管理规范：明确哪些状态字段可以被哪些智能体修改，避免并发写入冲突。

错误隔离机制：单个智能体的失败不应该导致整个系统崩溃，需要有降级策略。

# 智能体错误处理装饰器 def agent_error_handler(func): def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: logger.error(f"智能体执行失败: {e}") # 返回降级结果，而不是让整个工作流失败 return {"error": str(e), "fallback_result": "使用备用逻辑"} return wrapper

8.2 生产环境部署考虑

配置管理：

# config/settings.py import os from pydantic_settings import BaseSettings class Settings(BaseSettings): # LLM 配置 openai_api_key: str = os.getenv("OPENAI_API_KEY") model_name: str = "gpt-4" # 向量数据库配置 chroma_path: str = "./data/chroma" # MCP 服务器配置 drug_db_url: str = "http://localhost:8001" guidelines_url: str = "http://localhost:8002" # 性能配置 max_workers: int = 10 request_timeout: int = 30 settings = Settings()

监控与日志：

# 结构化日志配置 import structlog logger = structlog.get_logger() def log_workflow_execution(patient_id: str, duration: float, success: bool): logger.info( "workflow_completed", patient_id=patient_id, duration_ms=duration*1000, success=success, component="medical_workflow" )

8.3 安全与合规注意事项

在医疗等敏感领域，需要特别关注：

数据隐私：患者信息需要脱敏处理，遵守相关隐私法规。

审计追踪：记录所有的诊断过程和决策依据。

责任边界：明确系统是辅助工具，最终决策需要人类医生确认。

# 数据脱敏处理 def anonymize_patient_data(patient_data: dict) -> dict: """脱敏患者信息""" return { "id": hash(patient_data["id"]), # 使用哈希代替真实ID "symptoms": patient_data["symptoms"], "history": { "age_group": patient_data["history"].get("age_group"), # 不存储具体年龄等敏感信息 } }

9. 项目扩展与优化方向

9.1 性能优化策略

智能体并行化：当智能体之间没有依赖关系时，可以并行执行。

缓存机制：对频繁查询的药品信息、指南内容添加缓存。

异步处理：使用异步IO提高系统吞吐量。

# 并行执行无依赖的智能体 async def execute_parallel_agents(agents: list, state: dict): """并行执行多个智能体""" import asyncio tasks = [] for agent in agents: task = asyncio.create_task(agent.invoke(state)) tasks.append(task) results = await asyncio.gather(*tasks, return_exceptions=True) return results

9.2 功能扩展思路

多模态支持：集成影像分析智能体，处理医疗图像。

实时协作：支持多个医生同时参与诊断过程。

个性化适配：根据患者历史数据调整诊断策略。

持续学习：通过反馈机制不断优化智能体表现。

这个 LangGraph 多智能体医疗项目展示了如何将先进的大模型技术应用到实际业务场景中。通过合理的架构设计和工程化实践，我们构建了一个可扩展、可维护的智能医疗辅助系统。

关键是要理解：多智能体不是简单的功能堆砌，而是需要精心设计的协作体系。每个智能体各司其职，通过标准化协议相互通信，在统一的状态管理下协同工作，这样才能发挥出 1+1>2 的效果。

建议在实际项目中从小规模开始，先验证核心工作流的可行性，再逐步扩展智能体数量和功能范围。同时要建立完善的测试和监控体系，确保系统的可靠性和安全性。