基于LangChain与LangGraph构建医疗法律多智能体系统实战指南

最近在医疗AI项目中经常遇到需要同时处理医学诊断建议和法律合规审查的复杂场景。单一模型往往难以兼顾专业性和规范性而多智能体协作正好能解决这一痛点。本文将通过完整的实战案例展示如何基于LangChain和LangGraph构建医疗法律双领域的多智能体系统涵盖从环境搭建到生产部署的全流程。无论你是想入门AI应用开发的学生还是需要落地企业级解决方案的工程师这套方案都能直接复用。我们将从零开始构建一个能同时处理医疗咨询和刑法合规的智能系统每个步骤都提供可运行的代码示例和避坑指南。1. 多智能体系统核心概念与医疗法律场景价值1.1 什么是多智能体系统多智能体系统Multi-Agent System是由多个自治智能体组成的计算系统这些智能体能够相互协作、协调或竞争共同完成复杂任务。在AI大模型应用中每个智能体通常是一个专门化的语言模型实例负责特定领域的任务处理。与传统单模型应用相比多智能体系统具有以下优势专业化分工不同智能体专注不同领域提供更精准的回答错误纠正智能体间可以相互验证和纠正结果任务分解复杂问题被拆解为子任务并行处理知识互补结合不同领域的专业知识1.2 医疗法律双领域协同的必要性在医疗健康领域AI应用必须同时考虑两个方面医疗专业性和法律合规性。单纯的医疗诊断模型可能忽略法律风险而纯粹的法律分析又缺乏医学专业性。多智能体协作能够医疗智能体专注于症状分析、疾病诊断、治疗建议等医学专业内容法律智能体负责审查医疗建议的合规性、隐私保护、医疗事故风险等法律问题协调智能体整合双方意见生成既专业又合规的最终答案这种分工协作特别适合医疗咨询、药物推荐、治疗方案评估等场景能够显著降低医疗AI应用的合规风险。1.3 LangChain与LangGraph的技术定位LangChain是一个用于开发大语言模型应用的框架提供组件化工具链。而LangGraph是建立在LangChain之上的库专门用于构建多智能体工作流和状态机。关键区别LangChain侧重工具链集成和单智能体开发LangGraph专注于多智能体编排和复杂工作流管理关系LangGraph扩展了LangChain的多智能体能力两者可以协同使用2. 环境准备与依赖配置2.1 基础环境要求本项目需要以下基础环境Python 3.8至少8GB内存推荐16GB稳定的网络连接用于调用API# 创建虚拟环境 python -m venv medical_agent_env source medical_agent_env/bin/activate # Linux/Mac # medical_agent_env\Scripts\activate # Windows # 升级pip pip install --upgrade pip2.2 核心依赖安装# 安装LangChain和LangGraph pip install langchain langgraph # 安装社区扩展和工具包 pip install langchain-community langchain-core # 安装向量数据库以Chroma为例 pip install chromadb # 安装医疗专业库 pip install medspacy # 其他工具库 pip install python-dotenv requests beautifulsoup42.3 大模型API配置本项目支持多种大模型API以下以通义千问和OpenAI为例# config.py import os from dotenv import load_dotenv load_dotenv() # API密钥配置 class Config: # 通义千问配置 DASHSCOPE_API_KEY os.getenv(DASHSCOPE_API_KEY) # OpenAI配置备用 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) # 向量数据库路径 VECTOR_STORE_PATH ./vector_store # 医疗知识库路径 MEDICAL_KB_PATH ./medical_knowledge在项目根目录创建.env文件DASHSCOPE_API_KEYyour_dashscope_api_key_here OPENAI_API_KEYyour_openai_api_key_here3. 医疗知识库构建与RAG技术实现3.1 RAG技术原理与医疗场景适配RAGRetrieval-Augmented Generation通过检索相关知识来增强大模型的生成能力在医疗领域尤其重要准确性要求高医疗回答必须基于最新、最权威的医学知识知识更新快医学研究日新月异需要持续更新知识库专业性验证能够追溯答案来源便于专业医生验证# rag_medical.py from langchain.vectorstores import Chroma from langchain.embeddings import HuggingFaceEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.document_loaders import DirectoryLoader, TextLoader class MedicalRAGSystem: def __init__(self, knowledge_base_path): self.knowledge_base_path knowledge_base_path self.embeddings HuggingFaceEmbeddings( model_namesentence-transformers/all-MiniLM-L6-v2 ) self.vector_store None self._initialize_vector_store() def _initialize_vector_store(self): 初始化向量数据库 try: # 尝试加载现有的向量存储 self.vector_store Chroma( persist_directory./chroma_db, embedding_functionself.embeddings ) print(加载现有向量数据库成功) except: print(创建新的向量数据库) self._build_knowledge_base() def _build_knowledge_base(self): 构建医疗知识库 # 加载医疗文档 loader DirectoryLoader( self.knowledge_base_path, glob**/*.txt, loader_clsTextLoader ) documents loader.load() # 文本分割 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200 ) texts text_splitter.split_documents(documents) # 创建向量存储 self.vector_store Chroma.from_documents( documentstexts, embeddingself.embeddings, persist_directory./chroma_db ) print(f知识库构建完成共处理{len(texts)}个文本块)3.2 医疗专业文档处理医疗文档需要特殊处理包括医学术语识别、症状关系提取等# medical_processor.py import medspacy from medspacy.ner import TargetRule from medspacy.visualization import visualize_ent class MedicalDocumentProcessor: def __init__(self): self.nlp medspacy.load() # 添加医疗实体规则 self._add_medical_rules() def _add_medical_rules(self): 添加医疗实体识别规则 rules [ TargetRule(发热, SYMPTOM), TargetRule(咳嗽, SYMPTOM), TargetRule(头痛, SYMPTOM), TargetRule(高血压, DISEASE), TargetRule(糖尿病, DISEASE), ] # 获取实体识别器组件 ner self.nlp.get_pipe(medspacy_target_matcher) ner.add(rules) def extract_medical_entities(self, text): 提取医疗实体 doc self.nlp(text) entities {} for ent in doc.ents: if ent.label_ not in entities: entities[ent.label_] [] entities[ent.label_].append(ent.text) return entities4. 多智能体系统架构设计4.1 智能体角色定义我们的系统包含三个核心智能体# agents.py from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain_core.messages import HumanMessage, SystemMessage class MedicalAgent: def __init__(self, llm, rag_system): self.llm llm self.rag_system rag_system self.prompt self._create_medical_prompt() def _create_medical_prompt(self): 创建医疗智能体提示词 return ChatPromptTemplate.from_messages([ SystemMessage(content你是一名专业的医疗顾问具有丰富的临床经验。 你的职责是 1. 分析用户描述的症状 2. 提供专业的医学建议 3. 推荐合适的治疗方案 4. 提醒必要的医疗检查 重要原则 - 基于最新医学知识 - 保守治疗优先 - 强调专业医疗咨询的重要性 - 不替代医生诊断), HumanMessage(content{question}) ]) async def process_question(self, question): 处理医疗问题 # 检索相关知识 relevant_docs self.rag_system.vector_store.similarity_search( question, k3 ) # 构建增强的提示词 context \n.join([doc.page_content for doc in relevant_docs]) enhanced_question f 基于以下医疗知识 {context} 用户问题{question} response await self.llm.ainvoke( self.prompt.format(questionenhanced_question) ) return response.content class LegalAgent: def __init__(self, llm): self.llm llm self.prompt self._create_legal_prompt() def _create_legal_prompt(self): 创建法律智能体提示词 return ChatPromptTemplate.from_messages([ SystemMessage(content你是一名专业的法律顾问精通医疗相关法律法规。 你的职责是 1. 审查医疗建议的法律合规性 2. 识别潜在的医疗法律风险 3. 提供风险防范建议 4. 确保符合医疗伦理规范 重点关注 - 医疗事故责任 - 患者隐私保护 - 医疗广告合规 - 处方药物法律规定), HumanMessage(content{advice}) ]) async def review_advice(self, medical_advice): 审查医疗建议的法律合规性 response await self.llm.ainvoke( self.prompt.format(advicemedical_advice) ) return response.content4.2 LangGraph工作流设计# workflow.py from langgraph.graph import StateGraph, END from typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): question: str medical_advice: str legal_review: str final_answer: str iteration_count: int class MedicalLegalWorkflow: def __init__(self, medical_agent, legal_agent, coordinator_agent): self.medical_agent medical_agent self.legal_agent legal_agent self.coordinator_agent coordinator_agent self.graph self._build_graph() def _build_graph(self): 构建工作流图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(medical_analysis, self.medical_analysis) workflow.add_node(legal_review, self.legal_review) workflow.add_node(coordinate_results, self.coordinate_results) workflow.add_node(refine_advice, self.refine_advice) # 设置入口点 workflow.set_entry_point(medical_analysis) # 添加边 workflow.add_edge(medical_analysis, legal_review) workflow.add_conditional_edges( legal_review, self.should_refine, { refine: refine_advice, approve: coordinate_results } ) workflow.add_edge(refine_advice, legal_review) workflow.add_edge(coordinate_results, END) return workflow.compile() async def medical_analysis(self, state: AgentState): 医疗分析节点 question state[question] medical_advice await self.medical_agent.process_question(question) return {medical_advice: medical_advice} async def legal_review(self, state: AgentState): 法律审查节点 medical_advice state[medical_advice] legal_review await self.legal_agent.review_advice(medical_advice) return {legal_review: legal_review} def should_refine(self, state: AgentState): 判断是否需要优化建议 legal_review state[legal_review] # 简单的关键词检测实际项目应使用更复杂逻辑 risk_keywords [高风险, 不合规, 违法, 禁止] if any(keyword in legal_review for keyword in risk_keywords): return refine return approve async def refine_advice(self, state: AgentState): 优化建议节点 # 基于法律反馈优化医疗建议 current_advice state[medical_advice] legal_issues state[legal_review] refinement_prompt f 原始医疗建议{current_advice} 法律审查意见{legal_issues} 请优化医疗建议解决上述法律问题同时保持医疗专业性。 refined_advice await self.medical_agent.process_question(refinement_prompt) return { medical_advice: refined_advice, iteration_count: state.get(iteration_count, 0) 1 } async def coordinate_results(self, state: AgentState): 协调最终结果 final_answer f ## 医疗专业建议 {state[medical_advice]} ## 法律合规说明 {state[legal_review]} ## 重要提示 本建议基于AI分析生成不能替代专业医疗诊断和法律咨询。 如有具体健康问题或法律需求请咨询相关专业人士。 return {final_answer: final_answer}5. 完整实战案例医疗咨询与法律审查系统5.1 系统初始化与配置# main.py import asyncio from langchain_community.llms import Tongyi from config import Config from rag_medical import MedicalRAGSystem from agents import MedicalAgent, LegalAgent from workflow import MedicalLegalWorkflow class MedicalLegalAssistant: def __init__(self): self.config Config() self.llm self._initialize_llm() self.rag_system MedicalRAGSystem(self.config.MEDICAL_KB_PATH) self.medical_agent MedicalAgent(self.llm, self.rag_system) self.legal_agent LegalAgent(self.llm) self.workflow MedicalLegalWorkflow( self.medical_agent, self.legal_agent, self.llm ) def _initialize_llm(self): 初始化大语言模型 return Tongyi( modelqwen-plus, dashscope_api_keyself.config.DASHSCOPE_API_KEY, temperature0.1 # 医疗场景需要较低随机性 ) async def process_query(self, question): 处理用户查询 initial_state { question: question, medical_advice: , legal_review: , final_answer: , iteration_count: 0 } result await self.workflow.ainvoke(initial_state) return result[final_answer] # 使用示例 async def main(): assistant MedicalLegalAssistant() # 测试医疗咨询 question 患者出现持续发热和咳嗽症状应该怎么处理有哪些法律风险需要注意 result await assistant.process_query(question) print(最终回答) print(result) if __name__ __main__: asyncio.run(main())5.2 医疗知识库数据准备创建医疗知识文档medical_knowledge/common_diseases.txt发热处理指南 - 体温超过38.5℃建议物理降温 - 持续高热需就医检查血常规 - 注意鉴别感染性发热和非感染性发热 咳嗽诊疗原则 - 区分干咳和湿咳 - 关注咳嗽持续时间 - 排除肺部严重疾病 高血压管理 - 定期监测血压 - 生活方式干预 - 药物治疗的适应症和禁忌症 法律合规要点 - 医疗建议必须注明不能替代专业诊断 - 禁止提供具体的药物剂量建议 - 必须提醒患者及时就医5.3 系统运行与测试# test_system.py import asyncio from main import MedicalLegalAssistant async def test_medical_queries(): assistant MedicalLegalAssistant() test_cases [ 感冒发烧应该吃什么药, 高血压患者日常需要注意什么, 糖尿病患者可以吃水果吗, 怀孕期间用药有哪些禁忌 ] for i, question in enumerate(test_cases, 1): print(f\n{*50}) print(f测试案例 {i}: {question}) print(f{*50}) try: result await assistant.process_query(question) print(result) print(f{*50}) except Exception as e: print(f处理错误: {e}) if __name__ __main__: asyncio.run(test_medical_queries())6. 高级特性与优化方案6.1 智能体记忆与上下文管理# memory_manager.py from langchain.memory import ConversationBufferWindowMemory class AgentMemoryManager: def __init__(self, k5): self.medical_memory ConversationBufferWindowMemory(kk) self.legal_memory ConversationBufferWindowMemory(kk) self.conversation_history [] def add_interaction(self, role, content): 记录交互历史 self.conversation_history.append({ role: role, content: content, timestamp: asyncio.get_event_loop().time() }) def get_recent_context(self, n3): 获取最近上下文 return self.conversation_history[-n:] if self.conversation_history else []6.2 性能优化与缓存策略# cache_manager.py import hashlib import pickle from datetime import datetime, timedelta class ResponseCache: def __init__(self, ttl_hours24): self.cache {} self.ttl timedelta(hoursttl_hours) def _get_hash(self, question): 生成问题哈希值 return hashlib.md5(question.encode()).hexdigest() def get_cached_response(self, question): 获取缓存响应 question_hash self._get_hash(question) cached self.cache.get(question_hash) if cached and datetime.now() - cached[timestamp] self.ttl: return cached[response] return None def cache_response(self, question, response): 缓存响应 question_hash self._get_hash(question) self.cache[question_hash] { response: response, timestamp: datetime.now() }7. 生产环境部署方案7.1 Docker容器化部署创建DockerfileFROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ g \ rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建知识库目录 RUN mkdir -p medical_knowledge vector_store # 暴露端口 EXPOSE 8000 # 启动命令 CMD [python, -m, uvicorn, api:app, --host, 0.0.0.0, --port, 8000]创建docker-compose.ymlversion: 3.8 services: medical-agent-api: build: . ports: - 8000:8000 environment: - DASHSCOPE_API_KEY${DASHSCOPE_API_KEY} - OPENAI_API_KEY${OPENAI_API_KEY} volumes: - ./medical_knowledge:/app/medical_knowledge - ./vector_store:/app/vector_store restart: unless-stopped # 可选的监控服务 monitor: image: prom/prometheus ports: - 9090:9090 volumes: - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml7.2 API接口设计# api.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from main import MedicalLegalAssistant import asyncio app FastAPI(title医疗法律多智能体API) class QueryRequest(BaseModel): question: str user_id: str None session_id: str None class QueryResponse(BaseModel): answer: str session_id: str processing_time: float # 全局助手实例 assistant None app.on_event(startup) async def startup_event(): global assistant assistant MedicalLegalAssistant() app.post(/query, response_modelQueryResponse) async def process_query(request: QueryRequest): try: start_time asyncio.get_event_loop().time() answer await assistant.process_query(request.question) processing_time asyncio.get_event_loop().time() - start_time return QueryResponse( answeranswer, session_idrequest.session_id or default, processing_timeprocessing_time ) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health_check(): return {status: healthy, service: medical-legal-agent}8. 常见问题与解决方案8.1 模型响应问题排查问题现象可能原因解决方案响应速度慢API限流或网络问题增加超时设置实现请求重试机制回答质量下降提示词不够明确优化系统提示词增加约束条件法律审查过于严格风险检测阈值过高调整风险关键词和判断逻辑8.2 知识库更新策略# knowledge_updater.py import schedule import time from datetime import datetime class KnowledgeUpdater: def __init__(self, rag_system): self.rag_system rag_system def scheduled_update(self): 定时更新知识库 # 每天凌晨2点更新 schedule.every().day.at(02:00).do(self._update_knowledge_base) while True: schedule.run_pending() time.sleep(60) def _update_knowledge_base(self): 执行知识库更新 print(f{datetime.now()}: 开始更新知识库) # 这里可以添加从权威医学网站抓取最新指南的逻辑 # 例如CDC、WHO、国家卫健委等 print(知识库更新完成)8.3 性能监控与日志# monitoring.py import logging from prometheus_client import Counter, Histogram, start_http_server # 指标定义 QUERY_COUNTER Counter(medical_queries_total, Total medical queries) PROCESSING_TIME Histogram(query_processing_seconds, Query processing time) class Monitoring: def __init__(self, port8001): self.logger self._setup_logging() start_http_server(port) def _setup_logging(self): 设置日志配置 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s ) return logging.getLogger(__name__) PROCESSING_TIME.time() def record_query(self, question, successTrue): 记录查询指标 QUERY_COUNTER.inc() self.logger.info(fProcessed query: {question[:100]}... Success: {success})9. 最佳实践与工程建议9.1 安全与合规性考虑数据隐私保护用户医疗数据必须加密存储遵循HIPAA/GDPR等隐私法规定期进行安全审计医疗责任边界明确告知用户AI建议的局限性禁止提供具体药物剂量强制要求严重症状立即就医法律风险防控定期更新法律知识库建立法律咨询复核机制购买专业责任保险9.2 性能优化策略缓存策略常见问题答案缓存向量检索结果缓存API响应缓存异步处理使用异步IO提高并发能力智能体间并行处理批量处理相似请求资源管理监控API使用量实现请求限流优雅降级机制9.3 可维护性设计模块化架构智能体职责单一化工作流可配置易于扩展新领域配置化管理外部化所有配置参数环境特定的配置管理版本控制配置变更测试策略单元测试覆盖核心逻辑集成测试验证工作流端到端测试真实场景这套多智能体系统架构不仅适用于医疗法律场景还可以扩展到金融合规、教育咨询、技术支持等多个领域。关键在于明确各智能体的专业边界设计有效的工作流协调机制。在实际项目中建议先从简单场景开始验证逐步增加智能体数量和复杂度。同时要建立完善的监控和评估体系确保系统在实际使用中的可靠性和安全性。