【硬核实战】放弃调包！从零手写企业级 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 ↓ LLMToken 会越来越大。最终就会出现上下文越来越长、成本越来越高、响应越来越慢。所以真正的 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 自动化脚本。我们怎么判断哪些历史记忆相关传统 SQLSELECT * 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( contentcontent, vectorvector ) 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( contentmemory.content, vectormemory.vector, scorescore ) ) results.sort( keylambda x: x.score, reverseTrue ) 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_k3 ) 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或者pgvectorAgent 本身不需要修改。十一、Memory Manager 应该负责什么一个真正可扩展的 Memory Manager至少需要处理1. 写入 2. 查询 3. 删除 4. 更新 5. 去重 6. 过期 7. 重要性评分 8. 相似度评分 9. 用户隔离 10. 租户隔离例如memory_manager.remember( user_id10001, content用户喜欢 Python )查询memories memory_manager.recall( user_id10001, query推荐开发语言 )最终得到用户喜欢 Python然后把结果注入 Agentcontext \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 部署配置已打包欢迎在评论区或主页简介获取完整工具包。