山东大学-中医智能诊疗系统-项目实训（三）

一、任务概览：实现系统的记忆功能

在上一个开发阶段实现了会话Agent，能够进行有逻辑的引导式问诊。但此时的系统有个致命缺陷：每次会话都是独立的，系统不记得用户之前说过什么。

想象一下以下场景：

• 用户上一时刻问诊：头痛、怕冷、不出汗，AI建议多喝热水

• 用户下一时刻再次问诊：头痛加重了

• 系统：“您好，请问您哪里不舒服？”（完全不记得上一时刻的对话）

每次对话都是独立的，系统不能知道用户的历史输入，这显然不可接受。本阶段的任务就是解决这个问题——实现持久化的会话记忆。

具体工作包括：

1. 设计SessionMemory类：管理多会话的独立记忆空间

2. 实现诊疗记录管理：保存每次“开始诊疗”的完整报告

3. 患者维度记忆聚合：按患者ID聚合所有会话

---------------------------------------------------------------------------------------------------------------------------

具体设计过程：

在多轮中医问诊场景中，记忆不是简单存聊天记录，而是要支撑 “连续问诊、症状累积、患者绑定、复诊对比、诊疗溯源” 五大临床行为。但如果只做简单的对话存储，无法体现诊疗的连贯性与专业性。

因此我在设计时明确了几个核心判断：

1. 必须以 session_id 作为唯一数据主键所有数据（对话、症状、患者、诊疗）都挂在同一个会话 ID 下，保证一次就诊的数据完整性。这保证医疗数据必须可追溯、可归档、可独立管理的业务要求。

2. 记忆必须分层，不能全部塞给 LLM

   • 短时记忆：最近 10 轮对话（给模型看）
   • 长时记忆：提取后的症状列表（不随对话滚动）
   • 归档记忆：历次诊疗快照（用于复诊对比）这样设计既能控制上下文长度，又能保证关键信息不丢失，避免模型因信息过载导致辨证混乱。

3. 患者与会话必须解耦，但又能强关联一个患者可以多次就诊（多个 session），但每一次就诊必须独立记忆、独立症状、独立诊疗。这贴合真实门诊 “一次就诊一条记录” 的逻辑.

二、核心架构：内存缓存与数据库持久化的协同

2.1 为什么需要双重存储？

这次在设计记忆功能时，我面临一个经典的性能与持久性的权衡：

选择的解决方案：内存-数据库二级存储

用户发送消息 ↓ 写入数据库（持久化） ↓ 更新内存缓存（加速后续读取） ↓ 读取历史时优先从内存读，内存没有再从数据库加载

2.2 SessionMemory类的数据结构设计

class SessionMemory: def __init__(self): # 会话消息历史 self._messages: Dict[str, List[Dict]] = defaultdict(list) # 会话提取的症状 self._symptoms: Dict[str, Dict] = defaultdict( lambda: {"symptoms": [], "last_update": ""} ) # 会话元数据 self._metadata: Dict[str, Dict] = defaultdict(dict) # 会话诊疗记录 self._diagnoses: Dict[str, List[Dict]] = defaultdict(list) # 每个会话最多保留50条消息 self.max_messages_per_session = 50

设计决策：使用defaultdict

• 自动初始化不存在的key，避免KeyError

• 代码更简洁，不需要每次检查key是否存在

• 适合多会话并发的场景

一开始我考虑过用列表存储，但很快意识到：问诊必须支持多会话、多患者、多医生同时在线，必须用字典（key-value）结构。

key = session_id（唯一标识）

value = 该会话的完整档案最终确定如下存储结构：

_messages：存对话历史

_symptoms：存症状（长期记忆）

_diagnoses：存诊疗快照

2.3 关键设计：消息数量限制

def add_message(self, session_id: str, role: str, content: str) -> None: self._messages[session_id].append({ "role": role, "content": content, "timestamp": datetime.now().isoformat() }) # 限制消息数量，防止内存无限增长 if len(self._messages[session_id]) > self.max_messages_per_session: self._messages[session_id] = self._messages[session_id][-self.max_messages_per_session:]

开发过程中曾思考50条的限制是否合理？

所以经过测试，这已经足够记住当前症状、当前主诉、当前问题，带多了反而导致了大模型混乱、忘记重点，导致辩证出现参差。其余理由如下：

• 一次完整问诊大约10-15轮对话（20-30条消息）

• 50条足够覆盖3-4次完整问诊

• 超出部分自动丢弃，但数据库中有完整记录

三、诊疗记录管理

3.1 数据结构设计

def add_diagnosis(self, session_id: str, diagnosis_id: str, report: Dict) -> None: self._diagnoses[session_id].append({ "diagnosis_id": diagnosis_id, "timestamp": datetime.now().isoformat(), "report": report, "symptoms_snapshot": self.get_symptoms(session_id) })

关键设计：保存symptoms_snapshot症状快照

为什么需要症状快照？

• 症状会随着问诊不断更新

• 诊疗报告应该反映当时的症状，而不是后来的

• 多次诊疗对比功能需要知道每次诊疗时的症状状态

3.2 获取诊疗记录

def get_diagnoses(self, session_id: str) -> List[Dict]: """获取所有诊疗记录，按时间排序""" return sorted( self._diagnoses.get(session_id, []), key=lambda x: x["timestamp"] ) def get_diagnosis(self, session_id: str, diagnosis_id: str) -> Optional[Dict]: """获取指定诊疗记录""" for d in self._diagnoses.get(session_id, []): if d["diagnosis_id"] == diagnosis_id: return d return None

