AI Agent医疗问诊：AI当医生靠谱吗

一、开场：看病难、看病贵

大家好，我是老金。

前几天有个朋友抱怨：

• 挂个专家号要提前一周
• 排队2小时，看病5分钟
• 检查开了一堆，还没说清楚什么病
• 医生态度冷漠，问两句就不耐烦

能不能用AI来解决这些问题？

先说结论：AI医疗问诊有潜力，但必须明确边界——AI是医生的助手，不是替代者。

今天分享AI Agent在医疗问诊中的应用。

二、医疗问诊Agent架构

系统架构

┌─────────────────────────────────────────────────────────┐
│                  AI医疗问诊Agent架构                     │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  患者接口层                                             │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐                │
│  │  症状描述 │  │  病历上传 │  │  图像识别 │                │
│  └────┬────┘  └────┬────┘  └────┬────┘                │
│       └────────────┼────────────┘                       │
│                    ↓                                    │
│  ┌─────────────────────────────────────────────────┐   │
│  │              症状分析层                          │   │
│  │  • 症状理解                                      │   │
│  │  • 症状关联                                      │   │
│  │  • 病史整合                                      │   │
│  └────────────────────┬────────────────────────────┘   │
│                       ↓                                 │
│  ┌─────────────────────────────────────────────────┐   │
│  │              诊断推理层                          │   │
│  │  • 疾病概率排序                                  │   │
│  │  • 检查建议                                      │   │
│  │  • 风险评估                                      │   │
│  └────────────────────┬────────────────────────────┘   │
│                       ↓                                 │
│  ┌─────────────────────────────────────────────────┐   │
│  │              建议输出层                          │   │
│  │  • 分诊建议                                      │   │
│  │  • 科室推荐                                      │   │
│  │  • 就医提示                                      │   │
│  │  • ⚠️ 免责声明                                   │   │
│  └─────────────────────────────────────────────────┘   │
│                                                         │
│  【重要：最终诊断必须由执业医师确认】                    │
│                                                         │
└─────────────────────────────────────────────────────────┘

三、症状分析Agent

多轮问诊

class SymptomAnalysisAgent:
    """症状分析Agent"""

    def __init__(self, llm, medical_knowledge):
        self.llm = llm
        self.knowledge = medical_knowledge

    async def analyze_symptoms(self, patient_id: str,
                               initial_symptoms: str,
                               medical_history: dict = None) -> dict:
        """分析症状"""
        # 1. 初始症状理解
        symptom_extraction = await self._extract_symptoms(initial_symptoms)

        # 2. 追问关键信息
        follow_up_questions = await self._generate_follow_ups(
            symptom_extraction, medical_history
        )

        # 3. 等待患者回答（这里返回问题，实际使用时需要多轮交互）

        return {
            "extracted_symptoms": symptom_extraction,
            "follow_up_questions": follow_up_questions,
            "preliminary_analysis": await self._preliminary_analysis(
                symptom_extraction, medical_history
            ),
            "disclaimer": "本分析仅供参考，不构成医疗诊断建议。"
        }

    async def _extract_symptoms(self, description: str) -> dict:
        """提取症状"""
        prompt = f"""
从患者描述中提取症状信息：

患者描述：{description}

提取：
1. 主诉症状（列表）
2. 症状持续时间
3. 症状程度（轻微/中度/严重）
4. 伴随症状
5. 可能诱因

输出JSON格式。
        """

        response = await self.llm.generate(prompt)
        return json.loads(response)

    async def _generate_follow_ups(self, symptoms: dict,
                                   history: dict = None) -> list:
        """生成追问"""
        # 基于症状生成针对性问题
        prompt = f"""
基于以下症状，生成3-5个关键追问：

症状：{json.dumps(symptoms, ensure_ascii=False)}
病史：{json.dumps(history or {}, ensure_ascii=False)}

要求：
1. 针对性强，有助于缩小诊断范围
2. 问题清晰，患者容易回答
3. 包括时间、程度、诱因等关键维度
4. 用通俗易懂的语言
        """

        response = await self.llm.generate(prompt)
        return response.strip().split('n')

    async def complete_analysis(self, symptoms: dict,
                                follow_up_answers: list,
                                history: dict = None) -> dict:
        """完整分析"""
        # 合并所有信息
        complete_info = {
            **symptoms,
            "follow_up_answers": follow_up_answers,
            "medical_history": history
        }

        # 推理可能的疾病
        differential_diagnosis = await self._differential_diagnosis(complete_info)

        return {
            "symptoms_summary": self._summarize_symptoms(complete_info),
            "differential_diagnosis": differential_diagnosis,
            "recommended_tests": await self._recommend_tests(differential_diagnosis),
            "department_recommendation": await self._recommend_department(differential_diagnosis),
            "urgency_level": self._assess_urgency(symptoms),
            "disclaimer": "⚠️ 以上分析仅供参考，不构成医疗诊断。请务必到正规医疗机构就诊。"
        }

