Elastic 开源 Atlas:给 AI Agent 接上可检索的长期记忆

2026-06-30 36 预计阅读时间: 1 分钟
来源: infoq.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:10 分钟

Agent 真正难用的地方,往往不是一次回答不够聪明,而是下一次又把上下文忘干净。Elastic 开源的 Atlas 试图解决这个问题:它基于 Elasticsearch 构建,为 Agent 维护三类记忆,并通过 MCP 接入现有 Agent,同时支持按用户隔离记忆。在问答能力评估中,Atlas 达到了 0.89 的 Recall@10,这说明它在“把相关记忆找回来”这件事上有不错的基础表现。

Atlas 解决的是 Agent 的“记不住”问题

很多 Agent 系统会把历史对话直接塞进 prompt,短期看简单,长期看会遇到三个问题:上下文窗口有限、检索噪声越来越大、用户之间的数据边界不清晰。

Atlas 的方向更像是把 Agent 记忆产品化:

  • 用 Elasticsearch 作为底层检索与存储能力,而不是把记忆散落在临时缓存里。
  • 将记忆分成三类,方便 Agent 在不同场景下使用不同粒度的信息。
  • 通过 MCP 与 Agent 集成,减少对具体框架的绑定。
  • 维护 per-user isolation,避免一个用户的偏好、历史或事实泄漏到另一个用户。

这里的关键不是“Agent 有数据库”这么简单,而是 Agent 能以结构化方式写入、检索和更新记忆。

三类记忆为什么重要

来源摘要没有展开三类记忆的具体命名,因此不能把它们当作既定 API 描述。但从 Agent 记忆系统的实践看,可以这样理解这种设计思路:不同记忆承担不同工作。

一种可落地的划分方式是:

  • 事实记忆:用户明确告诉系统的稳定信息,例如姓名、公司、技术栈。
  • 偏好记忆:用户反复表达的工作习惯,例如“回答要短”“示例用 Python”。
  • 情节记忆:某次任务或会话中的事件,例如“上周排查过订单同步延迟”。

把这些内容混在一起会让检索变差。比如用户问“继续上次的排查”,Agent 需要的是情节记忆;用户说“以后都用中文回答”,这更像偏好记忆;用户告诉系统“我的集群运行在 EKS”,这更接近事实记忆。

可以这样实践:用 Elasticsearch 做一个最小记忆库

下面示例不是 Atlas 的官方接口,而是一个可改造的最小实现,用来演示“按用户隔离 + 三类记忆 + 语义检索”的基本形态。你需要本地或远程有一个 Elasticsearch 8.x 实例,并设置 ELASTICSEARCH_URL

python -m venv .venv
source .venv/bin/activate
pip install elasticsearch sentence-transformers
export ELASTICSEARCH_URL=http://localhost:9200

创建一个 agent_memory_demo.py

import os
from datetime import datetime, timezone
from elasticsearch import Elasticsearch
from sentence_transformers import SentenceTransformer

INDEX = "agent_memories"
USER_ID = "user_123"

es = Elasticsearch(os.environ.get("ELASTICSEARCH_URL", "http://localhost:9200"))
model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")


def ensure_index():
    if es.indices.exists(index=INDEX):
        return
    es.indices.create(
        index=INDEX,
        mappings={
            "properties": {
                "user_id": {"type": "keyword"},
                "memory_type": {"type": "keyword"},
                "content": {"type": "text"},
                "created_at": {"type": "date"},
                "embedding": {
                    "type": "dense_vector",
                    "dims": 384,
                    "index": True,
                    "similarity": "cosine"
                }
            }
        }
    )


def remember(memory_type, content):
    embedding = model.encode(content).tolist()
    es.index(
        index=INDEX,
        document={
            "user_id": USER_ID,
            "memory_type": memory_type,
            "content": content,
            "created_at": datetime.now(timezone.utc).isoformat(),
            "embedding": embedding
        }
    )


