Headroom Memory 深度解析:跨 Agent 分层记忆系统,让 LLM 在对话间持久记忆
【免费下载链接】headroomCompress tool outputs, logs, files, and RAG chunks before they reach the LLM. 20% fewer tokens for coding agents, 60-95% fewer tokens for JSON, same answers. Library, proxy, MCP server.项目地址: https://gitcode.com/GitHub_Trending/head/headroom
Headroom 的 Memory 模块是一套分层(Hierarchical)、带时间版本(Temporal)的记忆系统,目标只有两件事:解决 LLM 上下文窗口被历史撑爆的问题,以及解决"每次对话都从零开始"的无持久化问题。它的核心思路是时间压缩(temporal compression)——不携带 10,000 token 的对话历史,而是提取 100 token 的关键事实持久化下来,在下一次请求中按相关性自动注入。读完本文,你将掌握三种使用方式:通过代理(proxy)让多个 Agent 共享同一个记忆库、用一行with_memory(client)包装任意 OpenAI 兼容客户端、以及直接使用HierarchicalMemory底层 API 做细粒度检索与版本管理,并理解其 Protocol 化的插件式架构、LLM 中介去重机制和嵌入/存储层的完整配置。
一、设计动机:为什么需要"记忆"
LLM 有两个根本性限制:
- 上下文窗口溢出——历史太多,必须截断;
- 无持久化——每个会话都从零开始。
Headroom Memory 的解法是:提取关键事实 → 持久化 → 相关时注入(extract key facts, persist them, inject when relevant)。这与 Headroom 项目的主线(压缩工具输出、日志、RAG 分片以节省 token)一脉相承——记忆本质上是一种更激进的压缩:跨会话的、结构化的、可版本化的压缩。
与其他记忆方案的能力对比
| 特性 | Headroom | Letta (MemGPT) | Mem0 |
|---|---|---|---|
| 跨 Agent 记忆 | 任意 Agent 经代理共享一个 DB | 仅单 Agent 内 | 按用户隔离,无跨 Agent |
| Agent 来源追踪(Provenance) | 记录每条记忆由哪个 Agent 保存/更新 | 无 | 无 |
| LLM 中介去重 | 复用用户自己的 LLM 做合并决策 | 无 | 需要额外的 LLM 调用($) |
| 透明代理 | 零代码改动——请求经代理即可 | 需要 Agent 框架 | 需要 SDK 集成 |
| 分层 Scope | User → Session → Agent → Turn | 扁平(按 Agent) | 扁平(按用户) |
| 时间版本 | 完整的 supersession 链 | 无 | 无 |
| 零延迟提取 | 行内(Letta 风格) | 行内 | 独立调用 |
| 一行集成 | with_memory(client) | 需要 Agent 搭建 | 需要独立客户端 |
| 可插拔后端 | SQLite、HNSW、FTS5、任意 embedder | PostgreSQL | Qdrant/Chroma |
| 语义 + 全文检索 | 都有 | 仅语义 | 仅语义 |
| 记忆上浮(Bubbling) | 自动提升高重要度记忆 | 无 | 无 |
| Protocol 化架构 | 是(依赖注入) | 否 | 否 |
与 Letta/Mem0 的定位差异可以这样概括:Letta 是带记忆的完整 Agent 框架,Mem0 是托管记忆服务,Headroom 是把记忆作为一层叠加在你现有栈上的基础设施——内嵌部署、无外部服务、语义+全文双检索、Protocol 化可扩展。
二、最强玩法:代理模式下的跨 Agent 共享记忆
任何经过 Headroom 代理路由的 Agent 都共享同一个记忆存储。Claude 保存的事实,Codex 可以直接读出来——零配置。
# 以启用记忆的方式启动代理 headroom proxy --memory # 或者用 wrap(自动拉起代理) headroom wrap claude --memory # Claude Code 带上持久记忆 headroom wrap codex --memory # Codex 使用同一记忆库 headroom wrap aider --memory # Aider 也共享请求处理流程
Claude Code Codex CLI Gemini CLI │ │ │ └── /v1/messages ──┐ └── /v1/chat/completions ──┤ └── /generateContent ──┐ ▼ ▼ ▼ ┌──────────────────────────────────────────────────────────────────┐ │ Headroom Proxy (--memory) │ │ 1. 搜索记忆库中的相关上下文 │ │ 2. 以 provider 原生格式注入 system context │ │ 3. 注入 memory_save/search/update/delete 工具 │ │ 4. 转发到上游 LLM │ │ 5. 处理响应中的 memory 工具调用 │ │ 6. 后台异步去重(相似度 >92% 自动删除旧副本) │ └──────────────────────┬───────────────────────────────────────────┘ ▼ .headroom/memory.db (项目级作用域的 SQLite)这条流程在源码中对应 memory_handler.py:注入与工具调用处理都发生在代理层,各 provider 的上下文注入格式不同——
| Provider | 上下文注入方式 | 记忆工具格式 |
|---|---|---|
| Anthropic(Claude) | system参数 | Anthropictool_use |
| OpenAI(Codex, GPT) | system 消息 | OpenAI function calling |
| Gemini | systemInstruction | functionDeclarations |
| 任意 OpenAI 兼容 | system 消息 | function calling |
关键 CLI 参数(来自 proxy.py)
从源码中的选项定义可以看到完整的记忆开关体系:
--memory:启用持久记忆,自动识别 provider 并使用对应工具格式;--memory-db-path:记忆 DB 路径,默认{cwd}/.headroom/memory.db,可用环境变量HEADROOM_MEMORY_DB_PATH覆盖;--memory-storage:分区策略,取值project(默认)/user/global。project模式下每个解析后的 workspace 各自拥有<db_path_dir>/memories/projects/<basename>-<hash>/memory.db,杜绝跨项目串味;user模式按x-headroom-user-id一人一库;global为单一共享库(兼容旧行为);--no-memory-tools/--no-memory-context:分别禁用工具注入与上下文注入(只留其一也可以);--memory-top-k:每次请求注入的记忆条数;--memory-qdrant-url/--memory-qdrant-host/--memory-qdrant-port/--memory-qdrant-api-key:接入外部 Qdrant 向量库。
分区逻辑的实现入口在 storage_router.py,USER模式即"每个x-headroom-user-id一个 DB"。
项目级隔离的"文件系统契约"
文件系统契约注意:项目级记忆路径相对当前工作目录解析,且不遵循标准的
HEADROOM_WORKSPACE_DIR环境变量。这是刻意设计,用于保持"项目记忆隔离"不变量。如果你想要一个中央记忆库,请显式传入--memory-db-path。
用户身份
用户 ID 默认从$USER(操作系统用户名)自动检测,也可按请求用x-headroom-user-id头覆盖。该头的语义在 identity.py 中有明确声明:它是一个分区提示(partition hint),不是认证身份。所有记忆都按用户隔离——同一项目的多名开发者各拥有独立的记忆存储。
Agent 来源追踪(Provenance)
每条记忆都记录是哪个 Agent 创建/更新了它:
{ "content": "Project uses alembic for migrations", "metadata": { "source_agent": "claude", "source_provider": "anthropic", "created_via": "tool_call", "created_at_utc": "2026-04-10T17:30:00Z" } }Agent 更新记忆时同样留痕:
{ "reason": "Updated by codex via openai: Added version info" }LLM 中介的智能去重
当 LLM 调用memory_save时,Headroom 分四步处理:
- 立即保存(零延迟);
- 搜索相似已有记忆(余弦相似度);
- 发现相似项时返回增强提示,把合并决策交给用户的 LLM:
{ "status": "saved", "memory_id": "abc123", "note": "Similar memory exists (id: def456, 89% match, saved by codex): 'DB migration tool is alembic'. Call memory_update('def456', '<merged content>') to consolidate." }- 后台自动去重:相似度超过 92% 时,较旧的副本被异步、非阻塞地自动移除。
这个"提示阈值 0.75 / 自动移除阈值 0.92"的分层策略可以直接在源码中核实:memory_handler.py 定义了DEDUP_HINT_THRESHOLD = 0.75(相似度达到后向 LLM 建议合并),而自动去重阈值 0.92 定义在 bridge_config.py 的dedup_similarity_threshold: float = 0.92。这套机制的成本优势很关键:合并决策复用用户自己正在调用的 LLM,Headroom 侧不产生任何额外模型调用费用——这正是对比表中"LLM-Mediated Dedup: No extra cost"的来源。
三、快速上手:一行代码包装任意客户端
from openai import OpenAI from headroom import with_memory # 一行——就这样 client = with_memory(OpenAI(), user_id="alice") # 用法与平常完全一致 response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "I prefer Python for backend work"}] ) # 记忆在此响应中被行内提取——零额外延迟 # 稍后,在一段新对话中…… response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "What language should I use?"}] ) # → 响应会使用记忆里的 Python 偏好with_memory的完整签名(来自 wrapper.py):
def with_memory( client, # LLM 客户端(OpenAI、Anthropic、Mistral、Groq 等任意 OpenAI 兼容) user_id: str, # 记忆隔离的用户标识 db_path="headroom_memory.db",# SQLite 路径(默认 headroom_memory.db) top_k: int = 5, # 每次请求注入的记忆条数 session_id=None, # 会话作用域(可选) agent_id=None, # Agent 作用域(可选) embedder_backend=EmbedderBackend.LOCAL, # LOCAL 或 OPENAI 等 openai_api_key=None, # 使用 OpenAI 嵌入时的密钥 **kwargs, ) -> MemoryWrapperwith_memory在 headroom/init.py 中作为顶层懒加载导出("with_memory": ("headroom.memory", "with_memory")),因此无需安装任何额外组件即可from headroom import with_memory。
六步工作原理
┌─────────────────────────────────────────────────────────────┐ │ with_memory() │ │ 1. INJECT: 语义检索 → 拼接到用户消息前 │ │ 2. INSTRUCT: 追加记忆提取指令 │ │ 3. CALL: 转发给 LLM │ │ 4. PARSE: 从响应中解析 <memory> 块 │ │ 5. STORE: 连同嵌入 + 向量索引 + FTS 落库 │ │ 6. RETURN: 返回已剥离 memory 块的干净响应 │ └─────────────────────────────────────────────────────────────┘关键点:记忆提取是 LLM 响应的一部分(Letta 风格的行内提取),没有额外 API 调用、没有额外延迟。第 2、4 步的实现在 inline_extractor.py:MEMORY_INSTRUCTION告诉模型"如果有关于用户/实体的值得记住的事实,在响应末尾输出<memory>…</memory>块;无内容可存时输出<memory>{"memories": []}</memory>",然后用正则<memory>\s*(.*?)\s*</memory>解析并剥离该块——这就是第 6 步"干净响应"的来源。
四、分层作用域(Hierarchical Scoping)
记忆存在于不同的 scope 层级,实现细粒度控制:
USER (最宽) └── SESSION └── AGENT └── TURN (最窄)| Scope | 持久化范围 | 典型用例 |
|---|---|---|
| USER | 所有会话、永久 | 长期偏好、身份 |
| SESSION | 仅当前会话 | 当前任务上下文 |
| AGENT | 会话内的当前 Agent | Agent 专属上下文 |
| TURN | 仅单轮 | 临时工作记忆 |
这四个层级在源码中是一等公民:models.py 定义了ScopeLevel枚举(USER/SESSION/AGENT/TURN),Memory数据类携带user_id(必填)+session_id/agent_id/turn_id(可选)四个作用域字段,并有scope_level属性按"turn > agent > session > user"的优先级推导层级。
示例:跨会话记忆
from openai import OpenAI from headroom import with_memory # 会话 1:上午 client1 = with_memory( OpenAI(), user_id="bob", session_id="morning-session", ) response = client1.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "I prefer Go for performance-critical code"}], ) # 记忆以 USER 级存储(跨会话持久) # 会话 2:下午(不同会话,同一用户) client2 = with_memory( OpenAI(), user_id="bob", # 同一用户 session_id="afternoon-session", # 不同会话 ) response = client2.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "What language for my new microservice?"}] ) # → 从上午会话召回 Go 偏好!五、时间版本(Temporal Versioning / Supersession)
事实会随时间改变。当事实更新时,Headroom 不删除旧记忆,而是创建一条supersession 链保留历史:
from headroom.memory import HierarchicalMemory, MemoryConfig memory = await HierarchicalMemory.create() # 原始事实 orig = await memory.add( content="User works at Google", user_id="alice", category=MemoryCategory.FACT, ) # 用户换了工作——用 supersede 替代旧记忆 new = await memory.supersede( old_memory_id=orig.id, new_content="User now works at Anthropic", ) # 查询当前状态(默认排除已取代项) current = await memory.query( MemoryFilter( user_id="alice", include_superseded=False, # 默认值 ) ) # → 只返回 "User now works at Anthropic" # 查询完整历史(包含已取代项) history = await memory.query( MemoryFilter( user_id="alice", include_superseded=True, ) ) # → 返回两条记忆,带有效期时间戳 # 取整条链 chain = await memory.get_history(new.id) # → [ # Memory(content="User works at Google", valid_until=..., is_current=False), # Memory(content="User now works at Anthropic", valid_until=None, is_current=True), # ]版本机制的底层字段在 models.py 的Memory数据类中可以逐一对应:valid_from/valid_until(valid_until=None表示当前有效)、supersedes(本条取代了谁)、superseded_by(被谁取代)、is_current属性。对应的实现方法是 core.py 中的supersede()(L529)与get_history()(L640);MemoryFilter(ports.py)还支持valid_at时间点查询、has_supersedes血缘过滤等更细的能力。
为什么时间版本重要
- 审计追踪——任意时点"什么为真"都可查证;
- 调试——理解 LLM 为什么做出某个决策(当时它"记得"什么);
- 回滚——必要时恢复先前状态;
- 分析——跟踪用户偏好如何演变。
六、记忆分类与重要度
记忆按类别组织以便检索与调试。分类概念贯穿提取与保存链路(extraction.py 的提取指令就按"偏好 / 活动 / 关系 / 事件 / 计划目标"等维度引导模型):
| 类别 | 描述 | 示例 |
|---|---|---|
PREFERENCE | 喜好、偏好做法 | "Prefers Python"、"Likes dark mode" |
FACT | 身份、角色、约束 | "Works at fintech startup"、"Senior engineer" |
CONTEXT | 当前目标、进行中的任务 | "Migrating to microservices"、"Working on auth" |
ENTITY | 关于实体的信息 | "Project Apollo uses React"、"Team lead is Sarah" |
DECISION | 已做的决策 | "Chose PostgreSQL over MySQL"、"Using REST not GraphQL" |
INSIGHT | 推导出的洞见 | "User tends to prefer typed languages" |
除类别外,每条记忆还有一个0.0–1.0 的重要度分数(Memory.importance,默认 0.5)。分数越高越容易被检索命中,也是"记忆上浮"机制的判据——见下文配置中的bubble_threshold。
七、client.memory便捷 API
with_memory()包装器提供.memory属性用于直接操作:
client = with_memory(OpenAI(), user_id="alice") # 语义检索 results = client.memory.search("python preferences", top_k=5) for memory in results: print(f"{memory.content}") # 手动添加记忆 client.memory.add( "User is a senior engineer", category="fact", importance=0.9, ) # 取全部记忆 all_memories = client.memory.get_all() # 清空 client.memory.clear() # 统计 stats = client.memory.stats() print(f"Total memories: {stats['total']}")从 wrapper.py 的实现可以看到,这一层是对内部异步HierarchicalMemory的同步适配:get_all()组装MemoryFilter(user_id=...)调query(),clear()调clear_scope(user_id=...),stats()统计该用户名下记忆总数——隔离边界是 user_id,这也是多用户隔离的落点(见下节)。
八、进阶:直接使用 HierarchicalMemory API
需要完全控制时,直接使用HierarchicalMemory类(完整 API 在 core.py,create()为类方法工厂):
import asyncio from headroom.memory import ( HierarchicalMemory, MemoryConfig, MemoryCategory, EmbedderBackend, ) from headroom.memory.ports import MemoryFilter, VectorFilter async def main(): # 带自定义配置创建 config = MemoryConfig( db_path="my_memory.db", embedder_backend=EmbedderBackend.LOCAL, # 或 OPENAI、OLLAMA vector_dimension=384, cache_max_size=2000, ) memory = await HierarchicalMemory.create(config) # 完全控制地添加记忆 mem = await memory.add( content="User prefers functional programming", user_id="alice", session_id="sess-123", agent_id="code-assistant", category=MemoryCategory.PREFERENCE, importance=0.9, entity_refs=["functional-programming", "coding-style"], metadata={"source": "conversation", "confidence": 0.95}, ) # 语义检索 results = await memory.search( query="programming paradigm preferences", user_id="alice", top_k=5, min_similarity=0.5, categories=[MemoryCategory.PREFERENCE], ) for r in results: print(f"[{r.similarity:.3f}] {r.memory.content}") # 全文检索 text_results = await memory.text_search( query="functional", user_id="alice", ) # 带过滤条件的查询 memories = await memory.query( MemoryFilter( user_id="alice", categories=[MemoryCategory.PREFERENCE, MemoryCategory.FACT], min_importance=0.7, limit=10, ) ) # 便捷方法 await memory.remember("Likes coffee", user_id="alice", importance=0.6) relevant = await memory.recall("beverage preferences", user_id="alice") asyncio.run(main())MemoryFilter与VectorFilter的完整字段(ports.py)比示例展示的更多:除作用域、时间(created_after/created_before/valid_at/include_superseded)与重要度过滤外,还有实体过滤(entity_refs)、血缘过滤(has_supersedes/has_promoted_from)、分页(limit/offset)与排序(order_by可取created_at/importance/access_count/last_accessed)。search的默认参数为top_k=10、min_similarity=0.0(见 core.py)。
九、配置详解
9.1 嵌入后端(Embedder Backends)
from headroom.memory import MemoryConfig, EmbedderBackend # 本地嵌入(推荐——快、免费、私有) config = MemoryConfig( embedder_backend=EmbedderBackend.LOCAL, embedder_model="all-MiniLM-L6-v2", # 384 维,速度快 ) # OpenAI 嵌入(质量更高,收费) config = MemoryConfig( embedder_backend=EmbedderBackend.OPENAI, openai_api_key="sk-...", embedder_model="text-embedding-3-small", ) # Ollama 嵌入(本地服务器,模型多) config = MemoryConfig( embedder_backend=EmbedderBackend.OLLAMA, ollama_base_url="http://localhost:11434", embedder_model="nomic-embed-text", )从 config.py 看,EmbedderBackend实际还支持第四个值ONNX——ONNX Runtime 后端,无需 torch(约 86MB),源码注释标注为 "recommended"。本地模型默认值由 headroom/models/config.py 的ML_MODEL_DEFAULTS.sentence_transformer提供,即all-MiniLM-L6-v2(22M 参数、384 维、约 90MB)。MemoryConfig的__post_init__会做配置校验:选用 OpenAI 后端而未给openai_api_key直接抛ValueError;vector_dimension必须与嵌入模型输出维度一致(MiniLM 即 384)。
9.2 嵌入运行时 / GPU 卸载(Apple Silicon)
默认情况下,代理的记忆嵌入器跑在ONNX CPU后端——快、依赖轻,但纯 CPU。持续负载下嵌入步骤可能占满 CPU、拖慢代理响应。在 Apple Silicon 上可以显式切换到Apple GPU (MPS),把嵌入工作从 CPU 卸走,对无风扇 Mac(如 M5 Air)这类容易 CPU 饱和超时的机器尤其有用。
开启方式——安装 extra 并设置环境变量:
pip install 'headroom-ai[pytorch-mps]' # [pytorch_mps] 写法同样有效 export HEADROOM_EMBEDDER_RUNTIME=pytorch_mps设置后,嵌入器改为走 torch sentence-transformers 后端在 Apple GPU 上运行,取代默认的 ONNX CPU 嵌入器。行为细则(对应 memory_handler.py 的解析逻辑):
- 严格显式开启:
pytorch_mps是唯一被接受的值;其他任何值(或不设置)都保持默认 ONNX CPU 嵌入器,默认行为不变; - 自动回退:仅当 MPS 真正可用(Apple Silicon + torch)时才生效。MPS 不可用或 torch/sentence-transformers 未安装时,记录警告并回到既有的默认嵌入选择路径——ONNX 可用则用 ONNX,否则用既有的本地 sentence-transformers 回退;
- MPS 串行化:torch-MPS 不是线程安全的,嵌入器内部通过单 worker 执行器串行化 MPS 编码调用,全自动、无需配置。
pytorch-mpsextra 在 pyproject.toml 中声明,注释明确其为 macOS-only 且有意不并入[all]。
9.3 存储配置
config = MemoryConfig( db_path="memory.db", # SQLite 数据库路径 vector_dimension=384, # 必须与嵌入模型输出维度一致 hnsw_ef_construction=200, # HNSW 建图质量(越大越好、越慢) hnsw_m=16, # HNSW 每节点连接数 hnsw_ef_search=50, # HNSW 搜索质量 cache_enabled=True, # 启用 LRU 缓存 cache_max_size=1000, # 最大缓存条目数 )config.py 中MemoryConfig的默认值与上表一致(hnsw_ef_construction=200、hnsw_m=16、hnsw_ef_search=50、cache_enabled=True、cache_max_size=1000),此外还有几个值得知道的字段:
vector_backend:默认AUTO——自动选择SQLITE_VEC(若可用,内存有界,推荐),否则回退HNSW(hnswlib,不设hnsw_max_entries时无界);vector_cache_size_kb:SQLite 向量索引页缓存,默认 8MB;text_backend:默认FTS5;auto_bubble/bubble_threshold:记忆上浮开关与阈值,默认True/0.7——重要度超过阈值的记忆会被自动提升作用域层级(Memory数据类中的promoted_from/promotion_chain字段即记录这条上浮血缘,add()路径在 core.py 中触发_maybe_bubble);- 三个
EXTERNAL扩展点:store_backend_name/vector_backend_name/text_backend_name可从headroom.memory_store、headroom.memory_vector、headroom.memory_text入口点加载第三方适配器。
9.4 包装器配置
client = with_memory( OpenAI(), user_id="alice", db_path="memory.db", top_k=5, # 每次请求注入的记忆条数 session_id="optional-session", agent_id="optional-agent", embedder_backend=EmbedderBackend.LOCAL, )十、架构:Protocol 化的可插拔设计
Headroom Memory 的所有组件都通过Protocol 接口(ports)抽象,便于替换:
┌─────────────────────────────────────────────────────────────┐ │ HierarchicalMemory │ │ (Orchestrator) │ ├─────────────────────────────────────────────────────────────┤ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ MemoryStore │ │ VectorIndex │ │ TextIndex │ (Protocol) │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ ▼ ▼ ▼ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ SQLite │ │ HNSW │ │ FTS5 │ (Adapter) │ │ Adapter │ │ Adapter │ │ Adapter │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ ┌─────────────┐ ┌─────────────┐ │ │ │ Embedder │ │ MemoryCache │ (Protocol) │ └──────┬──────┘ └──────┬──────┘ │ │ ▼ ▼ │ │ ┌─────────────┐ ┌─────────────┐ │ │ │Local/OpenAI/ │ │ LRU Cache │ (Adapter) │ │ Ollama │ │ │ │ │ └─────────────┘ └─────────────┘ │ └─────────────────────────────────────────────────────────────┘| 组件 | Protocol | 默认适配器 | 职责 |
|---|---|---|---|
| MemoryStore | MemoryStore | SQLiteMemoryStore | CRUD + 过滤 + 版本取代 |
| VectorIndex | VectorIndex | HNSWVectorIndex | 语义相似度检索 |
| TextIndex | TextIndex | FTS5TextIndex | 全文关键词检索 |
| Embedder | Embedder | LocalEmbedder | 文本 → 向量 |
| Cache | MemoryCache | LRUMemoryCache | 热记忆缓存 |
从源码结构看,这套设计是真实落地的:Protocol 定义集中在 ports.py(含MemoryStore/VectorIndex/TextIndex/Embedder/MemoryCache及图存储相关的GraphStore),对应适配器文件一一对应位于 headroom/memory/adapters/——sqlite.py、hnsw.py、sqlite_vector.py、fts5.py、embedders.py、cache.py。编排器HierarchicalMemory只做组合与调度,组件通过依赖注入装配(factory.py 提供create_memory_system工厂)。[headroom/memory/](https://link.gitcode.com/i/7001ac8c44833ec4d1dabb0a705fcbff)目录下的其余文件则分别负责:wrapper.py(客户端包装)、wrapper_tools.py(工具调用式记忆with_memory_tools)、tools.py(MEMORY_TOOLS记忆函数定义)、extraction.py/inline_extractor.py(提取指令与解析)、bridge.py/bridge_parsers.py(代理侧桥接)、storage_router.py(项目/用户分区路由)、system.py(后端抽象)等。
十一、性能参考与多用户隔离
性能参考值(官方文档口径)
| 操作 | 延迟 | 说明 |
|---|---|---|
| 记忆注入 | <50ms | 本地嵌入 + 向量检索 |
| 记忆提取 | +50–100 token | 属于 LLM 响应的一部分(行内) |
| 记忆落库 | <10ms | SQLite + 向量索引 + FTS5 |
| 缓存命中 | <1ms | LRU 缓存查找 |
总开销:每次响应约 +100 输出 token(<memory>块)。注意这些是项目文档给出的参考量级,实际延迟取决于硬件与模型。
多用户隔离
记忆按user_id严格隔离:
# Alice 的记忆 alice_client = with_memory(OpenAI(), user_id="alice") # Bob 的记忆(完全独立) bob_client = with_memory(OpenAI(), user_id="bob") # 即使共用同一个数据库,Bob 也看不到 Alice 的记忆隔离在查询层生效:MemoryWrapper.get_all()永远附加MemoryFilter(user_id=...),search/recall同理,从 wrapper.py 的实现可以直接验证。
十二、完整对话流示例
from openai import OpenAI from headroom import with_memory client = with_memory(OpenAI(), user_id="developer_jane") # 对话 1:用户分享上下文 response = client.chat.completions.create( model="gpt-4o", messages=[ { "role": "user", "content": "I'm a Python developer at a fintech startup. We use PostgreSQL and FastAPI.", } ], ) # 提取出的记忆: # - [FACT] Python developer at fintech startup # - [PREFERENCE] Uses PostgreSQL for databases # - [PREFERENCE] Uses FastAPI for web APIs # 对话 2(新会话):用户提问 response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "What database should I use for my new project?"}], ) # 响应引用记忆中的 PostgreSQL 偏好: # → "Given your experience with PostgreSQL at your fintech company, # I'd recommend sticking with it for consistency..." # 查看已存记忆 print("Stored memories:") for m in client.memory.get_all(): print(f" [{m.category.value}] {m.content}")with_memory对任意 OpenAI 兼容客户端都成立——OpenAI 本身、Azure OpenAI(改base_url)、Groq 或其他兼容 SDK 直接传入即可:
# Azure OpenAI client = with_memory( OpenAI(base_url="https://your-resource.openai.azure.com/..."), user_id="alice", ) # Groq from groq import Groq client = with_memory(Groq(), user_id="alice")十三、故障排查与最佳实践
排错速查
记忆没有被提取
- 确认对话中有值得记忆的内容(不是寒暄);
- 确认 LLM 遵循了记忆指令;
- 打开调试日志:
import logging; logging.basicConfig(level=logging.DEBUG)。
记忆没有被检索到
- 确认跨会话的
user_id一致; - 用
client.memory.get_all()确认记忆存在; - 换更具体的检索词;
- 检查相似度阈值(
min_similarity)。
延迟偏高
- 使用本地嵌入:
embedder_backend=EmbedderBackend.LOCAL; - 调小
top_k减少检索/注入量; - 确保缓存开启(默认开启)。
记忆没有持久化
- 确认跨会话的
db_path一致; - 确认数据库文件可写;
- 检查日志中的异常。
最佳实践
user_id保持一致——同一 ID 跨会话才能延续;- 善用会话作用域——
session_id用于任务级上下文; - 从本地嵌入起步——快、免费、多数场景够用;
- 监控记忆增长——定期
client.memory.stats(); - 用重要度分数——越高越容易被检索与上浮;
- 用类别——方便调试与选择性检索;
- 事实变了用
supersede()——而不是再add()一条新的。
十四、测试与进一步阅读
记忆子系统有非常完整的测试覆盖,可作为行为依据继续深入:tests/test_memory/ 覆盖决策策略(test_memory_decision_policy.py)、注入预算(test_memory_injection_budget.py)、项目隔离(test_memory_handler_project_isolation.py)、存储路由(test_memory_storage_router.py)等;向量/全文索引与记忆系统集成测试在 tests/test_hnsw_only.py、tests/test_sqlite_vector_index.py、tests/test_memory_integration.py;代理侧记忆行为见 tests/test_proxy_memory_integration.py。
仓库内与本主题相关的其他文档:代理总览 wiki/proxy.md、文件系统契约(解释记忆路径为何不遵循HEADROOM_WORKSPACE_DIR)wiki/filesystem-contract.md、整体架构 wiki/ARCHITECTURE.md。
总结:Headroom Memory 把"跨 Agent 共享 + 分层作用域 + 时间版本 + 双模检索 + 零延迟行内提取"打包进一个内嵌式(无需外部服务)的 SQLite 体系,并全部做成 Protocol 可插拔。对现有栈最省力的接入方式是headroom proxy --memory或headroom wrap <agent> --memory;对代码集成则是一行with_memory(client);对需要精细控制的场景,直接使用HierarchicalMemory与MemoryConfig即可获得全部能力。
【免费下载链接】headroomCompress tool outputs, logs, files, and RAG chunks before they reach the LLM. 20% fewer tokens for coding agents, 60-95% fewer tokens for JSON, same answers. Library, proxy, MCP server.项目地址: https://gitcode.com/GitHub_Trending/head/headroom
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考