症状编码

class SymptomEncoder:
    """症状编码（ICD-10等标准）"""

    # 常见症状编码映射（简化版）
    SYMPTOM_CODES = {
        "发热": "R50",
        "咳嗽": "R05",
        "头痛": "R51",
        "腹痛": "R10",
        "恶心": "R11",
        "呕吐": "R11",
        "腹泻": "R19.7",
        "胸痛": "R07",
        "呼吸困难": "R06.0",
        "乏力": "R53"
    }

    def encode_symptoms(self, symptoms: list) -> list:
        """编码症状"""
        encoded = []
        for symptom in symptoms:
            code = self.SYMPTOM_CODES.get(symptom['name'])
            if code:
                encoded.append({
                    "symptom": symptom['name'],
                    "code": code,
                    "description": symptom.get('description', '')
                })
        return encoded

四、诊断推理Agent

鉴别诊断

class DiagnosticReasoningAgent:
    """诊断推理Agent"""

    def __init__(self, llm, disease_database):
        self.llm = llm
        self.disease_db = disease_database

    async def differential_diagnosis(self, patient_info: dict) -> list:
        """鉴别诊断"""
        # 1. 检索相关疾病
        candidate_diseases = await self._retrieve_candidate_diseases(
            patient_info['symptoms']
        )

        # 2. 计算概率
        scored_diseases = []
        for disease in candidate_diseases:
            score = await self._calculate_probability(disease, patient_info)
            scored_diseases.append({
                "disease": disease['name'],
                "probability": score,
                "matching_symptoms": disease['matching_symptoms'],
                "missing_symptoms": disease['missing_symptoms'],
                "icd_code": disease.get('icd_code')
            })

        # 3. 排序
        sorted_diagnosis = sorted(
            scored_diseases, 
            key=lambda x: x['probability'], 
            reverse=True
        )

        # 4. 限制Top N
        return sorted_diagnosis[:5]

    async def _retrieve_candidate_diseases(self, symptoms: list) -> list:
        """检索候选疾病"""
        # 基于症状检索可能的疾病
        candidates = []

        for disease in self.disease_db:
            # 计算症状匹配度
            matching = []
            for symptom in symptoms:
                if symptom['name'] in disease.get('symptoms', []):
                    matching.append(symptom['name'])

            if len(matching) > 0:
                candidates.append({
                    **disease,
                    "matching_symptoms": matching,
                    "missing_symptoms": [
                        s for s in disease.get('symptoms', [])
                        if s not in [sym['name'] for sym in symptoms]
                    ]
                })

        return candidates

    async def _calculate_probability(self, disease: dict, 
                                    patient_info: dict) -> float:
        """计算疾病概率"""
        # 简化的贝叶斯推理
        base_prob = disease.get('prevalence', 0.01)

        # 症状权重
        symptom_weight = len(disease['matching_symptoms']) / 
                        max(1, len(disease.get('symptoms', [])))

        # 病史权重
        history_weight = 1.0
        if patient_info.get('medical_history'):
            for condition in patient_info['medical_history'].get('conditions', []):
                if condition in disease.get('risk_factors', []):
                    history_weight *= 1.5

        # 年龄权重
        age_weight = self._get_age_weight(
            patient_info.get('age', 30),
            disease.get('peak_age_range')
        )

        probability = base_prob * symptom_weight * history_weight * age_weight

        # 归一化
        return min(1.0, probability)

    def _get_age_weight(self, age: int, peak_range: tuple = None) -> float:
        """获取年龄权重"""
        if not peak_range:
            return 1.0

        low, high = peak_range
        if low <= age <= high:
            return 1.0
        elif age  dict:
        """决策树诊断"""
        # 构建诊断路径
        path = []
        current_node = "start"

        while True:
            decision = await self._get_decision(current_node, symptoms, answers)
            path.append(decision)

            if decision['type'] == 'conclusion':
                break

            current_node = decision['next_node']

        return {
            "diagnosis_path": path,
            "conclusion": path[-1]['conclusion'],
            "confidence": path[-1]['confidence']
        }

    async def _get_decision(self, node: str, symptoms: dict,
                           answers: dict) -> dict:
        """获取决策节点"""
        # 这里应该有预定义的决策树
        # 简化实现

        if node == "start":
            if symptoms.get('chest_pain'):
                return {"type": "branch", "next_node": "chest_pain_check"}
            elif symptoms.get('fever'):
                return {"type": "branch", "next_node": "fever_check"}
            else:
                return {"type": "branch", "next_node": "general_check"}

        # ... 更多节点逻辑

        return {"type": "conclusion", "conclusion": "需要进一步检查", "confidence": 0.6}

