利用 EverOS 和 Milvus 构建长期智能体记忆
EverOS是一个面向 AI Agents 的“Markdown 优先”记忆系统。它能从对话中提取持久的记忆，将 Markdown 作为权威数据源，并构建可搜索的衍生索引。
在本教程中，我们将构建一个项目助手，使其能够记住不同对话中的发布决策。我们将添加关于“Project Atlas”发布的对话，以及与其他项目相关的无关对话。EverOS将使用大型语言模型（LLM）提取记忆，而Milvus则负责存储用于混合搜索的BM25和向量索引。
Conversations
|
v
EverOS + LLM ------> Markdown memory files
|
| embedding model
v
Milvus ------> BM25 + vector hybrid search
大语言模型（LLM）和嵌入模型各司其职。大语言模型将对话转化为结构化记忆；嵌入模型则将这些记忆及后续的搜索查询转换为向量。本教程中的基础混合搜索无需使用重新排序模型。
先决条件
您需要：
- Python 3.12 或更高版本
uv- 正在运行的Milvus 服务器
- 一个OpenAI API 密钥
本教程将连接至http://localhost:19530 上的 Milvus 服务器。EverOS 还支持通过相同的 URI 和令牌设置连接至Zilliz Cloud。其 Milvus 后端需要远程端点，不接受 Milvus Lite 文件路径。
安装 EverOS
创建本地项目并安装 EverOS 及其可选的 Milvus 依赖项：
mkdir everos-milvus-demo
cd everos-milvus-demo
uv init --bare --python 3.12
uv add "everos[milvus]"
该命令特意未指定版本，因此新安装时会自动获取最新的兼容 EverOS 版本。
为本教程初始化一个独立的内存根目录：
export EVEROS_ROOT="$PWD/everos-data"
uv run everos init --root "$EVEROS_ROOT"
EverOS 会在该目录下创建everos.toml 和ome.toml 文件，并在此处写入提取的内存数据。
配置 OpenAI 和 Milvus
通过环境变量设置 OpenAI API 密钥并配置 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 进行记忆提取和 Embeddings 处理。text-embedding-3-small 默认返回1536 维度的结果，但 EverOS 会将配置的dimensions 值转发给 OpenAI。本教程请求1024 维度的结果，以匹配由 EverOS 管理的 Milvus Schema。
chat 内存模式使本示例专注于用户内存。EverOS 负责管理 Milvus Collections 和其 Schemas，因此您无需自行创建。
启动 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将提取的片段写入Markdown格式，并异步将其嵌入到Milvus索引中。
检查 Markdown 缓存
生成的对话片段文件存储在应用、项目和用户范围下：
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 提取，具体措辞、标识符和时间戳可能会有所不同。原始 Markdown 文件仍是可靠的权威来源；Milvus 索引可基于这些文件重建。
搜索记忆
在 Atlas 正式上线前，使用混合搜索功能来确定应记录哪些内容。将以下代码保存为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 Embeddings 端点，向 Milvus 请求 Maya 的应用程序和项目范围内的 BM25 及向量候选项，并融合这两组结果列表。
检查 Milvus Collections
EverOS 为每种受支持的派生内存类型创建一个 Collection。使用MilvusClient 列出其行数：
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 输出而异。这十行记录对应于已刷出的十段对话。其他 Collections 合适用于本示例未涉及的 EverOS 内存模式和功能。
使用另一个 Milvus 部署
若要使用其他 Milvus Server 端点或Zilliz Cloud，请更新EVEROS_MILVUS__URI 。当端点需要身份验证时，请设置EVEROS_MILVUS__TOKEN 。数据摄入和搜索代码保持不变。
结论
通过将 EverOS 与 Milvus 结合使用，您可以将对话转化为持久的记忆，并通过关键词和语义信号进行检索。您可以采用相同的模式，为助手和其他代理类应用程序提供针对您自身用户、项目和工作流的长期记忆。