四、患者维度的记忆聚合

4.1 核心功能：按患者聚合会话

系统核心的功能之一是能够还原某个患者的完整“健康画像”：

def get_all_sessions(self, session_type=None, user_id=None, patient_id=None) -> List[Dict]: sessions = [] for sid, meta in self._metadata.items(): if session_type and meta.get("type") != session_type: continue if user_id and meta.get("user_id") != user_id: continue if patient_id and meta.get("patient_id") != patient_id: continue sessions.append({ "session_id": sid, "name": meta.get("name", ""), "type": meta.get("type", ""), "created_at": meta.get("created_at", ""), "updated_at": meta.get("updated_at", ""), "message_count": len(self._messages.get(sid, [])), "diagnosis_count": len(self._diagnoses.get(sid, [])) }) sessions.sort(key=lambda x: x.get("updated_at", ""), reverse=True) return sessions

4.2遇到问题：历史会话显示不全

现象：数据库中有多条会话记录，前端却只显示部分会话。

排查过程：

1. 检查前端API调用 → 正常

2. 检查API返回数据 → 只返回了部分

3. 检查get_patient_sessions方法 → 返回的是self.memory.get_all_sessions()

原因：
self.memory.get_all_sessions()只返回内存中的会话。内存中的_metadata只在用户与某个会话交互后才会同步，导致大量历史会话“消失”。

解决方案：
让查询操作直接返回数据库的查询结果，而不是依赖内存：

async def get_patient_sessions(self, patient_id: str, db: AsyncSession) -> List[Dict]: # 直接从数据库查询 db_sessions = await self.db_manager.get_sessions_by_patient(db, patient_id) sessions = [] for db_session in db_sessions: sessions.append({ "session_id": db_session.id, "name": db_session.name, "type": db_session.type, "created_at": db_session.created_at.isoformat(), "updated_at": db_session.updated_at.isoformat() }) for db_session in db_sessions: if not self.memory.get_session_metadata(db_session.id): self.memory.create_session(...) return sessions

吸取经验：

• 内存只应用于加速当前活跃会话的读写

• 归档性质的读取操作（如历史列表）应该直接从数据库读取

• 不能假设内存中的数据和数据库是一致的

五、记忆与对话的关联

5.1 核心问题：LLM的“遗忘”困境

在多轮对话中，LLM面临一个根本性挑战：上下文窗口有限 + 注意力机制稀释。

当对话轮次增加时，用户早期提到的症状信息（如第1轮说“头痛”，第2轮说“怕冷”）会随着新消息的加入而被“挤”出注意力范围。结果就是：第5轮时，LLM已经忘记了用户第1轮说的症状——这在中医问诊中不可接受

解决方案：不让LLM自己去“回忆”，而是在每次调用时主动喂给它关键信息。

5.2 构建包含记忆的提示词

format_history_for_prompt方法的本质是将非结构化的对话历史转换为LLM可理解的结构化上下文：

def format_history_for_prompt(self, session_id: str, max_turns: int = 10) -> str: """格式化会话历史用于提示词""" messages = self.get_last_n_messages(session_id, max_turns * 2) if not messages: return "" formatted = [] for msg in messages: if msg["role"] == "user": formatted.append(f"用户: {msg['content']}") elif msg["role"] == "assistant": formatted.append(f"助手: {msg['content']}") return "【历史对话】\n" + "\n".join(formatted) + "\n\n【当前对话】\n"

此处的思考：

max_turns参数的设定是一个工程权衡。设太小会丢失早期信息，设太大会浪费token。10轮对话（20条消息）约2000-3000 tokens，在调用大模型的上下文窗口内，同时覆盖了大部分中医问诊的完整流程。

5.3 获取Agent需要的完整上下文

get_context_for_agent方法的职责是聚合所有与诊疗相关的信息，为后续多个诊疗Agent提供统一的数据接口：

def get_context_for_agent(self, session_id: str) -> Dict[str, Any]: """获取Agent需要的完整上下文""" return { "symptoms": self.get_symptoms(session_id), "conversation_summary": self.get_conversation_summary(session_id), "diagnosis_history": [ {"id": d["diagnosis_id"], "timestamp": d["timestamp"]} for d in self._diagnoses.get(session_id, []) ] }

为什么需要这个方法？

在后续阶段，多个诊疗Agent都需要从会话中获取信息。如果每个Agent都自己去查内存、读数据库，会造成：

1. 代码重复：每个Agent都要写相同的查询逻辑

2. 数据不一致：不同Agent可能读到不同时间点的数据

3. 难以维护：修改数据结构需要改多个地方

这个方法的意义：创建一个统一的上下文契约——所有Agent都通过这个接口获取数据，保证一致性和可维护性。

返回数据结构的设计考量：

注意：diagnosis_history只返回id和timestamp，不返回完整报告。这是因为：

• 历史诊疗记录可能很多，完整返回会超出token限制

• 如果Agent需要某个历史报告的详细内容，可以通过diagnosis_id按需加载

六、总结与展望

6.1 本阶段完成的交付物

6.2 解决的问题

6.3 技术债务

1. 会话摘要未实现：长对话时token消耗大，需要实现历史摘要压缩

2. 记忆同步机制不完善：内存和数据库的同步策略还可以优化

3. 缺少记忆清理机制：长时间不活跃的会话应该从内存中淘汰