【2026.04.09】AI问诊助手技术架构全解析：从RAG到智能分诊的落地实践

from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

 1. 加载中文Embedding模型
model = SentenceTransformer("moka-ai/m3e-base")

 2. 医学知识文档库（实际生产环境会从医学文献中抽取）
docs = [
    "发烧超过38.5度持续三天建议就医",
    "胸闷胸痛可能与心血管疾病有关，需排查冠心病",
    "儿童咳嗽超过一周需排查肺炎，建议就医",
    "空腹血糖≥7.0 mmol/L可诊断为糖尿病"
]

 3. 向量化并构建FAISS索引
embeddings = model.encode(docs)
index = faiss.IndexFlatL2(768)   768维向量
index.add(np.array(embeddings).astype("float32"))

 4. 问题检索函数
def search_knowledge(query: str, top_k: int = 3):
    q_emb = model.encode([query])
    D, I = index.search(np.array(q_emb).astype("float32"), top_k)
    return [docs[i] for i in I[0]]

 5. 构建安全Prompt
def build_safe_prompt(question: str, knowledge: list):
    context = "\n".join(knowledge)
    return f"""
你是一名专业医生助理，只能依据以下医学资料回答，禁止编造信息：
【资料】
{context}
【问题】
{question}
请给出安全、保守、医学合规的建议。
"""

用户层（小程序/App/H5）
    ↓
问诊对话服务（会话管理、上下文追踪）
    ↓
AI能力层
├─ 大模型推理（LLM）
├─ 医疗知识库（RAG）
├─ 症状识别NLP
└─ 分诊规则引擎
    ↓
业务系统层
├─ 医生排班 │ 挂号系统
├─ 电子病历 │ 处方系统
└─ 支付系统

import redis
import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def save_session(uid: str, msg: dict):
    key = f"chat:{uid}"
    history = r.get(key)
    history = json.loads(history) if history else []
    history.append(msg)
    r.set(key, json.dumps(history), ex=3600)   1小时过期

def get_session(uid: str):
    history = r.get(f"chat:{uid}")
    return json.loads(history) if history else []

import re

SYMPTOM_RULES = {
    "发烧": r"发烧|高烧|发热|体温升高",
    "咳嗽": r"咳嗽|咳|干咳|咳痰",
    "胸痛": r"胸痛|胸闷|心口疼",
    "头痛": r"头痛|头疼|偏头痛"
}

def extract_symptoms(text: str):
    symptoms = []
    for symptom, pattern in SYMPTOM_RULES.items():
        if re.search(pattern, text):
            symptoms.append(symptom)
    return symptoms

def triage_decision(symptoms: list):
    """根据症状决定分流方向"""
    urgent_symptoms = {"胸痛", "呼吸困难", "大出血", "意识不清"}
    moderate_symptoms = {"高烧", "剧烈头痛", "持续呕吐"}
    
    if any(s in urgent_symptoms for s in symptoms):
        return {"level": "紧急", "action": "立即转人工急诊"}
    elif any(s in moderate_symptoms for s in symptoms):
        return {"level": "中等", "action": "推荐专科问诊"}
    else:
        return {"level": "普通", "action": "AI自助问答"}