企业官网建设流程全解析

1. 从内存列表到磁盘文件：Agent 记忆为什么必须落库

做 Agent 开发到一定阶段，几乎所有人都会撞上同一堵墙：对话一多，上下文就爆。最开始大家用的都是最朴素的办法——一个 Python 列表，把每轮对话 append 进去，调用模型时整个塞进 prompt。这个方案在 demo 阶段无比丝滑，但只要跑上二三十轮，你就会发现两个致命问题：一是 token 消耗像坐火箭，二是程序一重启，Agent 立刻失忆，昨天聊过什么完全不记得。

我在 Day 10 左右做过一个模拟项目，让 Agent 帮用户记录每天的待办事项。当时就是用内存列表存对话历史，测试的时候一切正常，结果第二天重新跑脚本，Agent 一脸茫然地问我"你之前提到过什么任务吗"。那一刻我才真正意识到，内存里的记忆是假的记忆，只有落到磁盘上的数据才算数。这就像你脑子里记着一堆事，但一睡觉全忘了，那不叫记忆，那叫缓存。

所以 Day 16 的核心任务很明确：给 Agent 接一个真正的持久化存储。而在一众数据库方案里，我最终选了 SQLite。原因不复杂，下面这张表能说明大部分问题：

对个人开发者和本地 Agent 来说，SQLite 几乎是"甜点区"：它是一个文件就是一个完整数据库，不需要你启动任何后台服务，Python 标准库直接import sqlite3就能用，零依赖。你甚至可以把整个数据库文件拷到 U 盘里带走，换台电脑照样跑。这种"无服务器"的特性，对还在单机阶段折腾 Agent 的人来说，友好到没朋友。

但我要先泼一盆冷水：SQLite 不是万能药。它的设计定位是嵌入式数据库，适合读多写少、单进程访问的场景。如果你的 Agent 要同时服务几百个用户高频写入，那 SQLite 会成为瓶颈，这时候该上 PostgreSQL 还是得上。不过对于 90% 的个人项目和原型验证，SQLite 完全够用，而且能让你把精力集中在 Agent 逻辑本身，而不是数据库运维上。

这一篇我会把从零接入 SQLite 的完整链路讲透：怎么设计记忆表结构、怎么封装增删改查、怎么处理 Agent 特有的"上下文窗口"问题、以及我在实操中踩过的几个坑。如果你也是从前后端转过来做 AI 的，对数据库只有模糊印象，那这篇正好帮你把这块地基打牢。

2. 拆解 Agent 记忆的真实结构：不是一张表能装下的

很多人第一次给 Agent 设计数据库，脑子里想的就是一张messages表，字段大概是 id、role、content、timestamp，然后往里塞就完事了。我一开始也是这么干的，结果做到第三天就发现不对劲——Agent 的记忆根本不是单一维度的对话流水，它至少包含三种性质完全不同的数据，混在一张表里会让后续查询和扩展变得极其痛苦。

2.1 对话消息、会话元数据、长期事实，三者要分开存

先说我踩的第一个坑。当时我把所有东西都塞进messages表，包括用户的偏好设置（比如"我喜欢简洁的回答"）、每次会话的标题、以及具体的对话内容。结果查询的时候，我要写一堆WHERE type = 'xxx'来区分，代码里到处是魔法字符串，维护起来想砸键盘。

正确的做法是按数据的生命周期和访问模式拆表。我最终设计了三张核心表：

• sessions 表：存会话级别的元数据，比如会话 ID、创建时间、最后活跃时间、会话标题。一个会话就是一段连续的对话。
• messages 表：存具体的对话消息，通过 session_id 关联到会话。这是数据量最大的一张表。
• facts 表：存从对话中提炼出来的长期事实，比如用户的名字、偏好、重要结论。这些数据跨会话存在，不随会话结束而消失。

为什么要这么拆？因为它们的读写频率和查询方式完全不同。messages 表是高频写入、按会话顺序读取；facts 表是低频写入、但每次对话开始都要全量或条件读取；sessions 表则是偶尔更新一下活跃时间。混在一起会导致索引设计顾此失彼。

2.2 字段类型的选择：为什么时间戳我坚持用整数

