第4章状态与记忆¶

学习目标¶

理解 Session 会话管理与上下文生命周期
区分短期记忆与长期记忆的适用场景
掌握 SQLite 持久化存储的实现方法
了解外部记忆的检索层（向量检索）原理
学会记忆压缩与摘要以控制上下文成本

问题导入¶

某团队上线了一个"个人健康助手"Agent。用户第一天说"我有乳糖不耐受"，第二天问"早餐能喝牛奶吗"，Agent 却回答"可以喝"——因为它完全不记得昨天的对话。另一个痛点是：长对话超过 20 轮后，上下文撑爆 token 上限，调用报错或成本飙升。

这两个问题的根因是记忆管理缺失。Agent 不能"健忘"，也不能"记太多撑爆脑子"。本章讲解如何用 Session 管理短期对话、用数据库做长期持久、用摘要压缩控制上下文，让 Agent 既记得住又用得起。

核心概念¶

1. Session 会话管理¶

Session 是一次连续对话的容器，唯一标识一次会话（如 session_id）。它负责：

维护本次会话的消息历史（user/assistant/tool 消息序列）
在多轮调用间传递上下文
会话级配置（用户身份、偏好）

OpenAI Agents SDK 通过 context 参数注入会话级数据，开发者用 dataclass 或 dict 承载。

2. 短期记忆（上下文窗口）¶

短期记忆即 LLM 的上下文窗口——当前会话的全部消息。特点：

容量有限：受 model max_tokens 限制（如 128k）
随轮次增长：每轮都累加消息，越长越贵越慢
会话内有效：新会话不继承旧会话的上下文

管理短期记忆的核心是控制长度：截断、摘要、选择性保留。

3. 长期记忆¶

长期记忆跨会话存在，让 Agent "记住"用户的历史信息：

用户画像：偏好、约束（如乳糖不耐受）
事实记忆：用户曾提供的关键事实（如家庭住址）
事件记忆：历史交互要点

实现方式：把关键信息抽取出来，存到数据库（SQLite/PostgreSQL），新会话开始时检索相关记忆注入 prompt。

4. SQLite 持久存储¶

SQLite 是轻量级嵌入式数据库，适合 Agent 记忆的快速落地：

import sqlite3
conn = sqlite3.connect("agent_memory.db")
conn.execute("CREATE TABLE memories (user_id, key, value, ts)")

每条记忆按 user_id + key 索引，新会话时按用户拉取全部画像注入 prompt。

5. 外部持久记忆（检索层）¶

当记忆量大（成百上千条）时，不能全部塞进 prompt，需引入检索层：

把记忆向量化存入向量库（如 Chroma、FAISS）
新会话或每轮用当前问题检索 top-k 相关记忆
只把相关记忆注入上下文

这就是 RAG 的思想在记忆场景的应用，详见第 6 卷。

6. 记忆压缩与摘要¶

控制短期记忆长度的两种手段：

滚动摘要：每 N 轮把早期对话用 LLM 压缩成一段摘要，替换原始消息
选择性截断：保留最近 K 轮原文 + 更早的摘要，丢弃中间细节

摘要能大幅降低 token 消耗，同时保留关键信息（如用户偏好、已做决策）。

原理图¶

flowchart TB
    User[用户消息] --> Session[Session 会话管理]

    Session --> Short[短期记忆<br/>上下文窗口消息序列]
    Session --> Long[长期记忆读写]

    subgraph 长期记忆层
        Long --> Extract[关键信息抽取]
        Extract --> DB[(SQLite<br/>用户画像/事实)]
        Extract --> Vector[(向量库<br/>事件记忆)]
    end

    subgraph 上下文组装
        New[新会话启动] --> Load[加载用户画像]
        Load --> Retrieve[向量检索 top-k 相关记忆]
        Retrieve --> Compress[摘要压缩早期对话]
        Compress --> Assemble[组装最终上下文]
    end

    Assemble --> LLM[送入 LLM 决策]

    style Short fill:#bbdefb
    style DB fill:#c8e6c9
    style Vector fill:#c8e6c9
    style Compress fill:#fff9c4
    style LLM fill:#ffe0b2