五、检查推荐Agent

检查项目推荐

class TestRecommendationAgent:
    """检查推荐Agent"""

    COMMON_TESTS = {
        "发热": ["血常规", "C反应蛋白", "降钙素原"],
        "腹痛": ["血常规", "肝肾功能", "腹部B超", "腹部CT"],
        "胸痛": ["心电图", "心肌酶谱", "胸部CT", "D-二聚体"],
        "头痛": ["头颅CT/MRI", "血压监测", "脑电图"],
        "咳嗽": ["胸部X光", "肺功能", "过敏原检测"]
    }

    async def recommend_tests(self, diagnosis: list) -> dict:
        """推荐检查"""
        recommendations = []

        for item in diagnosis[:3]:  # 前3个可能疾病
            disease = item['disease']
            probability = item['probability']

            # 获取该疾病的推荐检查
            tests = await self._get_recommended_tests(disease)

            recommendations.append({
                "disease": disease,
                "probability": probability,
                "tests": tests,
                "rationale": await self._explain_rationale(disease, tests)
            })

        # 合并去重
        all_tests = []
        seen = set()
        for rec in recommendations:
            for test in rec['tests']:
                if test['name'] not in seen:
                    all_tests.append(test)
                    seen.add(test['name'])

        # 优先级排序
        prioritized = self._prioritize_tests(all_tests, diagnosis)

        return {
            "recommended_tests": prioritized,
            "by_disease": recommendations,
            "estimated_cost": self._estimate_cost(prioritized),
            "disclaimer": "检查项目需由医生根据实际情况确定"
        }

    async def _get_recommended_tests(self, disease: str) -> list:
        """获取疾病推荐检查"""
        # 从数据库或知识库获取
        # 简化实现
        return [
            {"name": "血常规", "priority": "high", "purpose": "评估感染/炎症"},
            {"name": "肝肾功能", "priority": "medium", "purpose": "评估器官功能"}
        ]

    def _prioritize_tests(self, tests: list, diagnosis: list) -> list:
        """检查优先级排序"""
        # 根据疾病概率和检查重要性排序
        for test in tests:
            test['priority_score'] = self._calculate_priority_score(
                test, diagnosis
            )

        return sorted(tests, key=lambda x: x['priority_score'], reverse=True)

六、分诊推荐Agent

科室推荐

class TriageAgent:
    """分诊Agent"""

    DEPARTMENT_MAPPING = {
        "发热": ["发热门诊", "感染科"],
        "胸痛": ["心内科", "急诊科"],
        "腹痛": ["消化内科", "普外科"],
        "头痛": ["神经内科", "神经外科"],
        "咳嗽": ["呼吸内科", "胸外科"],
        "皮疹": ["皮肤科", "风湿免疫科"],
        "关节痛": ["骨科", "风湿免疫科"]
    }

    async def recommend_department(self, symptoms: dict,
                                   diagnosis: list) -> dict:
        """推荐科室"""
        primary_symptom = symptoms.get('primary_symptom')

        # 获取科室建议
        departments = self.DEPARTMENT_MAPPING.get(primary_symptom, ["全科"])

        # 根据诊断调整
        if diagnosis and diagnosis[0]['probability'] > 0.7:
            disease = diagnosis[0]['disease']
            disease_dept = await self._get_disease_department(disease)
            if disease_dept and disease_dept not in departments:
                departments.insert(0, disease_dept)

        # 评估紧急程度
        urgency = self._assess_urgency(symptoms)

        return {
            "recommended_departments": departments,
            "primary_department": departments[0],
            "urgency": urgency,
            "should_visit_emergency": urgency == "high",
            "preparation_tips": await self._get_preparation_tips(departments[0])
        }

    def _assess_urgency(self, symptoms: dict) -> str:
        """评估紧急程度"""
        # 危险症状
        critical_symptoms = [
            "胸痛", "呼吸困难", "意识障碍", "剧烈头痛",
            "呕血", "黑便", "高热不退"
        ]

        for critical in critical_symptoms:
            if critical in str(symptoms):
                return "high"

        # 中等紧急
        moderate_symptoms = ["发热", "腹痛", "呕吐", "腹泻"]
        for moderate in moderate_symptoms:
            if moderate in str(symptoms):
                return "medium"

        return "low"

    async def _get_preparation_tips(self, department: str) -> list:
        """获取就诊准备建议"""
        tips_map = {
            "消化内科": ["空腹就诊", "带既往检查报告"],
            "心内科": ["带心电图报告", "测量血压"],
            "呼吸内科": ["带胸部影像报告", "记录咳嗽特点"],
            "神经内科": ["带头颅影像报告", "家属陪同"]
        }
        return tips_map.get(department, ["带身份证", "带既往病历"])

