在本教程中，您将使用 crewai-mongodb-memory 包为 CrewAI 代理提供在 MongoDB Atlas 中持久存储的持久性长期内存。代理将用户偏好存储为内存记录，并使用 MongoDB 向量搜索从语义上回忆这些记录，因此，后续会话中的全新团队可以通过从上一个会话中检索学习内容来正确回答。
要学习；了解有关MongoDB CrewAI 集成的更多信息，请参阅将MongoDB与 CrewAI 集成。
先决条件
如要完成本教程，您必须具备以下条件：
Python 3.10 或更高版本。
以下MongoDB 集群类型之一：
使用Atlas CLI创建的本地Atlas部署。要学习；了解更多信息，请参阅创建本地Atlas部署。
安装了Search 和向量搜索的 MongoDB Community 集群。
Voyage AI API密钥。要创建API密钥，请参阅对API密钥建模。
大型语言模型 (LLM) API 密钥，例如 Anthropic 或 OpenAI，为代理的推理提供动力。
代理内存如何与 MongoDB 协作
crewai-mongodb-memory 后端实现 CrewAI 的 StorageBackend 协议。它将每个内存作为记录存储在 MongoDB 集合中，并使用 MongoDB 向量搜索以语义方式检索相关记录。后端使用以下组件：
MongoDBStorageBackend: 连接到 Atlas 集群，存储记忆记录，并运行$vectorSearch查询以回忆这些记录。MemoryRecord: 表示单个内存，包括其文本内容、层次结构范围（如/users/alex/preferences）、可选类别及其向量嵌入。embed_text(): 使用 Voyage AIvoyage-4模型为存储的文档和查询生成 1024 维嵌入。
由于内存存在 Atlas 中，而不是团队的进程上下文中，因此连接到同一集群和范围的任何团队都可以回忆存储的内存。
{ "_id": "cff7db8d-aa2e-4d82-8234-e4b8f1ccccb6", "content": "The user always prefers window seats on flights.", "scope": "/users/alex/preferences", "categories": [ "preference" ], "metadata": {}, "importance": 0.5, "created_at": { "$date": "2026-07-14T22:06:59.162Z" }, "last_accessed": { "$date": "2026-07-14T22:06:59.162Z" }, "source": null, "private": false, "embedding": [ 0.03147803992033005, 0.015906454995274544, -0.007409059442579746, 0.028631621971726418, ... ], "scope_ancestors": [ "/users", "/users/alex", "/users/alex/preferences" ] }
构建并运行代理
完成以下步骤以构建和运行存储和回忆用户偏好设置的代理：
1
设置环境。
在终端中运行以下命令，创建名为
crewai-memory-project的新目录并安装所需的依赖项：mkdir crewai-memory-project cd crewai-memory-project pip install crewai-mongodb-memory crewai "crewai[google-genai]" voyageai python-dotenv 在您的项目中，创建一个
.env文件并添加以下行：ATLAS_URI="<connection-string>" VOYAGE_API_KEY="<voyage-api-key>" ANTHROPIC_API_KEY="<anthropic-api-key>" MODEL="<anthropic-model>" 注意
将
<connection-string>替换为您的 Atlas 集群或本地部署的连接字符串。连接字符串应使用以下格式：
mongodb+srv://<db_username>:<db_password>@<clusterName>.<hostname>.mongodb.net 要学习；了解更多信息，请参阅通过客户端库连接到集群。
2
构建代理。
在项目中创建名为 main.py 的文件并粘贴以下代码：
from __future__ import annotations import os import sys import time import warnings from dotenv import load_dotenv load_dotenv() warnings.filterwarnings("ignore") from crewai import Agent, Crew, Task from crewai.tools import tool from crewai_mongodb_memory import MemoryRecord, MongoDBStorageBackend, embed_text from crewai.events import CrewKickoffStartedEvent, CrewKickoffCompletedEvent, AgentExecutionCompletedEvent from crewai.events import BaseEventListener DEMO_DB = "crewai_mem_agent_demo" SCOPE = "/users/alex/preferences" MODEL = os.environ.get("ANTHROPIC_MODEL", "anthropic/claude-3-5-sonnet-latest") EMBEDDING_MODEL = os.environ.get("VOYAGE_MODEL", "voyage-4") # Module-level backend handle so the function tools can reach it. _BACKEND: MongoDBStorageBackend | None = None _SESSION_LABEL = "" # Custom event listener for logging Crew and Agent events. class EventListener(BaseEventListener): def __init__(self): super().__init__() def setup_listeners(self, crewai_event_bus): def on_crew_started(source, event): print(f"Crew '{event.crew_name}' has started execution!") def on_crew_completed(source, event): print(f"Crew '{event.crew_name}' has completed execution!") print(f"Output: {event.output}") def on_agent_execution_completed(source, event): print(f"Agent '{event.agent.role}' completed task") print(f"Output: {event.output}") def banner(title: str) -> None: print(f"\n{'=' * 70}\n{title}\n{'=' * 70}") def embed(text: str, input_type: str) -> list[float]: """Generate embeddings with an explicit Voyage model.""" try: return embed_text(text, input_type=input_type, model=EMBEDDING_MODEL) except TypeError: # Backward compatibility for older crewai-mongodb-memory versions. return embed_text(text, input_type=input_type) def remember_preference(fact: str) -> str: """Store a durable user preference in MongoDB Atlas long-term memory.""" assert _BACKEND is not None rec = MemoryRecord( content=fact, scope=SCOPE, categories=["preference"], embedding=embed(fact, input_type="document"), ) _BACKEND.save([rec]) return f"Stored preference: {fact}" def recall_preferences(query: str) -> str: """Retrieve relevant user preferences from Atlas via $vectorSearch.""" assert _BACKEND is not None qv = embed(query, input_type="query") hits = _BACKEND.search(qv, scope_prefix=SCOPE, limit=3) if not hits: return "No relevant preferences found." for rec, score in hits: print(f" - [{score:.3f}] {rec.content}") return "\n".join(f"- {rec.content} (score={score:.3f})" for rec, score in hits) def build_agent() -> Agent: return Agent( role="Personal Concierge", goal="Help the user with durable long-term memory stored in MongoDB Atlas.", backstory=( """You remember user preferences across sessions. When the user shares a durable preference, call remember_preference. When a request may depend on what you know about them, call recall_preferences first and use the results.""" ), tools=[remember_preference, recall_preferences], llm=MODEL, verbose=True, ) def run_task(description: str, expected_output: str) -> str: """Run a single-task crew (a fresh crew each call is a fresh session).""" agent = build_agent() task = Task(description=description, expected_output=expected_output, agent=agent) crew = Crew(agent, [task]) return str(crew.kickoff()) def main() -> None: global _BACKEND, _SESSION_LABEL uri = os.environ.get("ATLAS_URI") # Fast fail if the required environment variables are not set. if not all([uri, os.environ.get("ANTHROPIC_API_KEY"), os.environ.get("VOYAGE_API_KEY")]): print("This script needs ATLAS_URI, ANTHROPIC_API_KEY, and VOYAGE_API_KEY.") sys.exit(1) print(f"=== CrewAI and MongoDB Atlas long-term memory (model={MODEL}) ===") _BACKEND = MongoDBStorageBackend(uri, database_name=DEMO_DB) _BACKEND.delete(scope_prefix=SCOPE) # clean slate for a repeatable run print("Ensuring Atlas Vector Search index (first build can take ~1 min)...") if not _BACKEND.ensure_vector_index(wait=True): print("Vector index did not become queryable in time.") sys.exit(1) print("Index queryable.") # Initialize the event listener for the agent. listener = EventListener() # Session 1: the agent learns and stores preferences. banner("SESSION 1 - agent stores durable preferences in Atlas") session_1_input = ( """The user says: 'I'm vegetarian, I avoid dairy, and I always prefer window seats on flights.' Store each durable preference.""" ) _SESSION_LABEL = "SESSION 1" print(f" {session_1_input}") out1 = run_task( description=session_1_input, expected_output="A short confirmation of what was stored.", ) print(f"\nAgent (session 1): {out1}") # Atlas indexing is asynchronous, so wait until the facts are searchable. print("\nWaiting for Atlas to index the new memories...") for _ in range(20): if _BACKEND.search(embed("food", input_type="query"), scope_prefix=SCOPE, limit=1): break time.sleep(2) print("Memories searchable.") # Session 2: a brand-new crew with no shared state recalls from Atlas. banner("SESSION 2 - fresh crew answers via recall_preferences") session_2_input = ( """The user is booking a long flight and pre-ordering an in-flight meal. Recommend a seat and a meal that fit their preferences.""" ) _SESSION_LABEL = "SESSION 2" print(f" {session_2_input}") out2 = run_task( description=session_2_input, expected_output="A seat and meal recommendation grounded in recalled preferences.", ) print(f"\nAgent (session 2, brand-new crew): {out2}") banner("Proof: direct $vectorSearch recall over stored preferences") hits = _BACKEND.search( embed("dietary and seating preferences", input_type="query"), scope_prefix=SCOPE, limit=5, ) for rec, score in hits: print(f" - [{score:.3f}] {rec.content}") _BACKEND.delete(scope_prefix=SCOPE) _BACKEND.close() print("\nDone - CrewAI stored and recalled preferences via MongoDB Vector Search.") if __name__ == "__main__": main()
此脚本执行以下操作：
3
运行文件。
运行以下命令以执行脚本：
python main.py
====================================================================== SESSION 1 - agent stores durable preferences in Atlas ====================================================================== The user says: 'I'm vegetarian, I avoid dairy, and I always prefer window seats on flights.' Store each durable preference. Crew 'crew' has started execution! ╭────────────────────────────── 🤖 Agent Started ──────────────────────────────╮ │ │ │ Agent: Personal Concierge │ │ │ │ Task: The user says: 'I'm vegetarian, I avoid dairy, and I always prefer │ │ window seats on flights.' Store each durable preference. │ │ │ ╰──────────────────────────────────────────────────────────────────────────────╯ Tool remember_preference executed with result: Stored preference: The user is vegetarian.... Tool remember_preference executed with result: Stored preference: The user avoids dairy.... Tool remember_preference executed with result: Stored preference: The user always prefers window seats on flights.... [Finalize] todos_count=0, todos_with_results=0 ╭─────────────────────────── ✅ Agent Final Answer ────────────────────────────╮ │ │ │ Agent: Personal Concierge │ │ │ │ Final Answer: │ │ Stored: │ │ - The user is vegetarian. │ │ - The user avoids dairy. │ │ - The user always prefers window seats on flights. │ │ │ ╰──────────────────────────────────────────────────────────────────────────────╯ Agent 'Personal Concierge' completed task Output: Stored: - The user is vegetarian. - The user avoids dairy. - The user always prefers window seats on flights. Crew 'crew' has completed execution! Output: Stored: - The user is vegetarian. - The user avoids dairy. - The user always prefers window seats on flights. Agent (session 1): Stored: - The user is vegetarian. - The user avoids dairy. - The user always prefers window seats on flights. Waiting for Atlas to index the new memories... Memories searchable. ====================================================================== SESSION 2 - fresh crew answers via recall_preferences ====================================================================== The user is booking a long flight and pre-ordering an in-flight meal. Recommend a seat and a meal that fit their preferences. Crew 'crew' has started execution! ╭────────────────────────────── 🤖 Agent Started ──────────────────────────────╮ │ │ │ Agent: Personal Concierge │ │ │ │ Task: The user is booking a long flight and pre-ordering an in-flight │ │ meal. │ │ Recommend a seat and a meal that fit their preferences. │ │ │ ╰──────────────────────────────────────────────────────────────────────────────╯ - [0.794] The user always prefers window seats on flights. - [0.748] The user is vegetarian. - [0.739] The user avoids dairy. Tool recall_preferences executed with result: - The user always prefers window seats on flights. (score=0.794) - The user is vegetarian. (score=0.748) - The user avoids dairy. (score=0.739)... [Finalize] todos_count=0, todos_with_results=0 ╭─────────────────────────── ✅ Agent Final Answer ────────────────────────────╮ │ │ │ Agent: Personal Concierge │ │ │ │ Final Answer: │ │ I recommend booking a **window seat** for the long flight, since you │ │ prefer window seats. │ │ │ │ For your in-flight meal, choose a **vegetarian dairy-free meal**. If the │ │ airline offers specific meal codes, look for something like **VGML / vegan │ │ meal**, since that will typically meet both preferences: vegetarian and no │ │ dairy. │ │ │ ╰──────────────────────────────────────────────────────────────────────────────╯ Agent 'Personal Concierge' completed task Output: I recommend booking a **window seat** for the long flight, since you prefer window seats. For your in-flight meal, choose a **vegetarian dairy-free meal**. If the airline offers specific meal codes, look for something like **VGML / vegan meal**, since that will typically meet both preferences: vegetarian and no dairy. Crew 'crew' has completed execution! Output: I recommend booking a **window seat** for the long flight, since you prefer window seats. For your in-flight meal, choose a **vegetarian dairy-free meal**. If the airline offers specific meal codes, look for something like **VGML / vegan meal**, since that will typically meet both preferences: vegetarian and no dairy. Agent (session 2, brand-new crew): I recommend booking a **window seat** for the long flight, since you prefer window seats. For your in-flight meal, choose a **vegetarian dairy-free meal**. If the airline offers specific meal codes, look for something like **VGML / vegan meal**, since that will typically meet both preferences: vegetarian and no dairy. ====================================================================== Proof: direct $vectorSearch recall over stored preferences ====================================================================== - [0.770] The user always prefers window seats on flights. - [0.768] The user is vegetarian. - [0.754] The user avoids dairy. Done - CrewAI stored and recalled preferences via MongoDB Vector Search.
注意
您的输出可能会有所不同，因为 LLM 响应和向量搜索分数在不同的运行之间可能会有所不同。