LangChain + RAG 构建医疗问答Agent

关于 LangChain

LangChain 是一个用于构建基于大语言模型应用程序的开源框架，旨在帮助开发者更高效地开发、部署和管理 LLM 驱动的应用。

其核心价值在于模块化设计、标准化接口和丰富的工具链，简化从原型设计到生产部署的AI应用开发全流程。它不依赖特定模型，而是提供统一接口对接OpenAI、Hugging Face等主流LLM，并支持自定义模型集成，使开发者能聚焦业务逻辑而非底层技术细节。

它提供了一系列模块化组件和工具，简化了与 LLM 交互、数据处理、工作流编排等任务。

• LangChain：提供七大组件（Agent, Models, Chains, Tools, Memory, Callbacks, RAG）。
• LangGraph：在其之上编排复杂流程，解决两类问题：
  1. 线性链的局限：基础 Chain 难以处理循环/分支。
  2. Agent 的扩展：增强 Agent 的鲁棒性（如自动重试、状态回溯）。

LangChain 主要解决以下问题：

• LLM 集成：支持多种大模型（如 OpenAI GPT、Anthropic、Llama2 等），统一接口调用。

• 上下文管理：处理长文本、多轮对话的上下文限制（如分块、摘要、记忆机制）。

• 数据增强：结合外部数据源（文档、数据库、API）进行检索增强生成（RAG）。

• 工作流编排：通过链（Chain）将多个步骤组合成复杂流程（如问答、摘要、代码生成）。

• 代理（Agent）：让 LLM 动态调用工具（如搜索、计算、自定义函数）完成复杂任务。

使用案例

# 翻译智能体
from fastapi import FastAPI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langserve import add_routes

# 1. Create prompt template
system_template = "Translate the following into {language}:"
prompt_template = ChatPromptTemplate.from_messages([
    ('system', system_template),
    ('user', '{text}')
])

# 2. Create model
model = ChatOpenAI()

# 3. Create parser（输出转换，只过滤需要的内容）
parser = StrOutputParser()

# 4. Create chain
chain = prompt_template | model | parser


# 4. App definition
app = FastAPI(
  title="LangChain Server",
  version="1.0",
  description="A simple API server using LangChain's Runnable interfaces",
)

# 5. Adding chain route
add_routes(
    app,
    chain,
    path="/chain",
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000)

关于 Ollama

Ollama 是一个强大的运行框架，旨在使运行LLM尽可能简单。Ollama 简化了在本地机器或服务器上下载、运行和管理大型语言模型的整个过程。使用 Ollama 很简单，可以在不同的平台上完成安装。

Ollama的API具有许多基本功能，使其成为开发人员的重要选择之一，其主要功能如下：

• 流支持：实时token生成，完全兼容OpenAI API，非常适合创建响应式应用程序。

• 多模型管理： 能够同时运行不同的模型，但有一个警告。 当 VRAM 有限时，Ollama 将停止一个模型来运行另一个模型，这需要仔细的资源规划。

• 参数控制： 通过 API 调用进行高度可定制的设置，它提供了很大的灵活性，但对于初学者和生产环境的服务器来说并不友好。

• CPU 兼容性： 当 VRAM 不足时，智能资源管理可以自动将模型卸载到 CPU执行，使得在 GPU 内存有限的系统上也可以运行大模型服务。

• 语言无关性： 可以自由使用Python、 JavaScript、 Go等编程语言，以及其他任何具有 HTTP 功能的编程语言。

# 安装模型
ollama pull nomic-embed-text
ollama pull qwen2

# 启动ollama
ollama server

# 进入对话模式
ollama run qwen2

# 调用 ollama 的接口
curl --location --request POST 'http://localhost:11434/api/generate' \
--header 'Content-Type: application/json' \
--data-raw '{
    "model": "qwen2",
    "prompt": "Why is the sky blue?",
    "stream": false
}'

关于 Chroma

