接续个人博客2的展望，我的核心工作是将前期的Mock逻辑全面替换为基于真实大语言模型的智能体网络。这项工作是一个系统工程，涵盖了从底层的API封装、中层的Prompt编写，再到上层业务逻辑编排与接口容错的全过程。

目录

一、 核心底座建设：统一的大模型调用封装

二、 智能体逻辑替换：Mock 退场，Prompt 登场

三、 工作流编排与接口层容错设计

遇到的问题与收获

总结与展望

---

一、 核心底座建设：统一的大模型调用封装

在构建多智能体系统时，最忌讳的是在每个 Agent 里重复编写 API 调用代码。我封装了 UnifiedLLMCaller类，并且实现模型供应商（如 Qwen、Kimi）的灵活切换。

基于 OpenAI SDK 的标准协议，通过配置文件 app/core/config.py 动态加载环境变量。实现业务逻辑与底层算力的彻底解耦。Agent 只需要调用 generate() 方法，无需关心底层是哪家模型的 API，这极大地增强了系统的可扩展性。

# app/core/llm.py 核心片段
class UnifiedLLMCaller:
    """统一的大模型调用基座，完全兼容 OpenAI SDK"""
    def __init__(self, provider: Optional[str] = None):
        self.provider = provider or settings.default_llm_provider
        
        # 动态切换模型提供商
        if self.provider.lower() == "qwen":
            self.api_key = settings.qwen_api_key    
            self.base_url = settings.qwen_base_url  
            self.model = "qwen-plus"
        elif self.provider.lower() == "kimi":
            self.api_key = settings.kimi_api_key
            self.base_url = settings.kimi_base_url
            self.model = "moonshot-v1-8k"
            
        self.client = AsyncOpenAI(api_key=self.api_key, base_url=self.base_url)

    async def generate(self, prompt: str, system_prompt: Optional[str] = None, temperature: float = 0.1) -> str:
        # 处理消息拼接与异步请求发起...

二、 智能体逻辑替换：Mock 退场，Prompt 登场

我们彻底移除了上周用于跑通流程的硬编码 Mock 数据，为五大核心 Agent 注入了基于 Prompt Engineering 的真实推理能力。

以 分诊智能体 (TriageAgent) 为例： 为了将家属口语化、模糊的病情描述精确提取为结构化数据，我使用了 JSON Schema 动态注入 技术。我通过 VitalSigns.model_json_schema() 获取了 Pydantic 模型的结构，并将其放入 System Prompt 中，强制 LLM 按规则输出。同时，借助LLM的推理能力 ，要求其不仅提取数据，还要推理出“紧急程度 (urgency_level)”和“建议科室 (suggested_department)”。

# app/agents/triage.py 核心片段
        # 1. 动态获取 Pydantic 模型的 JSON Schema
        schema = VitalSigns.model_json_schema()

        # 2. 赋予分诊护士真正的推理能力
        system_prompt = f"""你是一个三甲医院经验丰富的儿科分诊专家。
你的任务是从患儿家属口语化、模糊的病情描述中，精确提取结构化的体征特征。

【提取规则】：
1. 月龄 (age_months)：如果家属说“3岁半”，请精确计算并转换为 42
...
4. 紧急程度 (urgency_level): 根据体温和症状严重程度，主动推断并输出 low, medium, high 或 critical 之一。
5. 建议科室 (suggested_department): 根据症状给出建议的科室名称（如：儿科门诊、小儿急诊科、发热门诊等）。

除了严格返回以下 Schema 中要求的基础字段外，你必须在 JSON 根层级额外增加 "urgency_level" 和 "suggested_department" 两个键。
{json.dumps(schema, ensure_ascii=False, indent=2)}"""

此外，其他 Agent 也完成了相应升级：

诊断智能体 (DiagnosisAgent)：根据结构化体征生成包含“初步诊断”、“建议检查”和“处置建议”的专业 Markdown 报告。

评审智能体 (ReviewAgent)：扮演质控专家，严格审核诊断方案中的年龄禁忌和用药冲突。

定稿 (Plan) 与兜底 (Fallback) 智能体：分别负责最终报告的人文关怀润色，以及在博弈失败时的安全就医指导。

三、 工作流编排与接口层容错设计

在模型具备思考能力后，我们需要通过图计算与接口层确保其稳定运行。

LangGraph 工作流：在 app/graph/workflow.py 中，我定义了从 triage -> diagnosis -> review 的流转路径。通过编写条件路由 review_router，实现了“打回重写（回退 diagnosis）”和“熔断兜底（跳转 fallback）”的闭环博弈逻辑。

# app/graph/workflow.py
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from loguru import logger

# 导入共享状态字典与五大智能体节点（此处省略具体的节点包装函数定义）
from app.graph.state import PediaMindState
# ...

