用短期缓冲 + 哈希嵌入 + 语义检索实现 AI agent 的跨进程记忆,无需 LangChain/Pinecone,仅用 Llama 3.1 和纯 Python 实现。代码精简可复现。
“从零开始学 Agentic AI”系列的第 5 个项目,是一个具备记忆能力的对话式 Agent——整个项目的核心在于:你在一个进程中告诉它某个事实,终止这个进程,再启动一个全新的进程,它依然记得。
不用 LangChain,不用 Pinecone,也不用付费的 embedding API。只需要 Python、一个免费的 8B 模型(通过 NVIDIA NIM 使用 Llama 3.1),以及大约 200 行代码。下面来看看这四个部分是如何协同工作的。
最近几轮对话会逐字保留,让模型能够获得即时的对话上下文。它本质上是一个设有 token 预算的简单 deque——最近的对话轮次成本低、信号密度高,因此会以原始形式保留下来。
更早的事实会被转换成 embedding 并存储起来,以便后续进行语义召回。这里的 embedder 特意设计成零依赖:一个稳定的哈希 embedder。
class HashingEmbedder:
def embed(self, text):
vec = [0.0] * self.dim
for tok in text.lower().split():
h = int(hashlib.md5(tok.encode()).hexdigest(), 16)
vec[h % self.dim] += 1.0 if (h >> 8) & 1 else -1.0
return l2_normalize(vec) # cosine == dot product after this
带符号的桶可以防止向量坍缩成全正值;经过 L2 normalization 后,余弦相似度就能转化为普通的点积计算。检索则是在所有已存储向量上执行 argmax。承担核心功能的召回能力,所需基础设施为零。
当短期缓冲区溢出时,Agent 不会只是简单丢弃较早的对话,而是让模型将这些对话总结成一条紧凑的记忆笔记,再把它存入长期记忆。
在真实录制的运行过程中,这项机制在第一个 session 中触发了两次:原始对话中的 “I'm allergic to peanuts” 被压缩成了第 6 条记忆——“Devanshu is allergic to peanuts.”——在保留事实的同时,释放了大约 80 个 token。
用户每发起一轮对话,都会触发一次检索。只有相似度高于阈值的记忆才会被注入 prompt,其余记忆会被跳过,从而让上下文始终保持精简:
[INJECT] #6 (0.408) Devanshu is allergic to peanuts.
[INJECT] #2 (0.354) ...
[skip] #4 (0.121) ...
记忆存储会持久化到一个普通的 memory_store.json 文件中。录制的运行过程由两个真正独立的 OS 进程组成。
Session 1(进程 A):用户告诉 Agent 自己的名字、过敏信息以及一个项目细节;缓冲区发生溢出;压缩机制触发;进程退出。
Session 2(一次全新的 python run.py session2 运行):用户询问 “what am I allergic to?” → Agent 以 0.408 的分数召回第 6 条记忆,并回答花生。用户继续询问项目 → Agent 以 0.500 的分数召回了关于 Safar / Flutter + Supabase 的摘要。
整个系统只在两件事上使用模型:编写压缩摘要,以及生成最终答案。存储、embedding、检索和相关性评分全部由确定性的 Python 代码完成——也正因为如此,这套记忆机制是可审计的,而不是什么魔法。
真实录制的 transcript(包含压缩事件和每条记忆的相关性分数)以及完整代码均已提供。
现场演示:项目的 Memory Agent 演示页面。
代码仓库:dev48v/agentic-ai-from-zero。
接下来是项目 6:一个 human-in-the-loop 审批 Agent,在执行任何有风险的操作之前,它都会暂停并等待人工批准。
如果需要采取进一步措施,你可以考虑屏蔽此人和/或举报滥用行为。