Chroma 是一种高效的开源向量数据库，专为存储和检索高维向量数据设计，广泛应用于文本嵌入管理和相似性搜索。它通过将文本转化为向量嵌入，支持语义搜索、推荐系统等功能，尤其适合大模型应用场景。

Chroma 的核心基于 HNSW 算法，支持快速的近似最近邻搜索。其主要特点包括：

• 多存储后端：支持 DuckDB 和 ClickHouse 等存储选项。
• 多语言支持：提供 Python 和 JavaScript SDK。
• 简单易用：设计简洁，提升开发效率。
• 高性能：支持快速相似性搜索和结果分析。

使用案例

import chromadb
# 初始化客户端：
client = chromadb.Client()

# 创建集合：
collection = client.create_collection("all-my-documents")

# 添加数据：
collection.add(
    documents=["Document about AI", "Document about food"],
    metadatas=[{"category": "technology"}, {"category": "lifestyle"}],
    ids=["id1", "id2"]
)

# 查询数据，支持通过元数据和文档内容进行过滤查询，按元数据过滤：
results = collection.query(
    query_texts=["AI"],
    n_results=2,
    where={"category": {"$eq": "technology"}}
)
print(results)

# 组合条件查询：按元数据过滤 + 按文档内容过滤
results = collection.query(
    query_texts=["AI"],
    n_results=2,
    where={"$and": [{"category": {"$eq": "technology"}}, {"year": {"$gt": 2021}}]}
)
print(results)


# 更新数据：
collection.update(
    ids=["id1"],
    documents=["Updated document about AI"],
    metadatas=[{"category": "updated technology"}]
)

# 删除数据：
collection.delete(ids=["id1"])

医疗问答 Demo

构建 医疗问答系统的样例。需要安装 Ollama 以及 相关依赖，添加指定数据集。可参考 GitHub 源码

import os
import sys
import glob
import os
import warnings

from langchain_core._api import LangChainDeprecationWarning
warnings.filterwarnings("ignore", category=LangChainDeprecationWarning)

from langchain.agents import initialize_agent, Tool
from langchain.chains import RetrievalQA
# 按照最新的langchain导入规范，从langchain_community导入组件
from langchain_ollama import ChatOllama
from langchain_chroma import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain_community.tools import DuckDuckGoSearchRun

# 添加项目根目录到Python路径，以便导入tools模块
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../')))
# 导入医疗专用工具
from tools.medical_tools import MedicalTools

