【硬核实战】放弃调包！从零手写企业级 AI Agent 核心记忆引擎（附 Python 完整源码）
如果你最近在研究 AI Agent，大概率会遇到一个非常现实的问题：
为什么现在的 Agent 看起来很聪明，但聊几轮之后就开始“失忆”？
你告诉它自己的技术栈，它下一轮可能就忘了。
你让它记住某个项目的上下文，重新启动程序之后，它又像第一次见面一样。
更麻烦的是，当对话数量从几十轮增长到几千、几万轮之后，简单地把历史消息全部塞进 Prompt，不仅效果越来越差，还会带来巨大的 Token 消耗。
这其实暴露出了 AI Agent 一个非常核心的基础设施：
Memory —— 记忆系统。
很多教程直接告诉你：
from langchain.memory import ...
然后调用几个 API，记忆功能就实现了。
但如果你真正准备构建一个企业级 Agent，这远远不够。
今天我们不依赖任何 Agent 框架，直接从零拆解一个 AI Agent Memory Engine，看看：
-
短期记忆到底是什么？
-
长期记忆为什么需要向量数据库？
-
Embedding 在其中到底干什么？
-
如何进行语义检索？
-
如何设计 Memory 的生命周期？
-
如何避免历史消息无限膨胀？
-
一个最小可运行的记忆引擎应该怎么写？
最后，我们还会使用 Python 手写一个简化版核心实现。
一、AI Agent 为什么需要 Memory？
传统的大模型调用其实非常简单：
User
↓
Prompt
↓
LLM
↓
Answer
模型本身并不会天然保存你的历史聊天记录。
例如第一次：
用户：我是一名 Python 开发者。
AI：好的，我知道了。
第二次重新请求：
用户：我最擅长什么语言？
如果没有把上一轮信息传给模型，它并不知道答案。
因此最基础的解决办法就是：
历史消息
↓
拼接 Prompt
↓
LLM
例如：
messages = [
{"role": "user", "content": "我是 Python 开发者"},
{"role": "assistant", "content": "好的"},
{"role": "user", "content": "我最擅长什么语言？"}
]
这种方式可以工作。
但问题很快就出现了。
假设一个 Agent 连续运行 1000 轮：
第 1 轮
第 2 轮
第 3 轮
...
第 1000 轮
如果每次都把全部历史记录发送给模型：
1000 条消息
↓
Prompt
↓
LLM
Token 会越来越大。
最终就会出现：
上下文越来越长、成本越来越高、响应越来越慢。
所以真正的 Agent Memory，不是简单保存聊天记录。
而是：
根据当前任务，只把真正有价值的历史信息取出来。
这就是记忆引擎存在的意义。
二、Agent Memory 的两种核心形态
一个比较合理的 Agent 记忆系统，通常可以拆成两层：
AI Agent
│
┌─────────┴─────────┐
│ │
Short Memory Long Memory
│ │
当前上下文 历史知识
│ │
Redis/内存 Vector Database
简单来说：
短期记忆负责“我现在正在聊什么”。
长期记忆负责“以前发生过什么”。
三、短期记忆：解决当前任务上下文
Short-term Memory 可以理解成 Agent 的“工作记忆”。
例如：
用户：帮我分析一个 Python 项目
AI：可以，把代码发给我。
用户：项目使用 FastAPI。
AI：明白。
用户：数据库是 MySQL。
AI：好的。
这里：
FastAPI
MySQL
Python
当前项目
都是当前任务非常重要的信息。
因此短期记忆应该保存：
最近 N 轮对话
当前任务状态
当前用户输入
Agent 中间状态
工具调用结果
一个简单的数据结构可能是：
short_memory = [
{
"role": "user",
"content": "我的项目使用 FastAPI"
},
{
"role": "assistant",
"content": "明白"
}
]
但短期记忆不能无限增长。
通常需要设置：
最大 Token
最大消息数量
滑动窗口
自动摘要
例如只保留最近 20 条消息：
short_memory = short_memory[-20:]
但是这里还有一个问题。
如果第 1 轮对话里出现了非常重要的信息：
用户：我的服务器使用 Ubuntu 24.04。
到了第 100 轮，这条消息已经被滑动窗口删除。
Agent 又不知道了。
怎么办？
答案就是：
长期记忆。
四、长期记忆：让 Agent 真正“记住”用户
Long-term Memory 的核心思想是：
把重要信息从对话上下文中抽取出来，并永久保存。
例如：
用户：
我主要做 Python 和网络安全，
以后写代码尽量使用 Python。
↓
Memory Extractor
↓
{
"content": "用户主要使用 Python，并从事网络安全相关工作",
"type": "preference"
}
这条信息不应该一直占据 Prompt。
而应该进入长期记忆。
当未来用户问：
帮我写一个自动化工具。
Agent 可以先进行 Memory Search：
当前问题
↓
Embedding
↓
Vector Search
↓
找到相关历史记忆
↓
加入 Prompt
↓
LLM
这样 Agent 就重新获得了相关背景。
五、为什么长期记忆需要向量数据库？
这是整个 Memory Engine 最关键的一部分。
假设数据库里面保存了：
用户喜欢 Python
用户使用 Ubuntu
用户正在学习网络安全
用户正在开发一个 CTF 平台
用户喜欢使用 Markdown
现在用户输入：
帮我写一个 Linux 自动化脚本。
我们怎么判断哪些历史记忆相关？
传统 SQL：
SELECT * FROM memory
WHERE content LIKE '%Linux%';
存在一个明显的问题：
关键词相同，不代表语义相同。
例如：
Linux
Ubuntu
Debian
Kali
服务器
这些词可能没有完全相同的字符串，但语义非常接近。
所以需要把文本转换成向量。
六、Embedding 到底是什么？
Embedding 可以简单理解成：
把一句自然语言转换成一组数字。
例如：
"我喜欢使用 Python"
经过 Embedding：
[0.12, -0.43, 0.87, 0.21, ...]
另一句话：
"我经常用 Python 写代码"
可能得到：
[0.11, -0.41, 0.85, 0.23, ...]
两组向量非常接近。
而：
"我喜欢吃苹果"
得到的向量可能距离比较远。
于是我们就可以通过：
向量距离
判断两个文本的语义相似程度。
七、向量数据库的基本工作流程
整个长期记忆系统可以理解成：
用户消息
│
▼
Memory Extractor
│
▼
重要信息
│
▼
Embedding Model
│
▼
Vector
│
▼
Vector Database
当下一次用户提问：
用户 Query
│
▼
Embedding
│
▼
Query Vector
│
▼
Vector Search
│
▼
Top-K Memories
│
▼
Prompt
最终：
System Prompt
+
当前问题
+
相关历史记忆
↓
LLM
这就是目前很多 Agent Memory 系统的基本思想。
八、从零手写一个 Memory Engine
下面我们不使用 LangChain、LlamaIndex 等 Agent 框架。
直接使用 Python 实现一个简化版本。
为了让代码更容易理解，我们先用内存结构模拟向量数据库。
核心流程只有几个步骤：
add_memory()
↓
生成向量
↓
保存
search()
↓
Query Vector
↓
计算相似度
↓
返回 Top-K
九、Python 核心源码
下面这段代码就是整个 Memory Engine 的核心逻辑。
from dataclasses import dataclass
from typing import List
import math
import hashlib
@dataclass
class Memory:
content: str
vector: List[float]
score: float = 0.0
class MemoryEngine:
def __init__(self):
self.memories = []
def embed(self, text: str) -> List[float]:
"""
简化版 Embedding。
实际项目中应该替换成真实 Embedding API。
"""
digest = hashlib.sha256(text.encode()).digest()
vector = [
byte / 255.0
for byte in digest[:32]
]
return vector
def cosine_similarity(
self,
a: List[float],
b: List[float]
) -> float:
dot = sum(
x * y
for x, y in zip(a, b)
)
norm_a = math.sqrt(
sum(x * x for x in a)
)
norm_b = math.sqrt(
sum(x * x for x in b)
)
if norm_a == 0 or norm_b == 0:
return 0.0
return dot / (norm_a * norm_b)
def add_memory(self, content: str):
vector = self.embed(content)
memory = Memory(
content=content,
vector=vector
)
self.memories.append(memory)
def search(
self,
query: str,
top_k: int = 3
):
query_vector = self.embed(query)
results = []
for memory in self.memories:
score = self.cosine_similarity(
query_vector,
memory.vector
)
results.append(
Memory(
content=memory.content,
vector=memory.vector,
score=score
)
)
results.sort(
key=lambda x: x.score,
reverse=True
)
return results[:top_k]
if __name__ == "__main__":
engine = MemoryEngine()
engine.add_memory(
"用户主要使用 Python 进行开发"
)
engine.add_memory(
"用户正在学习网络安全"
)
engine.add_memory(
"用户的服务器系统主要使用 Linux"
)
engine.add_memory(
"用户喜欢使用 Markdown 编写技术文章"
)
results = engine.search(
"Linux 服务器开发",
top_k=3
)
for item in results:
print(
f"{item.score:.4f} -> "
f"{item.content}"
)
需要特别说明：
上面的 embed() 只是为了演示 Memory Engine 的原理。
真正的生产环境不能使用 SHA256 产生的伪向量作为语义 Embedding。
生产环境应该替换为：
OpenAI Embedding
BGE
M3E
Jina Embeddings
Voyage
其他 Embedding Model
然后把向量存入真正的 Vector Database。
十、真正的企业级架构应该怎么设计？
如果把刚才的 Demo 升级成生产系统，可以设计成：
User
│
▼
AI Agent
│
┌────────┴────────┐
│ │
Short Memory Memory Manager
│ │
Redis │
▼
Memory Extractor
│
▼
Embedding
│
▼
Vector Database
│
┌─────────┴─────────┐
│ │
Milvus Qdrant
│ │
└─────────┬─────────┘
│
▼
Memory Search
这里建议把 Memory Manager 独立出来。
不要让 Agent 自己直接操作数据库。
也就是说：
agent -> memory_manager -> vector_db
而不是：
agent -> vector_db
原因很简单：
未来如果你把：
Qdrant
换成：
Milvus
或者：
pgvector
Agent 本身不需要修改。
十一、Memory Manager 应该负责什么？
一个真正可扩展的 Memory Manager，至少需要处理：
1. 写入
2. 查询
3. 删除
4. 更新
5. 去重
6. 过期
7. 重要性评分
8. 相似度评分
9. 用户隔离
10. 租户隔离
例如：
memory_manager.remember(
user_id="10001",
content="用户喜欢 Python"
)
查询：
memories = memory_manager.recall(
user_id="10001",
query="推荐开发语言"
)
最终得到：
用户喜欢 Python
然后把结果注入 Agent：
context = "\n".join(
item.content
for item in memories
)
再构造 Prompt：
你是一个 AI Agent。
用户相关长期记忆：
用户喜欢 Python
用户主要从事网络安全相关工作
当前问题：
帮我写一个自动化工具。
这样模型就拥有了“个性化记忆”。
十二、企业级 Memory 不能只看相似度
这是很多初学者容易忽略的问题。
假设数据库里有：
Memory A：
用户昨天问过 Python。
Memory B：
用户明确要求以后所有示例优先使用 Python。
Memory C：
用户喜欢使用深色主题。
当查询：
Python 开发
可能 A 和 B 都非常相似。
但是：
B 的重要性显然更高。
因此企业级 Memory 通常需要综合计算：
最终分数 =
语义相似度
× 权重
+
重要性
× 权重
+
时间衰减
× 权重
例如：
final_score = (
similarity * 0.6
+ importance * 0.3
+ recency * 0.1
)
这会比单纯：
cosine similarity
更加合理。
十三、时间衰减机制
记忆并不是永久保持同样的重要性。
例如：
用户今天正在学习 Docker
这可能非常重要。
但是半年之后：
这条信息的重要性可能下降。
可以设计一个简单的时间衰减函数：
recency = e^(-λt)
其中：
t = 距离上次访问的时间
λ = 衰减速度
这样系统就可以自动降低长期没有使用的记忆权重。
最终形成：
新记忆
↓
高权重
长期未使用
↓
权重下降
长期无价值
↓
删除/归档
这实际上已经开始接近真正的 Agent Memory Architecture。
十四、短期记忆 + 长期记忆才是合理方案
最终可以把整个系统理解成：
AI Agent
│
Memory Manager
│
┌────────────┴────────────┐
│ │
Short Memory Long Memory
│ │
最近对话 用户画像
当前任务 项目知识
工具结果 历史经验
│ │
Redis Vector DB
短期记忆解决：
现在发生了什么？
长期记忆解决：
过去发生过什么？
而 Memory Manager 解决：
什么值得记住？什么时候应该想起来？
十五、真正难的其实不是“存”
很多人做 Agent Memory，第一个想到的是：
把聊天记录存起来。
但真正困难的问题其实是：
什么应该存？
例如用户说：
今天北京天气不错。
这通常没有长期价值。
但如果用户说：
以后写 Python 示例时，不要使用第三方库。
这就非常值得记忆。
因此 Memory Extractor 可以让模型进行判断：
当前消息
↓
LLM
↓
是否值得记忆？
│
├── NO → 丢弃
│
└── YES
↓
提取记忆
↓
写入 Vector DB
这才是真正意义上的：
智能记忆。
十六、Agent Memory 的下一步：Memory Consolidation
当 Agent 运行时间越来越长，会出现一个问题：
数据库里面可能存在大量重复记忆。
例如：
用户使用 Python
用户主要使用 Python
用户喜欢 Python
用户经常使用 Python
用户开发主要使用 Python
这些实际上表达的是同一件事情。
因此需要一个：
Memory Consolidator
定期执行：
重复记忆检测
↓
语义聚类
↓
合并
↓
生成新的长期记忆
↓
删除旧记忆
最终可能变成：
用户主要使用 Python 进行开发。
这就是从简单的：
Memory Storage
逐渐进化到：
Memory Management
十七、生产环境建议的技术栈
如果准备真正做一个企业级 AI Agent，可以考虑：
语言：
Python
API：
FastAPI
缓存：
Redis
关系数据库：
PostgreSQL
向量数据库：
Qdrant / Milvus / pgvector
Embedding：
BGE / OpenAI Embedding / Jina
LLM：
GPT / Claude / Gemini / Qwen 等
容器：
Docker
监控：
Prometheus + Grafana
整体架构：
Client
│
▼
FastAPI
│
▼
Agent
│
┌────────┴────────┐
│ │
Redis Memory Manager
│
┌───────┴───────┐
│ │
PostgreSQL Vector DB
这套架构已经足够支撑一个中小型 Agent 项目的 Memory 基础设施。
十八、最后总结
如果你只记住今天这篇文章里的几个核心概念，我建议记住下面这张图：
AI Agent
│
▼
Memory Manager
│
┌────────────┴────────────┐
│ │
Short Memory Long Memory
│ │
最近上下文 历史知识
当前任务 用户画像
工具状态 项目经验
│ │
Redis Vector DB
│
Embedding
│
Semantic Search
短期记忆不是长期记忆的替代品。
短期记忆负责保持当前上下文。
长期记忆负责保存真正有价值的信息。
而向量数据库解决的是：
如何从海量历史信息中，快速找到与当前问题最相关的记忆。
真正优秀的 Agent，并不是把所有历史记录全部塞进 Prompt。
而是：
知道什么时候记住、记住什么，以及什么时候应该想起来。
这才是 AI Agent Memory Engine 真正的核心。
如果你准备进一步把这个 Demo 做成真正可以运行的项目，那么下一步可以加入：
Qdrant
Docker Compose
FastAPI
Redis
PostgreSQL
真实 Embedding
Memory Extractor
Memory Consolidation
用户级 Memory 隔离
最终形成一个真正可以接入 GPT、Claude、Qwen 等大模型的独立 Memory Service。
完整项目源码和 Docker 部署配置已打包，欢迎在评论区或主页简介获取完整工具包。
中国智能体开发者社区，聚焦智能体与大模型开发，提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动，促进经验交流与协作，助力开发者快速构建创新智能应用。
更多推荐
19
0- 0
已为社区贡献4条内容
所有评论(0)