欢迎光临
我们一直在努力

【硬核实战】放弃调包!从零手写企业级 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 部署配置已打包,欢迎在评论区或主页简介获取完整工具包。

赞(0)
未经允许不得转载:171主机测评 » 【硬核实战】放弃调包!从零手写企业级 AI Agent 核心记忆引擎(附 Python 完整源码)
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址