使用 EverOS 和 Milvus 构建 Agent 长期记忆
EverOS 是一个以 Markdown 为核心的 AI Agent 记忆系统。它从对话中提取持久记忆，以 Markdown 作为唯一事实来源，并构建可搜索的派生索引。
本教程将构建一个项目助手，使其能够跨不同对话记住发布决策。我们会添加有关 Project Atlas 发布的对话，以及与其他项目相关的无关对话。EverOS 使用 LLM 提取记忆，而 Milvus 则存储 Hybrid Search 所需的 BM25 索引和向量索引。
Conversations
|
v
EverOS + LLM ------> Markdown memory files
|
| embedding model
v
Milvus ------> BM25 + vector hybrid search
LLM 和 Embedding 模型承担不同的职责。LLM 将对话转换为结构化记忆，Embedding 模型则将这些记忆以及后续的搜索查询转换为向量。本教程中的基础 Hybrid Search 不需要 Reranker 模型。
Prerequisites
你需要：
- Python 3.12 或更高版本
uv- 正在运行的 Milvus Server
- OpenAI API key
本教程连接到位于 http://localhost:19530 的 Milvus Server。EverOS 还支持通过相同的 URI 和 token 设置连接到 Zilliz Cloud。其 Milvus 后端需要远程 endpoint，不接受 Milvus Lite 文件路径。
Install EverOS
创建本地项目，并安装 EverOS 及其可选的 Milvus 依赖项：
mkdir everos-milvus-demo
cd everos-milvus-demo
uv init --bare --python 3.12
uv add "everos[milvus]"
该命令有意不固定版本，因此全新安装时会解析并使用最新的兼容 EverOS 版本。
为本教程初始化一个独立的 memory 根目录：
export EVEROS_ROOT="$PWD/everos-data"
uv run everos init --root "$EVEROS_ROOT"
EverOS 会在此目录下创建 everos.toml 和 ome.toml，并将提取的记忆写入此处。
Configure OpenAI and Milvus
设置 OpenAI API key，并通过环境变量配置 EverOS：
export OPENAI_API_KEY="YOUR_OPENAI_API_KEY"
export MILVUS_URI="http://localhost:19530"
export EVEROS_INDEX__BACKEND="milvus"
export EVEROS_MILVUS__URI="$MILVUS_URI"
export EVEROS_MILVUS__COLLECTION_PREFIX="everos_bootcamp"
export EVEROS_LLM__MODEL="gpt-5.4-mini"
export EVEROS_LLM__API_KEY="$OPENAI_API_KEY"
export EVEROS_LLM__BASE_URL="https://api.openai.com/v1"
export EVEROS_EMBEDDING__MODEL="text-embedding-3-small"
export EVEROS_EMBEDDING__API_KEY="$OPENAI_API_KEY"
export EVEROS_EMBEDDING__BASE_URL="https://api.openai.com/v1"
export EVEROS_EMBEDDING__DIMENSIONS="1024"
export EVEROS_MEMORIZE__MODE="chat"
EverOS 使用 OpenAI 提取 memory 并生成 Embedding。text-embedding-3-small 默认返回 1536 维向量，但 EverOS 会将配置的 dimensions 值传递给 OpenAI。本教程请求生成 1024 维向量，以匹配 EverOS 管理的 Milvus Schema。
chat memory 模式使本示例专注于用户 memory。EverOS 会管理 Milvus Collection 及其 Schema，因此你无需自行创建。
启动 EverOS
启动 EverOS HTTP 服务器：
uv run everos server start --root "$EVEROS_ROOT"
保持此终端处于打开状态。EverOS 会在启动时连接到 Milvus，并使用配置的前缀创建七个派生索引 Collection。
在同一项目目录中打开另一个终端并检查服务：
curl http://127.0.0.1:8000/health
参考输出：
{
"status": "ok",
"version": "1.3.0",
"capabilities": {
"llm": true,
"embed": true,
"rerank": false,
"multimodal_llm": false,
"parser": true
},
"cascade": {
"healthy": true,
"pending": 0
}
}
响应中还包含其他健康状态字段。本教程需重点关注的值为 status: "ok"、llm: true、embed: true 和 cascade.healthy: true。
添加项目对话
以下 Python 程序会向 EverOS 发送十段相互独立的对话。其中，Atlas 项目的发布和回滚分别在不同的对话中讨论；另外八段关于其他项目的对话则作为干扰信息，让后续搜索必须识别出正确的项目记忆。
将以下代码保存为 add_memories.py：
import json
import time
from urllib.request import Request, urlopen
API_URL = "http://127.0.0.1:8000/api/v2/memory"
NOW = int(time.time() * 1000)
conversations = [
(
"atlas-release",
[
{
"sender_id": "maya",
"sender_name": "Maya",
"role": "user",
"timestamp": NOW,
"content": (
"For Project Atlas, we decided to launch with a 10% canary "
"on September 30. Promote to all users only after the checkout "
"error rate stays below 1% for 30 minutes."
),
},
{
"sender_id": "assistant",
"role": "assistant",
"timestamp": NOW + 1_000,
"content": (
"Understood. I will remember the Atlas launch date, canary "
"percentage, and promotion gate."
),
},
],
),
(
"atlas-rollback",
[
{
"sender_id": "maya",
"sender_name": "Maya",
"role": "user",
"timestamp": NOW + 10_000,
"content": (
"The Atlas rollback owner is Priya. Roll back immediately if "
"checkout errors exceed 2% for five minutes, and keep the "
"previous container image available for 24 hours."
),
},
{
"sender_id": "assistant",
"role": "assistant",
"timestamp": NOW + 11_000,
"content": (
"Got it. Priya owns rollback, with the 2% five-minute trigger "
"and a 24-hour image retention window."
),
},
],
),
(
"orion-pricing",
[
{
"sender_id": "maya",
"sender_name": "Maya",
"role": "user",
"timestamp": NOW + 20_000,
"content": (
"Project Orion will test annual billing with the education "
"segment. The pricing review is scheduled for October 12."
),
},
{
"sender_id": "assistant",
"role": "assistant",
"timestamp": NOW + 21_000,
"content": (
"I will remember Orion's annual billing experiment and October "
"pricing review."
),
},
],
),
(
"vega-mobile",
[
{
"sender_id": "maya",
"sender_name": "Maya",
"role": "user",
"timestamp": NOW + 30_000,
"content": (
"For Project Vega, the mobile team chose offline drafts as the "
"next milestone. Elena will review the interaction design on "
"October 18."
),
},
{
"sender_id": "assistant",
"role": "assistant",
"timestamp": NOW + 31_000,
"content": (
"Noted. Vega's next milestone is offline drafts, followed by "
"Elena's design review."
),
},
],
),
(
"nova-warehouse",
[
{
"sender_id": "maya",
"sender_name": "Maya",
"role": "user",
"timestamp": NOW + 40_000,
"content": (
"Project Nova will migrate the analytics warehouse to Iceberg. "
"Marcus owns the checksum rehearsal scheduled for October 22."
),
},
{
"sender_id": "assistant",
"role": "assistant",
"timestamp": NOW + 41_000,
"content": (
"I will remember Nova's warehouse migration and Marcus's "
"checksum rehearsal."
),
},
],
),
(
"helios-support",
[
{
"sender_id": "maya",
"sender_name": "Maya",
"role": "user",
"timestamp": NOW + 50_000,
"content": (
"Project Helios needs weekend support coverage for the APAC "
"region. Imani will publish the rotation schedule on November 1."
),
},
{
"sender_id": "assistant",
"role": "assistant",
"timestamp": NOW + 51_000,
"content": (
"Noted. Helios needs APAC weekend coverage, and Imani owns the "
"rotation schedule."
),
},
],
),
(
"luna-onboarding",
[
{
"sender_id": "maya",
"sender_name": "Maya",
"role": "user",
"timestamp": NOW + 60_000,
"content": (
"Project Luna will replace the onboarding tour with a checklist. "
"The localized copy is due from the content team on October 25."
),
},
{
"sender_id": "assistant",
"role": "assistant",
"timestamp": NOW + 61_000,
"content": (
"I will remember Luna's checklist approach and the localization "
"deadline."
),
},
],
),
(
"aurora-observability",
[
{
"sender_id": "maya",
"sender_name": "Maya",
"role": "user",
"timestamp": NOW + 70_000,
"content": (
"Project Aurora will retain detailed telemetry for 30 days. "
"The operations team should alert after three consecutive "
"heartbeat misses."
),
},
{
"sender_id": "assistant",
"role": "assistant",
"timestamp": NOW + 71_000,
"content": (
"Understood. Aurora keeps 30 days of telemetry and alerts after "
"three missed heartbeats."
),
},
],
),
(
"comet-invoices",
[
{
"sender_id": "maya",
"sender_name": "Maya",
"role": "user",
"timestamp": NOW + 80_000,
"content": (
"Project Comet will add downloadable invoice PDFs for enterprise "
"accounts. Finance will approve the tax-field layout on October 28."
),
},
{
"sender_id": "assistant",
"role": "assistant",
"timestamp": NOW + 81_000,
"content": (
"Noted. Comet covers enterprise invoice PDFs and an October tax "
"layout review."
),
},
],
),
(
"solstice-research",
[
{
"sender_id": "maya",
"sender_name": "Maya",
"role": "user",
"timestamp": NOW + 90_000,
"content": (
"Project Solstice is prototyping voice notes for field researchers. "
"The research team will interview 12 participants in November."
),
},
{
"sender_id": "assistant",
"role": "assistant",
"timestamp": NOW + 91_000,
"content": (
"I will remember Solstice's voice-note prototype and the planned "
"participant interviews."
),
},
],
),
]
def post(path, payload):
request = Request(
f"{API_URL}/{path}",
data=json.dumps(payload).encode(),
headers={"Content-Type": "application/json"},
method="POST",
)
with urlopen(request, timeout=300) as response:
return json.load(response)["data"]
for session_id, messages in conversations:
added = post(
"add",
{
"session_id": session_id,
"app_id": "project-assistant",
"project_id": "launch-planning",
"messages": messages,
"defer_extraction": True,
},
)
flushed = post(
"flush",
{
"session_id": session_id,
"app_id": "project-assistant",
"project_id": "launch-planning",
},
)
print(f"{session_id}: {added['status']} -> {flushed['status']}")
在项目目录中运行：
uv run python add_memories.py
参考输出：
atlas-release: accumulated -> extracted
atlas-rollback: accumulated -> extracted
orion-pricing: accumulated -> extracted
vega-mobile: accumulated -> extracted
nova-warehouse: accumulated -> extracted
helios-support: accumulated -> extracted
luna-onboarding: accumulated -> extracted
aurora-observability: accumulated -> extracted
comet-invoices: accumulated -> extracted
solstice-research: accumulated -> extracted
将 defer_extraction 设置为 true 后，每段对话都会存入持久化缓冲区，而不会要求 LLM 检测边界。随后调用 /flush 会标记该会话结束，并触发一次提取。接着，EverOS 会将提取出的 episode 写入 Markdown，并以异步方式为其生成 Embedding，以写入 Milvus Index。
Inspect the Markdown memory
生成的 episode 文件存储在应用、项目和用户作用域下：
find "$EVEROS_ROOT/project-assistant/launch-planning/users/maya/episodes" \
-type f -name "*.md"
参考输出如下（文件名中的日期取决于你运行示例的时间）：
everos-data/project-assistant/launch-planning/users/maya/episodes/episode-2026-09-08.md
打开该文件即可查看 LLM 提取的记忆。以下是经过删减的内容片段：
## ep_20260908_00000001
**owner_id**: maya
**session_id**: atlas-release
**sender_ids**: [maya, assistant]
### Subject
Maya's Project Atlas Launch Decision: September 30 Canary and Promotion Criteria
### Content
Maya decided that Project Atlas would launch with a 10% canary on September 30.
The promotion to all users would occur only after the checkout error rate remained
below 1% for 30 minutes.
由于记忆由 LLM 提取，具体措辞、标识符和 Timestamp 可能有所不同。原始 Markdown 文件始终是持久化的唯一事实来源；Milvus Index 可以根据这些文件重建。
搜索记忆
在 Atlas 上线前，使用 Hybrid Search 查询应记住哪些内容。将以下代码保存为 search_memories.py：
import json
import time
from urllib.request import Request, urlopen
URL = "http://127.0.0.1:8000/api/v2/memory/search"
payload = {
"user_id": "maya",
"app_id": "project-assistant",
"project_id": "launch-planning",
"query": "What should I remember before Atlas goes live?",
"method": "hybrid",
"top_k": 4,
}
def search():
request = Request(
URL,
data=json.dumps(payload).encode(),
headers={"Content-Type": "application/json"},
method="POST",
)
with urlopen(request, timeout=300) as response:
return json.load(response)["data"]["episodes"]
expected_sessions = {"atlas-release", "atlas-rollback"}
for _ in range(30):
episodes = search()
top_results = episodes[:2]
if {episode["session_id"] for episode in top_results} == expected_sessions:
break
time.sleep(2)
else:
raise RuntimeError("The expected Atlas memories were not indexed in time")
for rank, episode in enumerate(top_results, start=1):
print(f"{rank}. {episode['session_id']} | score={episode['score']:.3f}")
print(f" {episode['subject']}")
运行检索：
uv run python search_memories.py
参考输出（分数和措辞可能有所不同）：
1. atlas-release | score=0.492
Project Atlas Launch Plan: 10% Canary Rollout on September 30 with Error Rate Gate
2. atlas-rollback | score=0.400
Atlas Rollback Plan Details: Priya as Owner, 2% Error Trigger, 24-Hour Image Retention
返回结果中，两条 Atlas 对话排在八条无关对话之前。EverOS 将查询发送到 OpenAI Embedding 端点，让 Milvus 在 Maya 的应用和项目范围内检索 BM25 和向量候选结果，并融合这两组结果列表。
Inspect the Milvus Collection
EverOS 会为每种受支持的派生记忆类型创建一个 Collection。使用 MilvusClient 列出各 Collection 的行数：
import os
from pymilvus import MilvusClient
prefix = "everos_bootcamp"
client = MilvusClient(uri=os.environ.get("MILVUS_URI", "http://localhost:19530"))
memory_kinds = [
"agent_case",
"agent_skill",
"atomic_fact",
"episode",
"foresight",
"knowledge_topic",
"user_profile",
]
for kind in memory_kinds:
name = f"{prefix}_{kind}"
if client.has_collection(collection_name=name):
result = client.query(
collection_name=name,
filter="",
output_fields=["count(*)"],
)
print(f"{kind}: {result[0]['count(*)']} rows")
client.close()
以下是验证运行的参考输出：
agent_case: 0 rows
agent_skill: 0 rows
atomic_fact: 50 rows
episode: 10 rows
foresight: 0 rows
knowledge_topic: 0 rows
user_profile: 1 rows
原子事实的确切数量可能因 LLM 输出而异。十行情景记忆对应已 Flush 的十次对话。其他 Collection 用于此示例未涉及的 EverOS 记忆模式和功能。
使用其他 Milvus 部署
要使用其他 Milvus Server endpoint 或 Zilliz Cloud，请更新 EVEROS_MILVUS__URI。如果该 endpoint 需要身份验证，请设置 EVEROS_MILVUS__TOKEN。数据摄取和检索代码无需更改。
Conclusion
通过结合使用 EverOS 与 Milvus，你可以将对话转化为持久记忆，并通过关键词和语义信号检索这些记忆。你还可以调整这一模式，为面向你的用户、项目和工作流的助手及其他 AI Agent 应用提供长期记忆。