图：Agent 记忆架构——短期记忆管理当前会话，长期记忆经 SQLite 与向量库持久化，通过加载与检索注入上下文，摘要压缩控制长度

示例代码¶

1. 带长期记忆的 Agent（SQLite 持久化）¶

"""
第4章示例：带长期记忆的 Agent
用 SQLite 持久化用户画像，跨会话记住关键信息
依赖：pip install openai-agents
"""
import asyncio
import sqlite3
import json
from dataclasses import dataclass
from agents import Agent, Runner, function_tool


# ========== 记忆存储层 ==========
def init_db():
    conn = sqlite3.connect("agent_memory.db")
    conn.execute(
        "CREATE TABLE IF NOT EXISTS user_memory "
        "(user_id TEXT, key TEXT, value TEXT, ts DATETIME DEFAULT CURRENT_TIMESTAMP)"
    )
    conn.commit()
    return conn


def save_memory(conn, user_id: str, key: str, value: str):
    conn.execute(
        "INSERT OR REPLACE INTO user_memory (user_id, key, value) VALUES (?,?,?)",
        (user_id, key, value),
    )
    conn.commit()


def load_memory(conn, user_id: str) -> dict:
    rows = conn.execute(
        "SELECT key, value FROM user_memory WHERE user_id=?", (user_id,)
    ).fetchall()
    return {k: v for k, v in rows}


# ========== 上下文：承载会话状态 ==========
@dataclass
class SessionContext:
    user_id: str
    conn: sqlite3.Connection


# ========== 记忆读写工具 ==========
@function_tool
def save_user_info(key: str, value: str) -> str:
    """保存用户的关键信息到长期记忆（如过敏原、地址、偏好）。

    Args:
        key: 信息键名，如 allergy、address
        value: 信息内容
    """
    # 通过 context 访问（这里用全局演示）
    save_memory(GLOBAL_CONN, CURRENT_USER, key, value)
    return f"已记住：{key}={value}"


@function_tool
def recall_user_info(key: str) -> str:
    """回忆用户的关键信息。

    Args:
        key: 信息键名
    """
    mem = load_memory(GLOBAL_CONN, CURRENT_USER)
    return mem.get(key, f"未记录 {key}")


GLOBAL_CONN = init_db()
CURRENT_USER = "u1001"


# ========== Agent ==========
agent = Agent(
    name="健康助手",
    instructions=(
        "你是个人健康助手。当用户提及个人特征（过敏、疾病、饮食限制）时，"
        "主动调用 save_user_info 保存。回答饮食建议前，先 recall_user_info "
        "查询用户是否有相关限制。"
    ),
    tools=[save_user_info, recall_user_info],
    model="gpt-4o-mini",
)


async def chat(user_input: str):
    result = await Runner.run(agent, user_input, max_turns=6)
    print("助手：", result.final_output)


async def main():
    # 会话 1：用户透露关键信息
    print("=== 会话 1 ===")
    await chat("我有乳糖不耐受，喝牛奶会拉肚子。")

    # 会话 2：跨会话验证记忆
    print("\n=== 会话 2（新会话）===")
    await chat("早餐能喝牛奶吗？")


if __name__ == "__main__":
    asyncio.run(main())

运行结果示例：

=== 会话 1 ===
助手： 已记住您的乳糖不耐受情况。建议您选择无乳糖牛奶或豆浆替代。
=== 会话 2（新会话）===
助手： 您有乳糖不耐受，不建议喝普通牛奶。推荐无乳糖牛奶或燕麦奶作为早餐替代。

2. 记忆压缩与摘要¶

"""
记忆压缩：对话超过阈值时，把早期消息摘要化
"""
from openai import OpenAI

client = OpenAI()


def summarize_messages(messages: list, keep_recent: int = 4) -> list:
    """把早期消息压缩为摘要，保留最近 keep_recent 轮原文"""
    if len(messages) <= keep_recent:
        return messages

    old = messages[:-keep_recent]
    recent = messages[-keep_recent:]

    summary_text = "\n".join(
        f"{m['role']}: {m['content']}" for m in old if m.get("content")
    )
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "把以下对话压缩为 200 字以内的摘要，保留关键事实和决策。",
            },
            {"role": "user", "content": summary_text},
        ],
    )
    summary = resp.choices[0].message.content
    return [{"role": "system", "content": f"[历史摘要]{summary}"}] + recent