class MedicalAgent:
    def __init__(self):
        # 初始化语言模型 - 使用Ollama并优化参数以提高响应速度
        self.llm = ChatOllama(
            model="gemma3:4b",  # Ollama中的模型名称，使用更通用的llama3
            temperature=0,
            base_url="http://localhost:11434",  # Ollama默认API地址
            # 增加超时时间以解决连接问题
            timeout=300,
            # 移除JSON格式要求，某些模型可能不支持
            # format="json"
        )
        
        # 加载RAG数据
        self.vectorstore = self.load_medical_knowledge()
        
        # 创建检索链 - 减少检索文档数量以提高速度
        self.retrieval_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.vectorstore.as_retriever(
                search_kwargs={"k": 2}  # 从3减少到2，减少需要处理的文档数量
            ),
            return_source_documents=True
        )
        
        # 创建网络搜索工具
        self.search = DuckDuckGoSearchRun()
        
        # 定义工具列表
        self.tools = [
            Tool(
                name="Medical Knowledge Base",
                func=self.query_medical_knowledge,
                description="适合用来回答医学知识相关的问题，包括疾病、药物、急救和健康生活方式等内容"
            ),
            Tool(
                name="Web Search",
                func=self.search.run,
                description="适合用来搜索最新的医疗信息、研究进展和新闻等互联网信息"
            ),
            Tool(
                name="Symptom Extractor",
                func=self.extract_symptoms,
                description="适合用来从文本中提取症状信息"
            ),
            Tool(
                name="Severity Assessment",
                func=self.assess_severity,
                description="适合用来评估症状的严重程度"
            )
        ]
        
        # 初始化智能体 - 关闭verbose输出以提高响应速度
        self.agent = initialize_agent(
            self.tools,
            self.llm,
            agent="zero-shot-react-description",
            verbose=False,  # 从True改为False，减少输出信息
            handle_parsing_errors=True,
            max_iterations=3,  # 限制最大迭代次数
            early_stopping_method="force"
        )
        
        # 初始化医疗工具
        self.medical_tools = MedicalTools()
    
    def load_medical_knowledge(self):
        """加载医疗知识库数据"""
        # 获取data目录下的所有txt文件
        data_dir = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../../data")
        files = glob.glob(os.path.join(data_dir, "*.txt"))
        
        if not files:
            print("警告: 未找到医疗知识库文件。将创建一个空的向量存储。")
            from langchain_community.embeddings import FakeEmbeddings
            from langchain_core.documents import Document
            # 创建空文档列表并使用from_documents方法初始化Chroma
            empty_docs = [Document(page_content="这是一个空的医疗知识库文档", metadata={"source": "empty"})]
            return Chroma.from_documents(empty_docs, FakeEmbeddings(size=768))
        
        # 加载文档
        documents = []
        for file in files:
            loader = TextLoader(file, encoding="utf-8")
            documents.extend(loader.load())
        
        # 分割文档
        text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
        texts = text_splitter.split_documents(documents)
        
        # 创建向量存储 - 使用FakeEmbeddings避免下载外部模型
        print("使用FakeEmbeddings创建向量存储...")
        from langchain_community.embeddings import FakeEmbeddings
        vectorstore = Chroma.from_documents(texts, FakeEmbeddings(size=768))
        
        return vectorstore
    
    def query_medical_knowledge(self, query):
        """查询医疗知识库 - 优化查询逻辑以提高响应速度"""
        try:
            # 快速查询
            result = self.retrieval_chain.invoke(query)
            
            # 格式化回答，包含来源信息
            answer = result["result"]
            
            # 简化处理逻辑，减少不必要的操作
            sources = set()
            for doc in result["source_documents"]:
                if "source" in doc.metadata:
                    sources.add(os.path.basename(doc.metadata["source"]))
            
            formatted_sources = "\n来源: " + ", ".join(sources) if sources else ""
            
            # 直接返回回答，减少额外处理
            return f"{answer}{formatted_sources}"
        except Exception as e:
            print(f"知识库查询错误: {e}")
            return f"知识库查询失败: {str(e)}"
        
    def extract_symptoms(self, text):
        """提取症状信息"""
        symptoms = self.medical_tools.extract_symptoms(text)
        if symptoms:
            return f"提取到的症状: {', '.join(symptoms)}"
        else:
            return "未提取到明显症状"
            
    def assess_severity(self, symptoms_text):
        """评估症状严重程度"""
        # 先从文本中提取症状
        symptoms = self.medical_tools.extract_symptoms(symptoms_text)
        if not symptoms:
            return "未提取到可评估的症状"
        
        # 评估严重程度
        severity = self.medical_tools.assess_severity(symptoms)
        return f"症状: {', '.join(symptoms)}\n{severity}"
    
    def run(self, question):
        """运行智能体回答问题"""
        # 验证医疗查询是否合适
        is_valid, error_msg = self.medical_tools.validate_medical_query(question)
        if not is_valid:
            return error_msg
        
        # 使用invoke方法替代已过时的run方法
        return self.agent.invoke(question)

if __name__ == "__main__":
    # 创建医疗智能体实例
    medical_agent = MedicalAgent()
    
    # 示例问题
    questions = [
        "什么是高血压？如何预防？",
        "老年高血压患者有哪些注意事项？",
        "脑血栓的高危因素有哪些？如何预防？",
        "糖尿病的预防措施有哪些？",
        "老年人如何保持健康的生活方式？",
        "高血压、糖尿病和脑血栓之间有什么关系？",
        "高危人群应该多久进行一次体检？"
    ]
    
    # 运行示例
    for q in questions:
        print(f"\n问题: {q}")
        result = medical_agent.run(q)
        print(f"回答: {result}")