医疗知识问答系统

┌──────────────────────────────────────────────────────┐
│               医生用户界面 (React)                     │
│  ┌──────────────────────────────────────────────┐    │
│  │  输入: "糖尿病合并肾功能不全应选择哪种降糖药？"   │    │
│  └───────────────────┬──────────────────────────┘    │
│                      ▼                                │
├──────────────────────────────────────────────────────┤
│              知识检索层 (LlamaIndex)                   │
│                                                       │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐            │
│  │ 语义检索  │  │ 关键词   │  │ 分层过滤  │            │
│  │ 向量相似  │  │ BM25     │  │ 科室/病种 │            │
│  └─────┬────┘  └────┬─────┘  └─────┬────┘            │
│        └────────────┴──────────────┘                  │
│                         │                             │
│  ┌──────────────────────▼──────────────────────────┐ │
│  │  重排序 (BGE Reranker) → Top-5                   │ │
│  └──────────────────────┬──────────────────────────┘ │
│                         │                             │
├─────────────────────────┴───────────────────────────┤
│              安全校验层                                │
│  ┌────────────┐  ┌────────────┐  ┌──────────────┐   │
│  │ 知识范围   │  │ 合规检查   │  │ 置信度评估    │   │
│  │ 是否在库内 │  │ 敏感内容   │  │ 低于阈值拒答  │   │
│  └────────────┘  └────────────┘  └──────────────┘   │
├─────────────────────────┬───────────────────────────┤
│              答案生成 + 引用标注                       │
│                                                       │
│  ┌──────────────────────────────────────────────┐    │
│  │  LLM 基于检索文档生成 + 自动标注引用序号        │    │
│  │  "根据《中国2型糖尿病防治指南(2023)》第4章...    │    │
│  └──────────────────────────────────────────────┘    │
└──────────────────────────────────────────────────────┘

from llama_index.core import (
    VectorStoreIndex, SimpleDirectoryReader,
    Settings, PromptTemplate
)
from llama_index.vector_stores.pinecone import PineconeVectorStore
from llama_index.core.postprocessor import SentenceTransformerRerank
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.response_synthesizers import TreeSummarize
from pinecone import Pinecone, ServerlessSpec
import openai

# 1. 初始化向量数据库
pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))
index = pc.Index("medical-knowledge-v2")

vector_store = PineconeVectorStore(
    pinecone_index=index,
    namespace="clinical-guidelines"
)

# 2. 构建检索器
retriever = VectorIndexRetriever(
    index=VectorStoreIndex.from_vector_store(vector_store),
    similarity_top_k=15,
    vector_store_query_mode="hybrid",
    alpha=0.5  # 语义与关键词权重均衡
)

# 3. 重排序
reranker = SentenceTransformerRerank(
    model="BAAI/bge-reranker-v2-m3",
    top_n=5
)

# 4. 安全约束 Prompt
MEDICAL_PROMPT = PromptTemplate(
    "你是一位资深临床医生，基于以下参考资料回答问题。
"
    "规则：
"
    "1. 只基于参考资料回答，不要添加任何外界知识
"
    "2. 如果参考资料不足以回答，请说：'根据现有资料无法确定'
"
    "3. 每个关键论断后面必须标注引用来源：[文档名，章节]
"
    "4. 如果问题涉及治疗方案，请注明证据等级

"
    "参考资料：\n{context_str}\n\n"
    "问题：{query_str}\n\n"
    "回答："
)

# 5. 构造查询引擎
query_engine = RetrieverQueryEngine.from_args(
    retriever=retriever,
    node_postprocessors=[reranker],
    response_synthesizer=TreeSummarize(),
    response_mode="compact",
    text_qa_template=MEDICAL_PROMPT,
)

# 6. 安全校验器
class MedicalSafetyValidator:
    """多层安全校验"""

    def check_relevance(self, nodes, query):
        """检查检索结果是否真正相关"""
        scores = [n.score for n in nodes]
        avg_score = sum(scores) / len(scores)
        return avg_score >= 0.65

    def check_sufficiency(self, response):
        """检查回答是否基于足够证据"""
        # 统计引用数量
        citations = len(re.findall(r'\[.*?\]', response))
        return citations >= 2

    def check_prompt_injection(self, query):
        """检测 Prompt 注入攻击"""
        injection_patterns = [
            "忽略", "ignore", "system prompt",
            "扮演", "pretend", "jailbreak"
        ]
        return not any(p in query.lower() for p in injection_patterns)

# 7. 执行查询
validator = MedicalSafetyValidator()

def answer_medical_query(query: str) -> dict:
    if not validator.check_prompt_injection(query):
        return {"answer": "抱歉，该问题超出回答范围。", "safe": False}

    response = query_engine.query(query)

    if not validator.check_relevance(response.source_nodes, query):
        return {"answer": "抱歉，所需信息不在医学知识库中。", "safe": False}

    answer = response.response
    if not validator.check_sufficiency(answer):
        answer += "\n\n⚠️ 注意：以上回答基于有限资料，建议结合临床判断使用。"

    return {
        "answer": answer,
        "sources": [n.metadata.get("source") for n in response.source_nodes],
        "safe": True
    }