def recall(query, memory_type=None, size=5):
    query_vector = model.encode(query).tolist()
    filters = [{"term": {"user_id": USER_ID}}]
    if memory_type:
        filters.append({"term": {"memory_type": memory_type}})

    result = es.search(
        index=INDEX,
        size=size,
        query={
            "script_score": {
                "query": {"bool": {"filter": filters}},
                "script": {
                    "source": "cosineSimilarity(params.query_vector, 'embedding') + 1.0",
                    "params": {"query_vector": query_vector}
                }
            }
        }
    )
    return [hit["_source"] for hit in result["hits"]["hits"]]


if __name__ == "__main__":
    ensure_index()
    remember("fact", "用户的生产环境运行在 AWS EKS 上。")
    remember("preference", "用户喜欢简洁回答,并希望示例优先使用 Python。")
    remember("episode", "上次排查订单同步延迟时,发现瓶颈在消息队列消费者并发数。")

    for item in recall("继续排查订单同步问题", memory_type="episode"):
        print(f"[{item['memory_type']}] {item['content']}")

运行:

python agent_memory_demo.py

这个示例里最值得注意的是 user_id 过滤条件。它不是附属功能,而是 Agent 记忆系统的安全边界。没有这个过滤条件,检索层可能把其他用户的历史拿出来喂给模型,后果比普通搜索结果错配更严重。

MCP 让记忆系统更像“工具”,而不是框架插件

Atlas 通过 MCP 与 Agent 集成,这一点很重要。MCP 的价值在于把外部能力暴露成统一的工具接口:Agent 不需要知道底层是 Elasticsearch、关系数据库还是其他检索系统,只需要知道如何调用“写入记忆”和“检索记忆”。

一个可改造的 MCP 工具设计可以长这样:

{
  "tools": [
    {
      "name": "memory.remember",
      "description": "Store a user-scoped memory for future agent interactions",
      "input_schema": {
        "type": "object",
        "properties": {
          "user_id": {"type": "string"},
          "memory_type": {"type": "string", "enum": ["fact", "preference", "episode"]},
          "content": {"type": "string"}
        },
        "required": ["user_id", "memory_type", "content"]
      }
    },
    {
      "name": "memory.recall",
      "description": "Retrieve relevant memories for a user-scoped query",
      "input_schema": {
        "type": "object",
        "properties": {
          "user_id": {"type": "string"},
          "query": {"type": "string"},
          "memory_type": {"type": "string"},
          "limit": {"type": "integer", "default": 10}
        },
        "required": ["user_id", "query"]
      }
    }
  ]
}

真正上线时,user_id 不应该完全由模型自由传入,而应来自认证上下文或服务端会话。模型可以决定“要不要检索记忆”,但不应该决定“自己是谁”。

Recall@10 说明了什么,也没说明什么

Atlas 在问答能力评估中取得 0.89 Recall@10。这个指标可以粗略理解为:系统返回的前 10 条候选记忆中,包含相关答案的比例很高。

但采用时不要只看这个数字。Agent 记忆系统还要关注:

  • 写入质量:模型会不会把临时噪声写成长期记忆。
  • 更新策略:用户偏好变化后,旧记忆如何降权或失效。
  • 权限边界:多租户、企业团队、个人用户之间如何隔离。
  • 可解释性:Agent 引用记忆时,能否说明依据来自哪里。
  • 删除能力:用户要求“忘记我说过的 X”时,系统能否真正删除。

检索指标告诉你“找得到”,但生产系统还要证明“该找的才找、该忘的能忘”。

采用建议:先把记忆当成基础设施

如果你已经在做 Agent,Atlas 这类系统值得关注,因为它把记忆从 prompt 工程里拆出来,变成可检索、可隔离、可评估的基础设施。

建议从小处开始:先只保存明确的事实和偏好,不要把完整聊天记录都写入长期记忆;检索时强制带上用户边界;让 Agent 在使用记忆时暴露引用;再逐步加入情节记忆、过期策略和人工纠错入口。

Agent 的长期价值不只是会回答一次问题,而是能在安全边界内持续理解用户。Atlas 的开源,给这条路线提供了一个值得研究和实践的样本。


相关推荐