CSDN 系列第六篇｜Python · SQLite · Memory · 上下文窗口管理

本文根据 2026-10-30 上传的 AgentProject 项目代码与阶段验收记录整理。重点是第六阶段已完成的 Memory 持久化；RAG 不纳入本篇已完成范围。代码片段取自当前实现并适当省略非核心部分。

一、前言：为什么已经有 ConversationHistory，还要做 Memory？

第三阶段我们已经能够保存多轮会话，但最初的 ConversationHistory 本质是 Python 字典：程序进程结束，内存中的历史就没了。重新启动服务后，Agent 无法继续之前的聊天。同时，用户数量多、对话时间长时，也不能无限制地把历史送给模型。

因此第六阶段的目标不是新增一个大模型功能，而是让历史记录具备可持久化、可隔离、可替换、可裁剪、可集成与可测试的能力。

二、这一阶段实际完成了什么

模块完成状态
ConversationHistory 基础会话记忆已完成
SQLite 持久化存储已完成
多用户历史隔离已完成
BaseConversationHistory 抽象接口已完成
ContextWindowManager已完成
动态 Token 预算与历史裁剪已完成
LLMPathHandler / Container 集成已完成
Tool Loop 回归测试用户确认通过
非法参数与边界测试用户确认通过

三、从内存字典开始：ConversationHistory

我先从最朴素的版本写起：基础版 ConversationHistory 使用 dict[str, list[LLMMessage]]，按 user_id 保存各自的消息列表。添加消息后，仅保留最近 max_messages 条；get_history 返回浅拷贝，避免调用方对列表本身的直接修改。

class ConversationHistory(BaseConversationHistory):
    def __init__(self, max_messages: int = 20):
        if max_messages <= 0:
            raise ValueError("max_messages 必须大于0")
        self.max_messages = max_messages
        self.messages: dict[str, list[LLMMessage]] = {}
def add_message(self, user_id, message):
    if user_id not in self.messages:
        self.messages[user_id] = []
    self.messages[user_id].append(message)
    if len(self.messages[user_id]) &gt; self.max_messages:
        self.messages[user_id] = self.messages[user_id][-self.max_messages:]
def get_history(self, user_id):
return self.messages.get(user_id, []).copy()</code>

这一版解决了基础的多用户隔离和长度限制，但数据只活在进程内存里，无法跨服务重启保留；因此我需要继续做持久化。

四、先设计抽象接口，再更换存储方案

如果 LLMPathHandler 直接依赖 ConversationHistory，那么换 SQLite 以后就得跟着修改调用逻辑。因此我先创建 BaseConversationHistory，把外部调用协议固定下来：只有 add_message 与 get_history 两个方法。

from abc import ABC, abstractmethod
class BaseConversationHistory(ABC):
@abstractmethod
def add_message(self, user_id: str, message: LLMMessage) -> None:
pass
@abstractmethod
def get_history(self, user_id: str) -&gt; list[LLMMessage]:
    pass</code>

我的设计思路是：调用方依赖“接口”，而不是依赖“字典实现”。这样上层 Handler 只认协议，底层换成 SQLite、Redis 还是别的存储，都不需要改动调用逻辑。

这里需要强调一点：当前接口只有读写历史的两个方法，create_table 是 SQLite 实现自身提供的初始化能力，并不属于抽象接口的一部分。这样接口保持最小，实现方可以按自己的存储特性补充初始化逻辑。

五、SQLiteConversationHistory：将历史写入数据库

接口定好之后，我开始实现真正的持久化版本。这里我选择 Python 标准库自带的 sqlite3，不需要另外启动数据库服务器，对个人项目来说足够轻量。数据库文件默认放在 data/agent_memory.db，初始化时建表；每条消息保存 user_id、role、content 三个字段。

CREATE TABLE IF NOT EXISTS messages (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id TEXT NOT NULL,
    role TEXT NOT NULL,
    content TEXT
);

建表语句里 id 自增主键很关键，它既作为消息的唯一标识，也天然承担了时间顺序的角色，后面查询最近记录和恢复时间顺序都要依赖它。

