在AI应用开发领域多智能体系统正成为解决复杂任务的关键技术。LangGraph作为新兴的多智能体架构框架能够有效协调多个AI智能体协同工作特别适合医疗咨询、企业知识库等需要专业领域知识的场景。本文将带你从零开始用600分钟系统掌握LangGraph多智能体架构结合MCP模型上下文协议和RAG检索增强生成技术实战构建一个完整的医疗咨询项目。无论你是刚接触AI Agent开发的初学者还是希望提升工业级落地能力的进阶开发者本文都将提供从基础概念到项目实战的完整路径。学完后你将能够独立设计并实现基于LangGraph的多智能体系统避免常见的开发陷阱大幅提升开发效率。1. LangGraph多智能体架构核心概念1.1 什么是LangGraph及其与LangChain的区别LangGraph是专门为构建多智能体系统而设计的框架它基于有向图的概念来组织和管理多个AI智能体之间的协作关系。与LangChain相比LangGraph更专注于智能体间的状态管理和工作流协调。LangChain主要解决的是单个智能体的工具调用和链式操作而LangGraph的核心优势在于处理多个智能体之间的复杂交互。在LangGraph中每个节点代表一个智能体或处理步骤边代表状态流转路径这种设计使得复杂的工作流变得清晰可控。关键区别点包括架构层面LangChain是链式结构LangGraph是图结构状态管理LangGraph提供更强大的状态跟踪和传递机制并发处理LangGraph天然支持多个智能体并行执行错误处理LangGraph提供更完善的错误恢复和重试机制1.2 多智能体系统的核心价值多智能体系统通过分工协作来解决单一模型无法处理的复杂问题。在医疗场景中这意味着可以同时拥有问诊智能体、药品知识智能体、治疗方案推荐智能体等各司其职又相互配合。多智能体架构的主要优势包括专业化分工每个智能体专注于特定领域提供更专业的服务错误容忍单个智能体失败不影响整个系统运行可扩展性容易添加新的智能体来扩展系统能力效率提升并行处理不同子任务减少总体响应时间1.3 MCP模型上下文协议详解MCPModel Context Protocol是一种标准化的协议用于在不同AI模型和工具之间传递上下文信息。在医疗多智能体系统中MCP确保各个智能体能够共享患者信息、诊断历史、药品知识等关键数据。MCP协议的核心组件包括上下文容器统一格式的数据存储结构消息格式标准化的智能体间通信协议权限控制确保敏感医疗数据的安全访问版本管理支持不同版本模型和工具的协同工作1.4 RAG在医疗领域的特殊价值RAGRetrieval-Augmented Generation技术通过检索相关知识来增强大模型的生成能力。在医疗应用中RAG能够确保生成的建议基于最新的医学指南和权威资料而不是仅仅依赖模型的训练数据。医疗RAG系统的关键要求数据准确性必须使用经过验证的医疗知识库实时更新能够及时纳入最新的医学研究成果多源检索从临床指南、药品数据库、病例库等多个来源获取信息可信度评估对检索结果进行可靠性评分2. 环境准备与工具配置2.1 基础环境要求在开始LangGraph项目之前需要确保开发环境满足以下要求操作系统要求Windows 10/11, macOS 10.14, 或 Ubuntu 18.04至少8GB内存推荐16GB以上Python 3.8-3.11版本Python环境配置# 创建虚拟环境 python -m venv langgraph-medical source langgraph-medical/bin/activate # Linux/macOS # 或 langgraph-medical\Scripts\activate # Windows # 安装基础依赖 pip install langgraph langchain-openai langchain-community pip install faiss-cpu sentence-transformers # 向量数据库和嵌入模型2.2 LangGraph及相关库安装安装完整的开发工具链# 核心框架 pip install langgraph0.0.40 # OpenAI集成 pip install openai langchain-openai # 向量数据库和检索 pip install faiss-cpu chromadb sentence-transformers # 医疗数据处理 pip install pandas numpy requests beautifulsoup4 # 开发工具 pip install jupyter ipython pytest2.3 开发环境验证创建环境验证脚本verify_environment.py#!/usr/bin/env python3 环境验证脚本检查所有必要依赖是否正常安装 import importlib import sys def check_package(package_name, import_nameNone): 检查包是否可正常导入 try: if import_name is None: import_name package_name importlib.import_module(import_name) print(f✅ {package_name} 导入成功) return True except ImportError as e: print(f❌ {package_name} 导入失败: {e}) return False # 检查核心依赖 required_packages [ (langgraph, langgraph), (openai, openai), (langchain, langchain), (faiss, faiss), (chromadb, chromadb), ] print(开始检查环境依赖...) all_passed all(check_package(pkg, imp) for pkg, imp in required_packages) if all_passed: print(\n 所有依赖检查通过环境配置成功) else: print(\n⚠️ 部分依赖检查失败请检查安装步骤) sys.exit(1)2.4 API密钥配置创建配置文件.env用于管理敏感信息# OpenAI API配置 OPENAI_API_KEYyour_openai_api_key_here # 其他服务API密钥如有 ANTHROPIC_API_KEYyour_anthropic_api_key SERPER_API_KEYyour_serper_api_key # 应用配置 DEBUGTrue LOG_LEVELINFO对应的配置读取代码config.pyimport os from dotenv import load_dotenv load_dotenv() class Config: 配置管理类 # API密钥 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) ANTHROPIC_API_KEY os.getenv(ANTHROPIC_API_KEY) # 模型配置 GPT_MODEL gpt-4o # 或 gpt-4, gpt-3.5-turbo EMBEDDING_MODEL text-embedding-3-small # 路径配置 DATA_DIR ./data VECTOR_DB_PATH ./vector_db LOG_DIR ./logs classmethod def validate(cls): 验证必要配置 if not cls.OPENAI_API_KEY: raise ValueError(OPENAI_API_KEY未设置请检查.env文件) # 创建必要目录 os.makedirs(cls.DATA_DIR, exist_okTrue) os.makedirs(cls.VECTOR_DB_PATH, exist_okTrue) os.makedirs(cls.LOG_DIR, exist_okTrue) # 初始化时验证配置 Config.validate()3. LangGraph基础架构与核心组件3.1 LangGraph状态管理机制LangGraph的核心是状态管理它通过StateGraph来维护整个工作流的状态。在医疗多智能体系统中状态包含患者信息、对话历史、诊断结果等关键数据。基础状态定义示例from typing import TypedDict, Annotated, List, Dict, Any from langgraph.graph import StateGraph, END import operator class MedicalState(TypedDict): 医疗咨询系统状态定义 patient_query: str # 患者查询 medical_history: Annotated[List[Dict], operator.add] # 医疗历史记录 current_symptoms: List[str] # 当前症状 preliminary_diagnosis: str # 初步诊断 drug_recommendations: List[Dict] # 药品推荐 final_advice: str # 最终建议 error_messages: List[str] # 错误信息3.2 智能体节点设计与实现每个智能体都是LangGraph中的一个节点负责特定的任务。以下是医疗系统中几个核心智能体的实现症状分析智能体from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI from config import Config class SymptomAnalyzerAgent: 症状分析智能体 def __init__(self): self.llm ChatOpenAI( modelConfig.GPT_MODEL, temperature0.1, # 医疗场景需要低随机性 api_keyConfig.OPENAI_API_KEY ) self.system_prompt 你是一个专业的医疗症状分析专家。你的任务是 1. 分析患者描述的症状 2. 识别可能的疾病方向 3. 评估紧急程度 4. 建议下一步检查方向 请保持专业、谨慎始终建议患者咨询专业医生。 def analyze(self, state: MedicalState) - MedicalState: 分析症状 try: messages [ SystemMessage(contentself.system_prompt), HumanMessage(contentf患者症状描述{state[patient_query]}) ] response self.llm.invoke(messages) analysis_result response.content # 更新状态 state[preliminary_diagnosis] analysis_result return state except Exception as e: state[error_messages].append(f症状分析失败{str(e)}) return state def symptom_analyzer_node(state: MedicalState) - MedicalState: 症状分析节点函数 agent SymptomAnalyzerAgent() return agent.analyze(state)药品知识智能体class DrugKnowledgeAgent: 药品知识智能体 def __init__(self): self.llm ChatOpenAI( modelConfig.GPT_MODEL, temperature0.1, api_keyConfig.OPENAI_API_KEY ) def search_drugs(self, symptoms: List[str], diagnosis: str) - List[Dict]: 根据症状和诊断搜索相关药品 # 这里可以集成真实的药品数据库 prompt f 基于以下信息推荐药品 - 症状{, .join(symptoms)} - 初步诊断{diagnosis} 请推荐3-5种相关药品包含 - 药品名称 - 适用症状 - 注意事项 - 是否需要处方 注意必须明确说明这些只是参考信息实际用药需遵医嘱。 messages [ HumanMessage(contentprompt) ] response self.llm.invoke(messages) # 解析响应提取药品信息 return self._parse_drug_recommendations(response.content) def _parse_drug_recommendations(self, text: str) - List[Dict]: 解析药品推荐结果 # 简化实现实际项目中需要更复杂的解析逻辑 drugs [] lines text.split(\n) current_drug {} for line in lines: line line.strip() if line.startswith(药品名称): if current_drug: drugs.append(current_drug) current_drug {name: line.replace(药品名称, )} elif line.startswith(适用症状): current_drug[symptoms] line.replace(适用症状, ) elif line.startswith(注意事项): current_drug[precautions] line.replace(注意事项, ) if current_drug: drugs.append(current_drug) return drugs def drug_knowledge_node(state: MedicalState) - MedicalState: 药品知识节点 agent DrugKnowledgeAgent() symptoms state[current_symptoms] diagnosis state[preliminary_diagnosis] recommendations agent.search_drugs(symptoms, diagnosis) state[drug_recommendations] recommendations return state3.3 图结构构建与路由逻辑将各个智能体节点连接成完整的工作流def build_medical_graph() - StateGraph: 构建医疗咨询图 # 创建图实例 workflow StateGraph(MedicalState) # 添加节点 workflow.add_node(symptom_analyzer, symptom_analyzer_node) workflow.add_node(drug_knowledge, drug_knowledge_node) workflow.add_node(advice_generator, advice_generator_node) # 设置入口点 workflow.set_entry_point(symptom_analyzer) # 定义边路由逻辑 workflow.add_edge(symptom_analyzer, drug_knowledge) workflow.add_edge(drug_knowledge, advice_generator) workflow.add_edge(advice_generator, END) # 编译图 return workflow.compile() def should_consult_doctor(state: MedicalState) - str: 判断是否需要直接建议就医 urgent_keywords [紧急, 严重, 立即,马上去医院, 危险] query state[patient_query].lower() if any(keyword in query for keyword in urgent_keywords): return emergency else: return continue # 添加条件路由 workflow.add_conditional_edges( symptom_analyzer, should_consult_doctor, { emergency: emergency_advisor, continue: drug_knowledge } )4. MCP协议在医疗多智能体系统中的实现4.1 MCP服务器设计与实现MCP服务器负责管理不同智能体间的上下文共享import json from typing import Dict, Any, List from datetime import datetime class MCPServer: MCP服务器实现 def __init__(self): self.context_store: Dict[str, Any] {} self.message_queue: List[Dict] [] self.agent_registry: Dict[str, Dict] {} def register_agent(self, agent_id: str, agent_info: Dict): 注册智能体 self.agent_registry[agent_id] { **agent_info, registered_at: datetime.now().isoformat(), last_heartbeat: datetime.now().isoformat() } def store_context(self, context_key: str, context_data: Any, agent_id: str): 存储上下文数据 self.context_store[context_key] { data: context_data, created_by: agent_id, created_at: datetime.now().isoformat(), version: 1 } def retrieve_context(self, context_key: str) - Dict[str, Any]: 检索上下文数据 return self.context_store.get(context_key, {}) def broadcast_message(self, message_type: str, payload: Dict, sender_id: str): 广播消息给所有智能体 message { type: message_type, payload: payload, sender: sender_id, timestamp: datetime.now().isoformat(), message_id: fmsg_{len(self.message_queue) 1:06d} } self.message_queue.append(message) def get_messages_for_agent(self, agent_id: str) - List[Dict]: 获取指定智能体的消息 # 简化实现返回所有消息 # 实际项目中应该实现更复杂的消息路由 return self.message_queue # MCP客户端实现 class MCPClient: MCP客户端 def __init__(self, server_url: str, agent_id: str): self.server_url server_url self.agent_id agent_id def update_context(self, key: str, data: Any): 更新上下文 # 实际实现应该包含HTTP请求 pass def get_context(self, key: str) - Any: 获取上下文 # 实际实现应该包含HTTP请求 pass4.2 基于MCP的智能体协作模式实现智能体间的协同工作class MCPEnabledAgent: 支持MCP的智能体基类 def __init__(self, agent_id: str, mcp_server: MCPServer): self.agent_id agent_id self.mcp_server mcp_server self.mcp_server.register_agent(agent_id, { capabilities: self.get_capabilities(), type: self.__class__.__name__ }) def get_capabilities(self) - List[str]: 返回智能体能力列表 return [] def process_with_context(self, task: str, context_keys: List[str] None) - Dict: 基于上下文处理任务 # 获取相关上下文 context {} if context_keys: for key in context_keys: context[key] self.mcp_server.retrieve_context(key) # 处理任务 result self.process_task(task, context) # 更新上下文 self.mcp_server.store_context( fresult_{self.agent_id}_{datetime.now().timestamp()}, result, self.agent_id ) return result def process_task(self, task: str, context: Dict) - Dict: 具体任务处理逻辑由子类实现 raise NotImplementedError5. 医疗RAG知识库构建与实践5.1 医疗知识数据收集与处理构建高质量的医疗RAG系统需要可靠的数据源import pandas as pd from sentence_transformers import SentenceTransformer import faiss import numpy as np import json class MedicalKnowledgeBase: 医疗知识库管理 def __init__(self, embedding_model_name: str all-MiniLM-L6-v2): self.embedding_model SentenceTransformer(embedding_model_name) self.index None self.knowledge_data [] def load_medical_data(self, data_path: str): 加载医疗数据 # 支持多种格式的数据加载 if data_path.endswith(.json): with open(data_path, r, encodingutf-8) as f: data json.load(f) elif data_path.endswith(.csv): data pd.read_csv(data_path).to_dict(records) else: raise ValueError(不支持的数据格式) self.knowledge_data data self._build_index() def _build_index(self): 构建向量索引 if not self.knowledge_data: raise ValueError(没有数据可构建索引) # 提取文本内容 texts [self._extract_text(item) for item in self.knowledge_data] # 生成嵌入向量 embeddings self.embedding_model.encode(texts) # 创建FAISS索引 dimension embeddings.shape[1] self.index faiss.IndexFlatIP(dimension) # 内积相似度 # 归一化向量以便使用内积相似度 faiss.normalize_L2(embeddings) self.index.add(embeddings) print(f知识库索引构建完成共{len(self.knowledge_data)}条数据) def _extract_text(self, item: Dict) - str: 从数据项中提取文本内容 # 根据实际数据结构调整 if content in item: return item[content] elif text in item: return item[text] else: return str(item) def search(self, query: str, top_k: int 5) - List[Dict]: 语义搜索 if self.index is None: raise ValueError(索引未构建) # 生成查询向量 query_embedding self.embedding_model.encode([query]) faiss.normalize_L2(query_embedding) # 搜索 similarities, indices self.index.search(query_embedding, top_k) # 返回结果 results [] for i, idx in enumerate(indices[0]): if idx len(self.knowledge_data): results.append({ data: self.knowledge_data[idx], similarity: float(similarities[0][i]) }) return results # 示例医疗数据 sample_medical_data [ { id: 1, disease: 感冒, symptoms: [流鼻涕, 咳嗽, 发热, 喉咙痛], treatment: 休息、多喝水、对症用药, precautions: 如发热超过38.5℃或症状持续3天以上需就医 }, { id: 2, disease: 高血压, symptoms: [头痛, 眩晕, 心悸], treatment: 规律服药、低盐饮食、适量运动, precautions: 需定期监测血压不可擅自停药 } ]5.2 RAG增强的医疗问答智能体集成RAG的医疗问答智能体class RAGMedicalAgent: RAG增强的医疗问答智能体 def __init__(self, knowledge_base: MedicalKnowledgeBase): self.knowledge_base knowledge_base self.llm ChatOpenAI( modelConfig.GPT_MODEL, temperature0.1, api_keyConfig.OPENAI_API_KEY ) def answer_question(self, question: str) - Dict: 回答医疗问题 # 1. 检索相关知识 relevant_knowledge self.knowledge_base.search(question, top_k3) # 2. 构建提示词 context_text self._format_context(relevant_knowledge) prompt f你是一个专业的医疗顾问。基于以下医学知识回答问题 相关医学知识 {context_text} 用户问题{question} 请根据以上知识回答问题并注意 1. 保持专业和谨慎 2. 如信息不足明确说明局限性 3. 始终建议用户咨询专业医生 4. 不要提供具体的用药剂量 messages [HumanMessage(contentprompt)] response self.llm.invoke(messages) return { answer: response.content, sources: [kb[data] for kb in relevant_knowledge], confidence: min([kb[similarity] for kb in relevant_knowledge]) if relevant_knowledge else 0 } def _format_context(self, knowledge_list: List[Dict]) - str: 格式化检索到的知识 context_parts [] for i, kb in enumerate(knowledge_list, 1): data kb[data] context_parts.append(f{i}. {data.get(disease, 未知疾病)}: ) context_parts.append(f 症状: {, .join(data.get(symptoms, []))}) context_parts.append(f 治疗: {data.get(treatment, )}) context_parts.append(f 注意事项: {data.get(precautions, )}) return \n.join(context_parts)6. 完整医疗多智能体项目实战6.1 项目架构设计构建完整的医疗咨询多智能体系统class MedicalMultiAgentSystem: 医疗多智能体系统 def __init__(self): # 初始化组件 self.mcp_server MCPServer() self.knowledge_base MedicalKnowledgeBase() self.workflow build_medical_graph() # 初始化智能体 self.agents { symptom_analyzer: SymptomAnalyzerAgent(), drug_knowledge: DrugKnowledgeAgent(), rag_consultant: RAGMedicalAgent(self.knowledge_base) } def load_knowledge_base(self, data_path: str): 加载知识库数据 self.knowledge_base.load_medical_data(data_path) def process_medical_query(self, patient_query: str) - Dict: 处理医疗查询 # 初始化状态 initial_state MedicalState( patient_querypatient_query, medical_history[], current_symptoms[], preliminary_diagnosis, drug_recommendations[], final_advice, error_messages[] ) try: # 执行工作流 final_state self.workflow.invoke(initial_state) return { success: True, preliminary_diagnosis: final_state[preliminary_diagnosis], drug_recommendations: final_state[drug_recommendations], final_advice: final_state[final_advice], errors: final_state[error_messages] } except Exception as e: return { success: False, error: str(e), preliminary_diagnosis: , drug_recommendations: [], final_advice: 系统处理失败请稍后重试或联系管理员 }6.2 系统集成与测试创建完整的测试用例def test_medical_system(): 测试医疗多智能体系统 # 初始化系统 medical_system MedicalMultiAgentSystem() # 加载示例数据 sample_data [ { disease: 偏头痛, symptoms: [单侧头痛, 恶心, 畏光], treatment: 休息、避免触发因素、对症止痛, precautions: 如头痛频繁发作或加重需神经科就诊 } ] # 保存示例数据到临时文件 import tempfile import json with tempfile.NamedTemporaryFile(modew, suffix.json, deleteFalse) as f: json.dump(sample_data, f, ensure_asciiFalse) temp_path f.name medical_system.load_knowledge_base(temp_path) # 测试查询 test_queries [ 我最近经常一侧头痛还恶心是什么情况, 感冒了应该吃什么药, 高血压患者平时要注意什么 ] for query in test_queries: print(f\n 测试查询: {query} ) result medical_system.process_medical_query(query) if result[success]: print(初步诊断:, result[preliminary_diagnosis]) print(药品推荐:, result[drug_recommendations]) print(最终建议:, result[final_advice]) else: print(处理失败:, result[error]) # 清理临时文件 import os os.unlink(temp_path) if __name__ __main__: test_medical_system()6.3 性能优化与监控添加系统监控和性能优化功能import time from functools import wraps from datetime import datetime def monitor_performance(func): 性能监控装饰器 wraps(func) def wrapper(*args, **kwargs): start_time time.time() result func(*args, **kwargs) end_time time.time() execution_time end_time - start_time print(f{func.__name__} 执行时间: {execution_time:.2f}秒) # 可以记录到日志或监控系统 with open(performance.log, a) as f: f.write(f{datetime.now()},{func.__name__},{execution_time:.2f}\n) return result return wrapper class OptimizedMedicalSystem(MedicalMultiAgentSystem): 优化版的医疗系统 monitor_performance def process_medical_query(self, patient_query: str) - Dict: 性能监控的查询处理 return super().process_medical_query(patient_query) def batch_process_queries(self, queries: List[str]) - List[Dict]: 批量处理查询并行优化 from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor(max_workers3) as executor: results list(executor.map(self.process_medical_query, queries)) return results7. 常见问题与解决方案7.1 LangGraph开发常见问题问题1状态管理混乱现象状态在不同节点间传递时数据丢失或混乱原因状态结构定义不清晰或节点修改了不该修改的状态字段解决方案使用TypedDict明确定义状态结构每个节点只修改自己负责的状态字段添加状态验证逻辑def validate_medical_state(state: MedicalState) - bool: 验证状态完整性 required_fields [patient_query, medical_history, current_symptoms] return all(field in state for field in required_fields)问题2智能体间通信失败现象智能体无法正确接收或处理其他智能体的输出原因消息格式不统一或通信协议实现错误解决方案定义标准的消息格式实现重试机制添加通信日志7.2 MCP协议实施问题问题1上下文数据不一致现象不同智能体看到的上下文数据不一致原因并发访问导致的数据竞争解决方案实现乐观锁或悲观锁机制添加版本控制使用事务性更新问题2性能瓶颈现象MCP服务器响应缓慢原因频繁的上下文更新或大量消息传递解决方案实现上下文缓存批量处理消息使用更高效的数据序列化格式7.3 RAG系统优化问题问题1检索结果不相关现象RAG系统返回的知识与问题不匹配原因嵌入模型不适合医疗领域或知识库质量差解决方案使用领域特定的嵌入模型优化知识库数据质量实现重排序机制问题2生成结果不准确现象即使检索到相关知识生成答案仍不准确原因提示词设计不合理或模型理解偏差解决方案优化提示词设计添加事实核查步骤实现多模型验证8. 生产环境部署与最佳实践8.1 安全考虑与合规要求医疗AI系统需要特别注意安全和合规class SecurityManager: 安全管理器 def __init__(self): self.sensitive_keywords [密码, 身份证, 病历号, 社保号] def sanitize_input(self, text: str) - str: 输入清洗 # 移除敏感信息 for keyword in self.sensitive_keywords: text text.replace(keyword, [已过滤]) return text def validate_medical_advice(self, advice: str) - bool: 验证医疗建议的安全性 dangerous_patterns [ 肯定能治好, 绝对有效, 不用去医院, 自己动手, 绝对安全, 无副作用 ] return not any(pattern in advice for pattern in dangerous_patterns) class CompliantMedicalSystem(MedicalMultiAgentSystem): 符合合规要求的医疗系统 def __init__(self): super().__init__() self.security_manager SecurityManager() def process_medical_query(self, patient_query: str) - Dict: 安全的查询处理 # 输入清洗 sanitized_query self.security_manager.sanitize_input(patient_query) # 处理查询 result super().process_medical_query(sanitized_query) # 输出验证 if result[success]: is_safe self.security_manager.validate_medical_advice(result[final_advice]) if not is_safe: result[final_advice] 建议咨询专业医生获取个性化诊断 result[needs_professional_review] True return result8.2 性能优化策略数据库优化class OptimizedKnowledgeBase(MedicalKnowledgeBase): 优化版知识库 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.query_cache {} # 查询缓存 self.cache_size 1000 def search(self, query: str, top_k: int 5) - List[Dict]: 带缓存的搜索 cache_key f{query}_{top_k} if cache_key in self.query_cache: return self.query_cache[cache_key] # 执行搜索 results super().search(query, top_k) # 更新缓存 if len(self.query_cache) self.cache_size: # 简单的LRU缓存淘汰 self.query_cache.pop(next(iter(self.query_cache))) self.query_cache[cache_key] results return results异步处理优化import asyncio class AsyncMedicalSystem(MedicalMultiAgentSystem): 异步版医疗系统 async def process_medical_query_async(self, patient_query: str) - Dict: 异步处理查询 # 将同步方法转换为异步 loop asyncio.get_event_loop() return await loop.run_in_executor( None, self.process_medical_query, patient_query )8.3 监控与日志系统实现完整的监控体系import logging from logging.handlers import RotatingFileHandler def setup_logging(): 配置日志系统 logger logging.getLogger(medical_agent) logger.setLevel(logging.INFO) # 文件处理器 file_handler RotatingFileHandler( medical_agent.log, maxBytes10*1024*1024, # 10MB backupCount5 ) # 控制台处理器 console_handler logging.StreamHandler() # 格式器 formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s ) file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger # 使用示例 logger setup_logging() class LoggedMedicalSystem(MedicalMultiAgentSystem): 带日志记录的医疗系统 def process_medical_query(self, patient_query: str) - Dict: 记录日志的查询处理 logger.info(f开始处理查询: {patient_query[:100]}...) try: result super().process_medical_query(patient_query) logger