设计表结构时有个细节值得单独说：时间字段到底用什么类型。SQLite 其实支持 TEXT、REAL、INTEGER 存时间，很多人图省事直接用字符串存"2024-01-01 12:00:00"。我一开始也这么干，直到要做"查询最近 7 天的对话"这种操作时，字符串比较的坑就来了——时区、格式不统一、排序出错，全是问题。

我现在的做法是统一用 Unix 时间戳（整数）存储，需要展示时再在应用层转成可读格式。这样做的好处有三个：一是排序和范围查询天然正确，整数比较没有歧义；二是跨时区无压力，时间戳是全球统一的；三是存储空间小，一个整数比一长串字符串省地方。

import sqlite3 import time def init_db(db_path="agent_memory.db"): conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS sessions ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, created_at INTEGER NOT NULL, last_active_at INTEGER NOT NULL ) """) cursor.execute(""" CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id INTEGER NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at INTEGER NOT NULL, FOREIGN KEY (session_id) REFERENCES sessions(id) ) """) cursor.execute(""" CREATE TABLE IF NOT EXISTS facts ( id INTEGER PRIMARY KEY AUTOINCREMENT, key TEXT NOT NULL UNIQUE, value TEXT NOT NULL, updated_at INTEGER NOT NULL ) """) conn.commit() return conn

这段代码里有个细节：facts表的key字段加了UNIQUE约束。因为长期事实本质上是键值对，同一个 key 应该只有最新值，用INSERT OR REPLACE就能实现"更新或插入"的语义，避免重复记录。

2.3 索引不是越多越好：我只建了两个

新手容易犯的另一个错是给每个字段都加索引，觉得"查得快"。实际上索引会拖慢写入速度，还会占空间。对 Agent 记忆库来说，真正高频的查询只有两类：按会话查消息和按时间排序。所以我只在messages表的session_id上建了索引，在created_at上建了索引，其他一概不加。

cursor.execute("CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_messages_created ON messages(created_at)")

提示：索引的取舍原则是"为查询建索引，不为字段建索引"。先想清楚你的 Agent 会怎么查数据，再决定索引建在哪。盲目加索引，写入性能会悄悄下降，而你未必察觉得到。

3. 封装一个能用的记忆层：别让 SQL 散落在业务代码里

表建好了，接下来是把数据库操作封装起来。我见过不少人的代码，SQL 语句直接写在 Agent 主逻辑里，东一句cursor.execute西一句conn.commit，改个字段名要全局搜索替换，痛苦不堪。正确的做法是把记忆层封装成一个类，对外只暴露语义清晰的方法，业务代码完全不碰 SQL。

3.1 MemoryStore 类的接口设计思路

我设计的MemoryStore类，对外提供的方法都是"人话"级别的，比如add_message、get_recent_messages、set_fact、get_fact。这样 Agent 主逻辑读起来就像自然语言，可维护性直接拉满。

class MemoryStore: def __init__(self, db_path="agent_memory.db"): self.conn = sqlite3.connect(db_path, check_same_thread=False) self.conn.row_factory = sqlite3.Row self._init_tables() def _init_tables(self): # 建表逻辑同上，略 pass def create_session(self, title="新会话"): now = int(time.time()) cursor = self.conn.cursor() cursor.execute( "INSERT INTO sessions (title, created_at, last_active_at) VALUES (?, ?, ?)", (title, now, now) ) self.conn.commit() return cursor.lastrowid def add_message(self, session_id, role, content): now = int(time.time()) cursor = self.conn.cursor() cursor.execute( "INSERT INTO messages (session_id, role, content, created_at) VALUES (?, ?, ?, ?)", (session_id, role, content, now) ) cursor.execute( "UPDATE sessions SET last_active_at = ? WHERE id = ?", (now, session_id) ) self.conn.commit() return cursor.lastrowid def get_recent_messages(self, session_id, limit=20): cursor = self.conn.cursor() cursor.execute( """SELECT role, content FROM messages WHERE session_id = ? ORDER BY created_at DESC LIMIT ?""", (session_id, limit) ) rows = cursor.fetchall() return [{"role": r["role"], "content": r["content"]} for r in reversed(rows)]

这里有个关键细节：get_recent_messages里我先用DESC取最新的 N 条，然后在 Python 里reversed反转回时间正序。为什么要这么绕？因为 SQL 的LIMIT配合DESC才能高效拿到"最近 N 条"，如果直接用ASC加LIMIT，拿到的是最老的 N 条，完全不是你要的。这个坑我踩过，当时调试了半天才发现取出来的是最早的对话。

3.2 参数化查询：防注入不是可选项

上面代码里所有 SQL 都用了?占位符，这是参数化查询，绝对不能省。有些人图快用 f-string 拼 SQL，比如f"SELECT * FROM messages WHERE content = '{user_input}'"，这在 Agent 场景下尤其危险，因为用户输入的内容会直接进 SQL。如果用户输入里带个引号，轻则报错，重则数据被删。

参数化查询的原理是：SQL 语句和数据分开传给数据库引擎，引擎先编译语句结构，再把数据填进去，数据永远不会被当成 SQL 代码执行。这是防注入的根本手段，没有之一。

3.3 连接管理：check_same_thread 到底该不该关

sqlite3.connect有个参数check_same_thread，默认是True，意思是这个连接只能在创建它的线程里用。但 Agent 经常涉及多线程（比如一边接收用户输入一边后台处理），这时候就会报SQLite objects created in a thread can only be used in that same thread。

我一开始的做法是直接设check_same_thread=False，问题确实没了。但后来了解到，这其实是把线程安全的锅甩给了自己——关掉检查后，多个线程同时写同一个连接可能出问题。更稳妥的方案是每个线程用独立连接，或者加锁串行化写入。对个人项目来说，check_same_thread=False配合简单的写锁通常够用，但心里要清楚这个取舍。

import threading class MemoryStore: def __init__(self, db_path="agent_memory.db"): self.conn = sqlite3.connect(db_path, check_same_thread=False) self.conn.row_factory = sqlite3.Row self._lock = threading.Lock() self._init_tables() def add_message(self, session_id, role, content): with self._lock: # 写入逻辑 pass

加个threading.Lock把写操作串起来，成本极低，但能避免大部分并发写入的诡异问题。

4. 上下文窗口的取舍：从数据库里捞多少历史才合适

记忆落库只是第一步，真正考验功力的是每次调用模型时，从数据库里捞多少历史对话塞进 prompt。捞少了，Agent 记不住前文；捞多了，token 爆炸还拖慢响应。这个平衡点怎么找，是 Agent 记忆管理的核心问题。

4.1 固定条数截断的局限与改进

最直觉的方案是"取最近 20 条消息"。我一开始就这么干，但很快发现两个问题：一是如果某条消息特别长（比如用户粘贴了一大段代码），20 条可能就超 token 了；二是如果对话很简短，20 条又太少，Agent 记不住更早的关键信息。

改进思路是按 token 预算动态截断，而不是按条数。具体做法是从最新消息往前累加，估算每条消息的 token 数（粗略可以用字符数除以 2 到 3 来估），累加到接近预算上限就停。这样无论消息长短，都能把上下文控制在合理范围。

def get_context_within_budget(self, session_id, max_chars=4000): cursor = self.conn.cursor() cursor.execute( """SELECT role, content FROM messages WHERE session_id = ? ORDER BY created_at DESC""", (session_id,) ) rows = cursor.fetchall() selected = [] total = 0 for r in rows: msg_len = len(r["content"]) if total + msg_len > max_chars and selected: break selected.append({"role": r["role"], "content": r["content"]}) total += msg_len return list(reversed(selected))

这个函数从最新往回捞，直到超出字符预算为止。注意and selected这个条件，它保证至少会返回一条消息，避免单条超长消息导致返回空列表。

4.2 长期事实的注入时机：每次对话开头都带上

对话历史解决的是"短期记忆"，而facts表解决的是"长期记忆"。我的做法是：每次新会话开始时，把 facts 表里的所有事实拼成一段系统提示，注入到对话最前面。这样 Agent 一开口就"记得"用户的偏好和重要信息，不用等用户重复。

def build_system_prompt(self): cursor = self.conn.cursor() cursor.execute("SELECT key, value FROM facts ORDER BY updated_at DESC") rows = cursor.fetchall() if not rows: return "你是一个有帮助的助手。" facts_text = "\n".join([f"- {r['key']}: {r['value']}" for r in rows]) return f"你是一个有帮助的助手。以下是关于用户的已知信息：\n{facts_text}"

这里有个经验：facts 不要存太多，控制在十几条以内。存太多会挤占上下文，而且很多"事实"其实没那么重要。我一般只存用户明确表达的偏好、身份信息、以及跨会话必须记住的结论。

4.3 什么时候该把对话"提炼"成事实

长期事实从哪来？两个途径：一是用户明确说"记住我喜欢 X"，直接写入；二是让模型在对话结束时自动提炼。第二种更智能，但也更容易出错，我的做法是只在会话空闲一段时间后触发提炼，并且提炼结果先给用户确认再入库，避免模型瞎记。

def extract_facts_prompt(self, session_id): messages = self.get_recent_messages(session_id, limit=50) dialogue = "\n".join([f"{m['role']}: {m['content']}" for m in messages]) return f"""请从以下对话中提炼出值得长期记住的用户信息， 以 key: value 格式输出，每行一条，没有则输出"无"。 对话内容： {dialogue}"""

注意：自动提炼事实一定要加人工确认环节。我测试时遇到过模型把用户随口说的玩笑话当成真实偏好存下来，导致后续对话一直跑偏。记忆这东西，宁可少记，不可错记。

5. 实操中踩过的坑：那些文档不会告诉你的细节

理论和代码讲完了，这一节专门讲我在实际接入 SQLite 过程中踩的坑。这些细节在官方文档里基本找不到，但每一个都能让你卡上半天。

5.1 数据库文件被锁：最常见也最烦人

SQLite 有个特性叫文件锁：同一时刻只允许一个写操作。如果你开了两个 Python 进程同时写同一个 db 文件，第二个会报database is locked。我调试时经常一边跑 Agent 一边用命令行工具查数据，结果两边打架。

解决办法有几个：一是设置timeout参数，让连接在锁定时等待而不是立刻报错；二是开启 WAL 模式，它允许读写并发，大幅减少锁冲突。

self.conn = sqlite3.connect(db_path, timeout=10, check_same_thread=False) self.conn.execute("PRAGMA journal_mode=WAL")

WAL 模式（Write-Ahead Logging）的原理是：写操作先写到一个单独的日志文件，读操作继续读主文件，两者互不阻塞。实测下来，开启 WAL 后锁冲突几乎消失，强烈建议默认开启。

5.2 忘记 commit：数据"凭空消失"

这是新手最常犯的错。SQLite 默认在事务里操作，你execute了 INSERT，但没commit，数据只存在于连接的内存里，程序一退出就没了。我早期调试时经常遇到"明明插入了却查不到"，排查半天发现是漏了 commit。

我的习惯是每个写方法结尾必写 commit，或者用上下文管理器统一管理。如果你用with conn:语法，退出时会自动 commit，能省不少心。

5.3 时间戳精度问题：同一秒内的消息顺序乱了

用整数秒做时间戳有个隐患：如果同一秒内插入多条消息，它们的created_at完全相同，ORDER BY created_at的顺序就不确定了。我遇到过 Agent 把用户和助手的消息顺序搞反的情况，就是因为这个。

解决办法有两个：一是用毫秒级时间戳（int(time.time() * 1000)），二是排序时加个id作为次级排序键。我两个都用上了，双保险。

cursor.execute( """SELECT role, content FROM messages WHERE session_id = ? ORDER BY created_at DESC, id DESC LIMIT ?""", (session_id, limit) )

5.4 数据库文件膨胀：定期清理旧数据

Agent 跑久了，messages 表会越来越大。我有个测试项目跑了两周，db 文件涨到了几十兆，虽然不算大，但查询明显变慢。后来我加了个清理逻辑：删除 30 天前的消息，但保留 facts 表。因为对话历史的价值随时间衰减，而长期事实是持久的。

def cleanup_old_messages(self, days=30): cutoff = int(time.time()) - days * 86400 cursor = self.conn.cursor() cursor.execute("DELETE FROM messages WHERE created_at < ?", (cutoff,)) self.conn.commit() cursor.execute("VACUUM") # 回收空间

VACUUM这个命令值得说一下：SQLite 删除数据后，文件大小不会自动缩小，空间只是被标记为可复用。VACUUM会重建整个数据库文件，真正释放空间。但它比较耗时，别频繁执行，我一般一个月跑一次。

6. 把记忆层接进 Agent 主循环：完整链路串一遍

前面都是零件，这一节把它们组装成一台能跑的机器。我会用一个最小的 Agent 主循环，展示记忆层是怎么嵌入进去的。

6.1 启动时初始化与恢复

Agent 启动时，第一件事是初始化数据库连接，然后决定是新建会话还是恢复旧会话。我的做法是：默认恢复最近活跃的会话，如果用户想开新的，再显式创建。

def get_or_create_session(store): cursor = store.conn.cursor() cursor.execute("SELECT id FROM sessions ORDER BY last_active_at DESC LIMIT 1") row = cursor.fetchone() if row: return row["id"] return store.create_session("默认会话")

6.2 每轮对话的读写流程

一轮完整的对话，记忆层的参与分四步：读历史、拼 prompt、调模型、写结果。顺序不能乱，尤其是写结果要在模型返回之后。

def chat_round(store, session_id, user_input, llm_call): # 1. 写入用户消息 store.add_message(session_id, "user", user_input) # 2. 读取上下文 history = store.get_context_within_budget(session_id, max_chars=4000) system_prompt = store.build_system_prompt() # 3. 调用模型 messages = [{"role": "system", "content": system_prompt}] + history reply = llm_call(messages) # 4. 写入助手回复 store.add_message(session_id, "assistant", reply) return reply

这个流程看起来简单，但有个细节要注意：用户消息要在读取上下文之前写入，否则这一轮的用户输入不会出现在 history 里。我第一次写的时候顺序搞反了，导致模型总是"看不到"用户刚说的话，回答驴唇不对马嘴。

6.3 会话切换与事实持久化

当用户开启新会话时，旧会话的对话历史留在数据库里，但 facts 表是全局共享的。这意味着用户在新会话里，Agent 依然记得他的长期偏好。这正是分表设计带来的好处——会话隔离，但事实共享。

def start_new_session(store, title): new_id = store.create_session(title) return new_id

新会话创建后，build_system_prompt依然会读取全局 facts，所以 Agent 的"人格记忆"是连续的，只有具体对话是断开的。这个设计我觉得挺符合直觉：就像你换了个话题跟朋友聊天，但你朋友还是记得你叫什么、喜欢什么。

6.4 一个容易忽略的点：异常时的数据一致性

如果模型调用失败抛异常，用户消息已经写进数据库了，但助手回复没有。这时候数据库里就留下了一条"孤儿"用户消息。下次读取上下文时，模型会看到用户说了话但自己没回，可能产生困惑。

我的处理方式是用 try-except 包住模型调用，失败时要么回滚用户消息，要么写入一条错误提示作为助手回复。我倾向于后者，因为保留用户输入更符合真实场景。

try: reply = llm_call(messages) except Exception as e: reply = f"[系统提示：本次回复生成失败，原因：{e}]" store.add_message(session_id, "assistant", reply)

7. 从 SQLite 出发：记忆库还能怎么演进

SQLite 解决了持久化的问题，但它只是记忆库的起点。当你把基础跑通之后，会发现还有更大的空间可以折腾。

第一个方向是语义检索。现在取历史靠的是时间顺序，但有时候 Agent 需要的是"跟当前话题相关的历史"。这就需要把消息转成向量存起来，查询时按相似度召回。SQLite 本身不支持向量检索，但可以配合专门的向量库，或者用 SQLite 的扩展。这块我打算后面单独花几天研究。

第二个方向是记忆的衰减与强化。人脑的记忆是有权重的，常用的记得牢，不用的慢慢忘。Agent 也可以这样：给每条记忆加个权重，被引用一次就加权，长期不用就降权，清理时优先删低权重的。这个思路能让记忆库更"聪明"，而不是简单按时间删。

第三个方向是多 Agent 共享记忆。如果多个 Agent 协作，它们能不能共享一个记忆库？这时候 SQLite 的单文件特性反而成了限制，可能需要换成支持网络访问的数据库。不过这是后话，现阶段单机 SQLite 完全够用。

我个人在实际操作中的体会是：别一上来就追求完美方案。我见过有人为了"一步到位"，直接上向量数据库加复杂架构，结果基础对话还没跑通就卡在环境配置上。先用 SQLite 把记忆跑起来，让 Agent 真正能记住东西，再根据实际痛点逐步升级，这才是靠谱的路径。数据库这东西，够用就好，过度设计只会拖慢你的迭代速度。

最后分享一个小技巧：调试记忆库时，我习惯写一个简单的命令行工具，能直接查最近的消息和所有 facts。这样排查问题时不用改代码，敲几行命令就能看到数据库里到底存了什么。这个工具我用了很久，比任何花哨的调试器都实在。