5.1 写入消息：参数化 SQL 与事务

写入消息时，我用参数化 SQL 把值作为独立参数传入，而不是把用户文本拼接进 SQL 字符串，这样能避免 SQL 注入问题。同时用 commit 提交变更，用 finally 保证连接一定被关闭。

conn = sqlite3.connect(self.db_path)
try:
    conn.execute(
        "INSERT INTO messages (user_id, role, content) values (?, ?, ?)",
        (user_id, message.role.value, message.content)
    )
    conn.commit()
finally:
    conn.close()

这里的问号是占位符，值作为独立参数传入，避免把用户文本拼接进 SQL。commit 负责提交变更，finally 保证连接关闭，即使中途抛异常也不会泄漏连接。

5.2 查询历史：先取最近 N 条，再恢复时间顺序

查询历史时有一个容易忽略的细节：我需要的是最近的 N 条，所以内部先按 id 倒序并 LIMIT；但提供给模型的消息必须按原先的时间顺序，因此外层再重新按 id 升序排列。

SELECT role, content
FROM (
    SELECT id, role, content
    FROM messages
    WHERE user_id = ?
    ORDER BY id DESC LIMIT ?
)
ORDER BY id ASC;

通过 WHERE user_id = ? 保证每个用户只取自己的记录，实现多用户历史隔离。内层倒序加 LIMIT 拿到最近 N 条，外层升序还原成正常聊天顺序，这样交给模型的历史才是符合对话逻辑的。

这里还要说明一点：当前 max_messages 只用于限制“读取返回的条数”，并不意味着数据库会自动删除更早的历史。也就是说，数据库里的记录会持续增长，后续仍需要归档或清理策略来处理长期数据。

六、ContextWindowManager：为什么不能把全部历史都给 LLM？

即使 SQLite 可以存下长期历史，模型的上下文窗口也是有限的。需要在 system 提示词、当前输入、输出预留空间和历史记录之间分配预算。

当前项目实现了一个轻量级估算器：以字符串长度粗估 Token，而不是调用模型专用 tokenizer，因此预算是近似值，不等于真实模型计费 Token 数。