# ==========================================
# 1. 定义红蓝对抗的条件路由逻辑 (Conditional Edges)
# ==========================================
def review_router(state: PediaMindState) -> str:
    """根据评审结果决定下一步走向：通过、重试还是熔断"""
    feedback = state.get("review_feedback", {})
    is_passed = feedback.get("is_passed", False)
    retry_count = state.get("retry_count", 0)

    if is_passed:
        logger.info("🟢 路由判定：评审通过，流转至主任定稿 (plan)")
        return "plan"
    elif retry_count >= 3:
        logger.warning("🔴 路由判定：触发熔断！博弈次数达上限，流转至安全兜底 (fallback)")
        return "fallback"
    else:
        logger.info(f"🟠 路由判定：评审打回，返回诊断节点进行第 {retry_count + 1} 次重试 (diagnosis)")
        return "diagnosis"

# ==========================================
# 2. 构建并编译状态图 (构建多智能体博弈网络)
# ==========================================
def build_diagnosis_graph():
    """编译 PediaMind 多智能体工作流"""
    logger.info("构建 PediaMind 状态机...")
    workflow = StateGraph(PediaMindState)

    # 注册所有 Agent 节点
    workflow.add_node("triage", triage_node)
    workflow.add_node("diagnosis", diagnosis_node)
    workflow.add_node("review", review_node)
    workflow.add_node("plan", plan_node)
    workflow.add_node("fallback", fallback_node)

    # 定义基本的边（入口与单向流转）
    workflow.set_entry_point("triage")
    workflow.add_edge("triage", "diagnosis")
    workflow.add_edge("diagnosis", "review")

    # 定义带条件的边（状态机核心：基于路由逻辑进行分发，实现博弈与打回重写）
    workflow.add_conditional_edges(
        "review",
        review_router,
        {
            "plan": "plan",
            "diagnosis": "diagnosis",
            "fallback": "fallback"
        }
    )

    # 定义系统出口（终点）
    workflow.add_edge("plan", END)
    workflow.add_edge("fallback", END)

    # 使用 MemorySaver 持久化中间态，隔离上下文并支持多轮日志追溯
    checkpointer = MemorySaver()
    return workflow.compile(checkpointer=checkpointer)

API 层的防御性编程：大模型偶尔会产生幻觉，输出非预期的值。在 app/api/diagnosis.py 中，针对 LLM 返回的 urgency_level，我增加了健壮的 try-except 枚举解析逻辑。如果模型输出了非法的紧急度，系统会自动捕捉 ValueError 并将其回退至安全的 UrgencyLevel.medium，从而彻底避免了后端抛出 500 崩溃错误。

# app/api/diagnosis.py 核心容错片段
    # 【核心修改点】：安全获取紧急度，防止 LLM 输出非法枚举值导致后端 500 崩溃
    raw_urgency = triage.get("urgency_level", "medium")
    try:
        urgency_val = UrgencyLevel(raw_urgency.lower())
    except ValueError:
        logger.warning(f"[{request_id}] 模型输出了非法的紧急度: {raw_urgency}，回退至 medium")
        urgency_val = UrgencyLevel.medium

遇到的问题与收获

1.配置管理的“坑”： 在接入初期，系统报错 API Key 为空。经排查发现，pydantic-settings 会将 .env 中的大写变量映射为小写属性，而在代码中错误使用了大写读取。这次教训让我深刻理解了框架内部的映射规则。

2.Streamlit 的渲染机制： 前端界面在获取结果后偶现白屏消失。这是因为 st.rerun() 触发了不必要的整页刷新。通过优化 Session State 的状态更新逻辑并移除重复渲染，最终实现了流畅的“AI 诊断轨迹”展示。

3.JSON 格式的稳定性： 大模型偶尔会返回带有 ```json 的 Markdown 块导致解析失败。我在 Agent 基类中添加了健壮的字符串清洗函数，确保了数据流的稳定性。

总结与展望

本周完成了从“空架子”到“具备思考大脑”的蜕变。通过这一系列的开发，我深刻体会到架构设计中“高内聚、低耦合”的魅力，也验证了多智能体博弈在医疗问诊场景下的可行性，借助 LLM 自身强大的语言理解与生成能力，即便在没有引入RAG的情况下，系统依然展现出了令人满意的效果。这让我不禁更加期待，融入知识库后，系统将能释放出怎样的潜力。

上图为当前系统的输出，可见目前的prompt具备了初步的约束能力，也完成了初步的任务。

下周展望： 作为医疗辅助系统，仅靠大模型内置的通识是不够严谨的。下周，将挑战本项目强化 RAG系统的构建。引入 ChromaDB 向量数据库，对《儿科学》教材进行切片与向量化，让 PediaMind 的每一句医嘱，皆有权威文献可查！