# 使用：当 messages 超过 10 条时压缩
# messages = summarize_messages(messages, keep_recent=4)

企业案例¶

案例 1：个性化助手¶

某银行的"私人银行助手"为高净值客户服务。Agent 记忆体系分三层：①短期记忆保持当前咨询上下文；②SQLite 存用户画像（风险偏好、持仓、家庭情况），新会话自动加载；③向量库存历史咨询记录，客户问"上次说的那个产品"时语义检索。关键约束：敏感数据加密存储、合规审计、用户可查看与删除自己的记忆。上线后客户满意度提升 30%，重复信息确认减少 70%。

案例 2：长对话压缩¶

某法律咨询 Agent 单次对话常超 30 轮。团队用滚动摘要：每 6 轮压缩一次早期对话，保留最近 4 轮原文 + 一段摘要。token 消耗降低 60%，且关键事实（如当事人姓名、案由）不丢失。经验：摘要 prompt 要明确"保留所有人名、金额、时间、已做决策"。

常见问题¶

Q1：短期记忆和长期记忆的边界在哪？

A：短期记忆是当前会话的消息序列，会话结束即失效；长期记忆是跨会话持久化的数据，需主动写入和读取。短期记忆靠 context 传递，长期记忆靠数据库存储。

Q2：什么时候该用向量检索，什么时候用 SQLite 键值存储？

A：键值明确（用户画像：过敏原=牛奶）用 SQLite；语义模糊需相似度匹配（"上次聊的那个理财产品"）用向量检索。简单场景先 SQLite，量大再上向量库。

Q3：摘要会丢失信息吗？

A：会。压缩本质是有损的。降低损失的技巧：①摘要 prompt 明确保留项（人名、数字、决策）；②保留最近几轮原文不压缩；③关键事实同步写入长期记忆。

Q4：记忆怎么过期和清理？

A：①加时间戳，定期清理过期记忆；②用户主动删除（GDPR 合规要求）；③画像类记忆设 TTL，长期未更新则重新确认；④事件类记忆按重要性分级保留。

Q5：多用户场景下记忆如何隔离？

A：所有记忆以 user_id 为索引，读写都带 user_id 过滤。Session 的 context 注入当前 user_id，工具通过 context 访问对应数据，严禁跨用户查询。

面试题¶

简答题：短期记忆与长期记忆的区别及各自实现方式。

参考要点：短期=上下文窗口消息，会话内有效；长期=数据库持久化，跨会话。短期靠 context 传递，长期靠 SQLite/向量库。

设计题：为"学习辅导 Agent"设计记忆方案，需要记住学生的学习进度和薄弱知识点。

参考要点：短期记忆保持当前题目上下文；SQLite 存进度与薄弱点；向量库存历史错题便于检索。

实现题：如何实现记忆的滚动摘要？写出核心逻辑。

参考要点：超阈值时取早期消息调 LLM 摘要，替换为 system 摘要消息，保留最近 N 轮原文。

原理题：为什么不能把所有历史记忆都塞进上下文？

参考要点：token 上限、成本飙升、检索精度下降（信息淹没）、延迟增加。

对比题：SQLite 键值存储与向量库检索在记忆场景的优劣。

练习题¶

记忆持久化：扩展示例 1，增加"删除我的某条记忆"和"列出我的全部记忆"两个工具。
跨会话验证：让 Agent 在会话 1 记住"用户住在上海"，会话 2 问"明天出门要带伞吗"，验证是否结合记忆与天气工具回答。
摘要实验：构造 20 轮对话，分别用"全量保留"和"滚动摘要"两种策略，对比 token 数与关键信息保留率。

小结¶

记忆让 Agent 从"金鱼记忆"进化为"长期伙伴"。本章核心：Session 管理会话级短期记忆；SQLite 持久化跨会话的长期记忆；量大时用向量检索按需注入；滚动摘要控制上下文成本。记忆设计要在"记得住、用得起、不越权"间平衡。下一章进入多 Agent 协作——当任务超出单 Agent 能力，如何让多个 Agent 协同工作。

第4章 状态与记忆¶