def estimate_tokens(self, message: LLMMessage) -> int:
    content = message.content or ""
    char_count = len(content)
    return max(1, (char_count + 1) // 2)

6.1 动态预算计算

history_budget = (
    model_context_limit - system_tokens
    - user_tokens - reserved_tokens
)
return max(0, history_budget)

举例：模型上下文限制 8000，系统提示词估算 500，用户输入估算 300，回答预留 1500，那么历史预算约为 5700 Token。预算不够时返回 0；非法负值参数则主动抛出 ValueError。

6.2 历史裁剪：优先保留最近消息

for message in reversed(history_messages):
    token = self.estimate_tokens(message)
    if used_tokens + token > budget:
        break
    selected.append(message)
    used_tokens += token
return list(reversed(selected))

倒序遍历的意义是先选择最新消息，最后再反转为正常聊天顺序。这里使用 break：遇到第一条放不下的历史即停止，而不是跳过该条再挑更早的短消息。这是一种连续尾部裁剪策略。

七、把 Memory 接入 LLMPathHandler

Handler 首先通过 conversation_history.get_history(user_id) 加载消息；如果存在 ContextWindowManager，就构建不带历史的基础 Prompt，估算 system 与当前 user 所需 Token，然后计算历史预算并进行裁剪，最后交给 PromptBuilder 生成实际 messages。

history_messages = self.conversation_history.get_history(user_id)
base_messages = self.prompt_builder.build(
user_message=message, history_messages=[]
)
system_tokens = self.context_window_manager.estimate_tokens(base_messages[0])
user_tokens = self.context_window_manager.estimate_tokens(base_messages[-1])
history_budget = self.context_window_manager.calculate_history_budget(
model_context_limit=8000,
system_tokens=system_tokens,
user_tokens=user_tokens,
reserved_tokens=1500,
)
history_messages = self.context_window_manager.trim_history(
history_messages, max_tokens=history_budget
)
messages = self.prompt_builder.build(
user_message=message, history_messages=history_messages
)

上述片段是带有 ContextWindowManager 时的分支。当前代码保留了 manager=None 的兼容路径。

八、成功才保存：Memory 与 Tool Loop 的生命周期

当前 Handler 在模型调用成功、Tool Loop 正常结束后，才将本轮 USER 与最终 ASSISTANT 消息加入历史。遇到模型失败、工具不存在或迭代上限等提前返回的情形，不执行这一轮的历史追加。

这一点需要与“完整记录工具全过程”区分：当前持久化只存 USER/ASSISTANT 的 content，没有将中间 tool_calls、TOOL 结果作为数据库记录保存。若日后希望完整回放工具交互，还需要扩展数据结构。

九、ApplicationContainer：将 SQLite 作为实际实现注入

conversation_history = SQLiteConversationHistory(
    db_path="data/agent_memory.db",
    max_messages=20
)
conversation_history.create_table()
context_window_manager = ContextWindowManager(max_tokens=1000)
llm_path_handler = LLMPathHandler(
llm_client=llm_client,
prompt_builder=prompt_builder,
conversation_history=conversation_history,
tool_registry=tool_registry,
context_window_manager=context_window_manager,
)

好处是 Handler 无需知道底层存的是 Python dict 还是 SQLite。注意：目前实际执行预算来自 Handler 中 model_context_limit=8000、reserved_tokens=1500 的设置；ContextWindowManager(max_tokens=1000) 是其默认预算值，并不意味着所有请求实际固定按 1000 裁剪。

十、测试验证：从功能正确到边界可靠

这一阶段测试覆盖：不同 user_id 的数据隔离；同一个 Runtime 中复用 Handler 与 History 实例；调用成功后历史增长；模拟 LLM 失败后不增加历史；窗口预算计算与最近消息裁剪；不合法 max_messages/max_tokens 参数的 ValueError；以及集成后 Tool Loop 的回归行为。

用户已确认 Tool Loop 回归测试和非法参数、边界测试通过。由于当前文档不是一次新的 pytest 执行报告，这里不编造测试总数或具体本次运行结果；如发布文章，可在本地补充最新 pytest 终端截图。

十一、这阶段我真正理解的设计思路

第一，持久化和上下文不是一回事：SQLite 解决“能存多久”，ContextWindowManager 解决“每次给模型多少”。

第二，抽象接口并非为了增加文件数量：它让存储实现可替换，测试也更容易隔离。

第三，历史的读取顺序很重要：SQL 负责获取最近记录，但 Prompt 需要按时间顺序呈现。

第四，保存时机决定语义：当前实现只保留成功完成的用户输入与最终回复，失败过程不会自动写入会话历史。

十二、现阶段仍有哪些不足？

当前方案已具备可用的持久化记忆骨架，但仍是基础版：① Token 数是字符估算；② 历史截取可能切断 USER/ASSISTANT 配对；③ 数据库没有自动过期、清理与索引策略；④ Tool Loop 中间交互未持久化；⑤ 模型上下文限制与输出预留值暂时写在 Handler 内。以上作为后续可优化点，而非本阶段已完成成果。

十三、阶段总结与下一篇预告

第六阶段真正解决了“Agent 聊天记录只能活在进程内”的问题：先用抽象接口隔离具体存储，再用 SQLite 将消息落盘，通过 ContextWindowManager 限制每次送入 LLM 的历史长度，最终由 Container 接入原有 Runtime/LLMPathHandler 链路，并通过测试确认基本行为。

接下来可以在这一基础上继续探索 RAG 知识库与检索增强，但它属于另一条技术主线，不应混写成第六阶段 Memory 已完成的内容。

# python # 语言模型 # 后端 # 单元测试
Logo
北京朝阳AI社区

更多推荐

cover

OLink对比其他大模型聚合平台有什么不同？看平台差异的几个维度？

avatar 北京朝阳AI社区
cover

DMXAPI模型测评：客观解读MiniMax-M3.1-Flash迭代亮点

avatar 北京朝阳AI社区
cover

FDE 怎么选大模型？GPT-4 / Claude / DeepSeek 对比

avatar 北京朝阳AI社区