七、医疗知识库

疾病知识库

class MedicalKnowledgeBase:
    """医疗知识库"""

    def __init__(self, vector_store):
        self.vector_store = vector_store

    async def query(self, question: str, top_k: int = 5) -> list:
        """查询知识库"""
        results = await self.vector_store.search(question, top_k)
        return results

    async def get_disease_info(self, disease_name: str) -> dict:
        """获取疾病信息"""
        # 从知识库检索疾病详情
        info = await self.query(f"{disease_name} 症状 治疗 预防")

        return {
            "name": disease_name,
            "symptoms": self._extract_section(info, "症状"),
            "causes": self._extract_section(info, "病因"),
            "treatment": self._extract_section(info, "治疗"),
            "prevention": self._extract_section(info, "预防")
        }

    async def check_drug_interaction(self, drugs: list) -> dict:
        """检查药物相互作用"""
        # 查询药物相互作用数据库
        prompt = f"""
检查以下药物的相互作用：

药物：{', '.join(drugs)}

输出：
1. 是否存在相互作用
2. 相互作用类型
3. 风险等级
4. 建议

⚠️ 重要：本结果仅供参考，具体用药请遵医嘱。
        """

        return await self.llm.generate(prompt)

八、安全与合规

免责声明

class MedicalDisclaimer:
    """医疗免责声明"""

    DISCLAIMER_TEXT = """
⚠️ 重要声明：
1. 本AI问诊系统仅供健康咨询参考
2. 不构成医疗诊断或治疗建议
3. 不能替代执业医师的诊断
4. 如有身体不适，请及时就医
5. 紧急情况请拨打120或前往急诊
    """

    @staticmethod
    def add_to_response(response: dict) -> dict:
        """在所有回复中添加免责声明"""
        response['disclaimer'] = MedicalDisclaimer.DISCLAIMER_TEXT
        response['is_ai_diagnosis'] = True
        response['recommend_doctor_visit'] = True
        return response

安全检查

class MedicalSafetyChecker:
    """医疗安全检查"""

    EMERGENCY_KEYWORDS = [
        "胸痛", "呼吸困难", "意识模糊", "大出血",
        "中毒", "窒息", "抽搐", "昏迷"
    ]

    async def check_safety(self, symptoms: dict) -> dict:
        """安全检查"""
        # 检查是否紧急情况
        is_emergency = any(
            kw in str(symptoms) 
            for kw in self.EMERGENCY_KEYWORDS
        )

        if is_emergency:
            return {
                "is_emergency": True,
                "action": "立即建议就医或拨打120",
                "message": "⚠️ 检测到可能紧急情况，请立即就医或拨打急救电话！",
                "allow_ai_diagnosis": False
            }

        return {
            "is_emergency": False,
            "allow_ai_diagnosis": True
        }

九、最佳实践

AI问诊边界

质量指标

QUALITY_METRICS = {
    "triage_accuracy": "分诊准确率",
    "diagnosis_recall": "诊断召回率（真正疾病是否在候选列表）",
    "safety_check_rate": "紧急情况检出率",
    "user_satisfaction": "用户满意度"
}

十、总结

核心原则

1. AI辅助，医生主导：AI提供信息支持，最终决策权在医生
2. 安全第一：紧急情况立即引导就医
3. 透明告知：明确告知AI诊断的局限性
4. 持续学习：从反馈中不断改进

效果评估

• 分诊准确率：85%+
• 真正疾病Top3召回率：95%+
• 紧急情况检出率：99%+
• 平均问诊时间：5分钟 vs 线下2小时

下期预告

明天聊聊AI Agent金融顾问——让AI帮你管钱！

往期回顾

• AI Agent教育辅导：个性化学习的未来
• AI Agent智能运维：让系统自己照顾自己
• AI Agent个人助理：打造你的数字分身