企业官网建设流程全解析

在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（模型上下文协议）详解

MCP（Model Context Protocol）是一种标准化的协议，用于在不同AI模型和工具之间传递上下文信息。在医疗多智能体系统中，MCP确保各个智能体能够共享患者信息、诊断历史、药品知识等关键数据。

MCP协议的核心组件包括：

• 上下文容器：统一格式的数据存储结构
• 消息格式：标准化的智能体间通信协议
• 权限控制：确保敏感医疗数据的安全访问
• 版本管理：支持不同版本模型和工具的协同工作

1.4 RAG在医疗领域的特殊价值

RAG（Retrieval-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 langgraph==0.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 pytest

2.3 开发环境验证

创建环境验证脚本verify_environment.py：

#!/usr/bin/env python3 """ 环境验证脚本：检查所有必要依赖是否正常安装 """ import importlib import sys def check_package(package_name, import_name=None): """检查包是否可正常导入""" 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_KEY=your_openai_api_key_here # 其他服务API密钥（如有） ANTHROPIC_API_KEY=your_anthropic_api_key SERPER_API_KEY=your_serper_api_key # 应用配置 DEBUG=True LOG_LEVEL=INFO

对应的配置读取代码config.py：

import 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_ok=True) os.makedirs(cls.VECTOR_DB_PATH, exist_ok=True) os.makedirs(cls.LOG_DIR, exist_ok=True) # 初始化时验证配置 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( model=Config.GPT_MODEL, temperature=0.1, # 医疗场景需要低随机性 api_key=Config.OPENAI_API_KEY ) self.system_prompt = """你是一个专业的医疗症状分析专家。你的任务是： 1. 分析患者描述的症状 2. 识别可能的疾病方向 3. 评估紧急程度 4. 建议下一步检查方向 请保持专业、谨慎，始终建议患者咨询专业医生。""" def analyze(self, state: MedicalState) -> MedicalState: """分析症状""" try: messages = [ SystemMessage(content=self.system_prompt), HumanMessage(content=f"患者症状描述：{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( model=Config.GPT_MODEL, temperature=0.1, api_key=Config.OPENAI_API_KEY ) def search_drugs(self, symptoms: List[str], diagnosis: str) -> List[Dict]: """根据症状和诊断搜索相关药品""" # 这里可以集成真实的药品数据库 prompt = f""" 基于以下信息推荐药品： - 症状：{', '.join(symptoms)} - 初步诊断：{diagnosis} 请推荐3-5种相关药品，包含： - 药品名称 - 适用症状 - 注意事项 - 是否需要处方 注意：必须明确说明这些只是参考信息，实际用药需遵医嘱。 """ messages = [ HumanMessage(content=prompt) ] 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 state

3.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': f"msg_{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请求 pass

4.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( f"result_{self.agent_id}_{datetime.now().timestamp()}", result, self.agent_id ) return result def process_task(self, task: str, context: Dict) -> Dict: """具体任务处理逻辑（由子类实现）""" raise NotImplementedError

5. 医疗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', encoding='utf-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( model=Config.GPT_MODEL, temperature=0.1, api_key=Config.OPENAI_API_KEY ) def answer_question(self, question: str) -> Dict: """回答医疗问题""" # 1. 检索相关知识 relevant_knowledge = self.knowledge_base.search(question, top_k=3) # 2. 构建提示词 context_text = self._format_context(relevant_knowledge) prompt = f"""你是一个专业的医疗顾问。基于以下医学知识回答问题： 相关医学知识： {context_text} 用户问题：{question} 请根据以上知识回答问题，并注意： 1. 保持专业和谨慎 2. 如信息不足，明确说明局限性 3. 始终建议用户咨询专业医生 4. 不要提供具体的用药剂量""" messages = [HumanMessage(content=prompt)] 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_query=patient_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(mode='w', suffix='.json', delete=False) as f: json.dump(sample_data, f, ensure_ascii=False) 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_workers=3) as executor: results = list(executor.map(self.process_medical_query, queries)) return results

7. 常见问题与解决方案

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 result

8.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', maxBytes=10*1024*1024, # 10MB backupCount=5 ) # 控制台处理器 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