记忆系统：三层记忆架构与上下文管理
为什么需要记忆？
一个没有记忆的旅行助手是残缺的。用户说"帮我规划杭州3天"后，再问"酒店换便宜点的"，系统应该知道上下文是杭州、知道之前的预算，而不是重新问一遍。
这里我设计了三层记忆体系，在单次请求、跨请求、长期三个时间尺度上管理信息。
三层记忆架构
┌─────────────────────────────────────────────────┐
│ 用户输入 │
│ ↓ │
│ ┌───────────────────────────────────────────┐ │
│ │ 意图识别 (注入上下文 + 用户画像) │ │
│ └───────────────────────────────────────────┘ │
│ ↓ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ L1 瞬时 │ │ L2 短期 │ │ L3 画像 │ │
│ │ Checkpoint│ │ SQLite │ │ SQLite │ │
│ │ 内存 │ │ 磁盘 │ │ 磁盘 │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────┘
为什么没有RAG
起码我认为旅游规划师不需要特别多的长期记忆的（当然也只是目前认为）
L1: 瞬时记忆（LangGraph State）
存储位置: 内存中的 LangGraph State
生命周期: 单次请求（supervisor → agents → aggregator → 结束）
内容: AgentState 的全部字段——user_input、destination、days、budget、各 Agent 的中间结果
实现方式: LangGraph 的 StateGraph 天然管理状态流转。每个节点（supervisor、route、hotel 等）读取 state、返回增量更新，LangGraph 自动合并。
每个agent有自己的state字段，防止了字段读写覆盖冲突
class AgentState(TypedDict):
messages: Annotated[list, add_messages] # 对话消息列表（由checkpoint自动管理）
running_summary: dict # 摘要状态（由 langmem.summarize_messages 自动维护，序列化为dict）
user_input: str # 用户发的原始消息
destination: str # 目的地（如"成都"）
days: int # 旅行天数
budget: str # 预算（如"5000元"）
preferences: str # 用户偏好（如"喜欢自然风光"）
departure: str # 出发城市（如"深圳"）
transport_mode: str # 交通方式（如"高铁"、"自驾"）
agents_to_call: list[str] # 要调用的agent列表（如 ["route","hotel","food"]）
route_result: str # 路线agent的结果（JSON字符串）
hotel_result: str # 酒店agent的结果（JSON字符串）
food_result: str # 美食agent的结果（JSON字符串）
info_result: str # 文化agent的结果
final_result: str # 最终汇总结果
request_id: str # 请求ID，用于SSE推送
is_casual: bool # 是否是闲聊（如"你好"）
need_clarify: bool # 是否需要追问用户（如目的地不明确）
clarify_msg: str # 追问的内容
user_id: str # 用户ID
session_id: str # 会话ID（同一个用户的多次对话属于同一会话）
并行写入: 当多个 Agent 通过 Send() 并行执行时，每个 Agent 独立写入自己的字段（route_result、hotel_result 等），不会冲突。LangGraph 在 fan-in 时自动合并各分支的 state 更新。
那么fan-in最最核心的问题，就是LangGraph 的执行引擎使用计数器屏障（Counter Barrier）：
supervisor_node 执行完毕 │ ├─ 返回 [Send("route", s), Send("hotel", s), Send("food", s), Send("info", s)] │ │ LangGraph 执行引擎: │ ┌─────────────────────────────────────────────────────┐ │ │ 1. 解析出 4 个 Send │ │ │ 2. 将 4 个分支入队，设置 barrier_count = 4 │ │ │ 3. 并行调度执行（各自独立，互不阻塞） │ │ │ ├─ route 分支执行 → 写入 state_update_1 │ │ │ ├─ hotel 分支执行 → 写入 state_update_2 │ │ │ ├─ food 分支执行 → 写入 state_update_3 │ │ │ └─ info 分支执行 → 写入 state_update_4 │ │ │ 4. 每完成一个分支，barrier_count -= 1 │ │ │ 5. barrier_count == 0 时： │ │ │ a. 合并所有 state_update（应用各自的 Reducer） │ │ │ b. 将合并后的 state 发送给 aggregator 节点 │ │ │ c. 执行 aggregator_node(state_merged) │ │ └─────────────────────────────────────────────────────┘ │ ▼ aggregator_node 执行（收到的 state 已包含 route_result, hotel_result 等）
L2: 短期记忆（SQLite 对话历史）
存储位置: data/memory.db SQLite 数据库
生命周期: 跨请求持久化，服务重启后仍在
表结构:
CREATE TABLE conversations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL DEFAULT 'guest',
session_id TEXT NOT NULL,
role TEXT NOT NULL, -- 'user' 或 'assistant'
content TEXT NOT NULL,
agent TEXT, -- 哪个Agent产生的
created_at REAL NOT NULL
);
CREATE TABLE summaries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
session_id TEXT NOT NULL,
summary TEXT NOT NULL, -- 压缩后的摘要
msg_count INTEGER NOT NULL, -- 摘要覆盖的消息数
created_at REAL NOT NULL
);
写入时机:
- 用户发送消息时 →
save_message(uid, sid, "user", text) - Agent 生成回复后 →
save_message(uid, sid, "assistant", result, "主管")
读取时机: Supervisor 执行前，通过 build_context() 组装上下文注入 prompt。
L3: 用户画像（长期记忆）
存储位置: 同一个 SQLite 数据库
生命周期: 长期持久化，随使用不断丰富
CREATE TABLE user_profiles (
user_id TEXT PRIMARY KEY,
preferred_budget TEXT, -- 预算偏好
preferred_style TEXT, -- 旅行风格
visited_cities TEXT DEFAULT '[]', -- JSON数组，常去城市
last_destination TEXT, -- 最近一次目的地
trip_count INTEGER DEFAULT 0,
updated_at REAL
);
更新时机: 每次规划完成后，从 state 中提取 destination/budget 更新画像。
使用场景: 用户说"再帮我规划一次"时，系统从画像中读取 last_destination 和 preferred_budget。
TODO（这里有一个问题，什么时候覆盖什么时候是追加？？？）
上下文窗口控制
这是记忆系统最核心的设计——如何在有限的 token 窗口内塞入足够的上下文。
策略
消息数 < 6 → 直接拼接最近N条消息
消息数 6-12 → 摘要 + 最近3条原始消息
消息数 > 12 → 触发压缩：生成新摘要，删除已摘要的旧消息
摘要触发算法
SUMMARY_THRESHOLD = 6 # 每6条新消息触发一次
KEEP_RECENT = 3 # 保留最近3条原始消息
def _maybe_summarize(user_id, session_id, llm):
# 1. 计算自上次摘要以来的新消息数
last_count = get_last_summary_msg_count()
current_count = get_message_count()
new_since_last = current_count - last_count
# 2. 未达阈值则跳过
if new_since_last < SUMMARY_THRESHOLD:
return
# 3. 获取需要摘要的消息（排除最近3条）
to_summarize = messages[:-KEEP_RECENT]
# 4. 调用LLM生成摘要
summary = llm.invoke(f"将以下对话压缩为200字摘要...")
# 5. 保存摘要
save_summary(summary, current_count)
# 6. 删除已被摘要覆盖的旧消息
delete_old_messages(keep_recent=KEEP_RECENT)
为什么是6条？
这是一个经验值：
- 太小（如3条）：频繁触发LLM摘要调用，增加延迟和成本
- 太大（如20条）：上下文窗口膨胀，Supervisor 意图识别变慢
- 6条：大约2-3轮对话，刚好覆盖一个完整的"规划→追问→修改"流程
上下文拼接格式
def build_context(user_id, session_id, llm) -> str:
parts = []
# 1. 用户画像
profile = get_user_profile(user_id)
if profile["visited_cities"]:
parts.append(f"[用户画像] 常去城市: {', '.join(profile['visited_cities'][-5:])}")
if profile["preferred_budget"]:
parts.append(f"[用户画像] 预算偏好: {profile['preferred_budget']}")
# 2. 最新摘要
summary = get_latest_summary()
if summary:
parts.append(f"[对话摘要] {summary}")
# 3. 最近3条原始消息
recent = get_recent_messages(limit=3)
for msg in recent:
parts.append(f" {role}: {content[:200]}")
return "\n".join(parts)
意图识别增强
没有记忆时，Supervisor 的意图识别是"无状态"的——只看当前输入。有了记忆后，它变成"有状态"的：
无记忆 vs 有记忆
| 用户输入 | 无记忆 | 有记忆 |
|---|---|---|
| "酒店换便宜点的" | 不知道目的地，随机推荐 | 从上下文知道是杭州，从画像知道预算 |
| "再帮我规划一次" | 不知道规划什么 | 从画像读取上次目的地 |
| "有什么好吃的" | 不知道去哪 | 从摘要知道正在规划成都 |
| "3天够吗" | 不知道在说什么 | 从上下文知道在讨论杭州行程 |
Prompt 注入
prompt = SUPERVISOR_PROMPT
if context:
prompt += f"\n\n--- 上下文记忆 ---\n{context}\n---\n"
Supervisor 收到的 prompt 结构：
[系统指令] 你是TravelAgent-5...
--- 上下文记忆 ---
[用户画像] 常去城市: 杭州, 成都
[用户画像] 预算偏好: 5000元
[对话摘要] 用户要求规划杭州3天旅行，预算5000元。已推荐西湖...
[最近对话]
用户: 酒店换便宜点的
---
State 读写流程图
用户消息到达
│
├─ save_message(user) ←── 写入SQLite
│
├─ build_context() ←── 读取SQLite（画像+摘要+最近消息）
│
├─ Supervisor
│ ├─ 读: user_input + context
│ 写: destination, days, agents_to_call
│
├─ Route Agent (Send)
│ ├─ 读: destination, days
│ 写: route_result
│
├─ Hotel Agent (Send, 并行)
│ ├─ 读: destination, budget
│ 写: hotel_result
│
├─ Food Agent (Send, 并行)
│ ├─ 读: destination
│ 写: food_result
│
├─ Info Agent (Send, 并行)
│ ├─ 读: destination
│ 写: info_result
│
├─ Aggregator
│ ├─ 读: route/hotel/food/info_result
│ 写: final_result
│
├─ save_message(assistant) ←── 写入SQLite
├─ update_user_profile() ←── 写入SQLite
└─ _maybe_summarize() ←── 可能写入摘要+删除旧消息
性能考量
| 操作 | 耗时 | 是否阻塞 |
|---|---|---|
| SQLite 读取（最近消息） | <1ms | 否 |
| SQLite 写入（保存消息） | <1ms | 否 |
| LLM 摘要调用 | 1-3s | 是（后台线程） |
| 用户画像更新 | <1ms | 否 |
摘要调用是最重的操作，但它只在每6条消息时触发一次，且在后台线程中执行，不阻塞 SSE 流式输出。
导航
← 上一篇：部署：2核2G服务器上的部署 → 下一篇：从单 Agent 到多 Agent：一次真实的架构改造