​
CrewAI 中的内存系统简介

CrewAI 框架引入了一个复杂的内存系统，旨在显著增强 AI 代理的能力。该系统包括短期记忆、长期记忆、实体记忆和上下文记忆，每种内存都服务于独特目的，帮助代理从过去的交互中记住、推理和学习。

​
内存系统组件

组件描述
短期记忆使用 RAG 暂时存储最近的交互和结果，使代理能够在当前执行期间回忆和利用与其当前上下文相关的信息。
长期记忆保留从过去执行中获得的宝贵见解和学习，使代理能够随着时间建立和完善知识。
实体记忆捕获和组织在任务中遇到的实体（人物、地点、概念）的信息，促进更深层次的理解和关系映射。使用 RAG 存储实体信息。
上下文记忆通过结合短期记忆、长期记忆和实体记忆来维持交互的上下文，有助于在一系列任务或对话中保持代理响应的连贯性和相关性。
外部记忆启用与外部内存系统和提供商（如 Mem0）的集成，允许在不同应用中进行专门的内存存储和检索。支持自定义存储实现以实现灵活的内存管理。
用户记忆⚠️ 已弃用：此组件已弃用，并将在未来的版本中移除。请改为使用外部记忆。

​
内存系统如何增强代理能力

  1. 上下文感知能力：通过短期记忆和上下文记忆，代理能够在一个对话或任务序列中保持上下文，从而产生更连贯和相关的响应。

  2. 经验积累：长期记忆使代理能够积累经验，从过去的行动中学习，以改进未来的决策和问题解决能力。

  3. 实体理解：通过维护实体记忆，代理可以识别并记住关键实体，增强其处理和交互复杂信息的能力。

​
在您的团队中实现内存功能

配置团队时，您可以启用和自定义每个内存组件，以适应团队的目标及其将执行的任务的性质。默认情况下，内存系统处于禁用状态，您可以通过在团队配置中设置 memory=True 来确保其激活。内存默认使用 OpenAI 嵌入，但您可以通过将 embedder 设置为不同的模型来更改它。您也可以使用自己的实例初始化内存。

“嵌入器”(embedder) 仅应用于使用 Chroma 进行 RAG 的短期记忆。长期记忆使用 SQLite3 存储任务结果。目前，无法覆盖这些存储实现。数据存储文件保存在使用 appdirs 包查找的平台特定位置，并且可以使用 CREWAI_STORAGE_DIR 环境变量覆盖项目名称。

​
示例：为团队配置内存

代码
from crewai import Crew, Agent, Task, Process

# Assemble your crew with memory capabilities
my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True
)

​
示例：使用自定义内存实例，例如使用 FAISS 作为向量数据库

代码
from crewai import Crew, Process
from crewai.memory import LongTermMemory, ShortTermMemory, EntityMemory
from crewai.memory.storage.rag_storage import RAGStorage
from crewai.memory.storage.ltm_sqlite_storage import LTMSQLiteStorage
from typing import List, Optional

# Assemble your crew with memory capabilities
my_crew: Crew = Crew(
    agents = [...],
    tasks = [...],
    process = Process.sequential,
    memory = True,
    # Long-term memory for persistent storage across sessions
    long_term_memory = LongTermMemory(
        storage=LTMSQLiteStorage(
            db_path="/my_crew1/long_term_memory_storage.db"
        )
    ),
    # Short-term memory for current context using RAG
    short_term_memory = ShortTermMemory(
        storage = RAGStorage(
                embedder_config={
                    "provider": "openai",
                    "config": {
                        "model": 'text-embedding-3-small'
                    }
                },
                type="short_term",
                path="/my_crew1/"
            )
        ),
    ),
    # Entity memory for tracking key information about entities
    entity_memory = EntityMemory(
        storage=RAGStorage(
            embedder_config={
                "provider": "openai",
                "config": {
                    "model": 'text-embedding-3-small'
                }
            },
            type="short_term",
            path="/my_crew1/"
        )
    ),
    verbose=True,
)

​
安全注意事项

配置内存存储时

  • 使用环境变量来设置存储路径（例如，CREWAI_STORAGE_DIR）
  • 切勿硬编码敏感信息，如数据库凭据
  • 考虑存储目录的访问权限
  • 尽可能使用相对路径以保持可移植性

使用环境变量的示例

import os
from crewai import Crew
from crewai.memory import LongTermMemory
from crewai.memory.storage.ltm_sqlite_storage import LTMSQLiteStorage

# Configure storage path using environment variable
storage_path = os.getenv("CREWAI_STORAGE_DIR", "./storage")
crew = Crew(
    memory=True,
    long_term_memory=LongTermMemory(
        storage=LTMSQLiteStorage(
            db_path="{storage_path}/memory.db".format(storage_path=storage_path)
        )
    )
)

​
配置示例

​
基本内存配置

from crewai import Crew
from crewai.memory import LongTermMemory

# Simple memory configuration
crew = Crew(memory=True)  # Uses default storage locations

请注意，当设置 memory=True 时，外部记忆不会被定义，因为我们无法推断哪种外部记忆适合您的情况。

​
自定义存储配置

from crewai import Crew
from crewai.memory import LongTermMemory
from crewai.memory.storage.ltm_sqlite_storage import LTMSQLiteStorage

# Configure custom storage paths
crew = Crew(
    memory=True,
    long_term_memory=LongTermMemory(
        storage=LTMSQLiteStorage(db_path="./memory.db")
    )
)

​
集成 Mem0 以增强用户记忆

Mem0 是一个用于 LLM 应用程序的自改进内存层，可实现个性化 AI 体验。

​
使用 Mem0 API 平台

要包含用户特定的内存，您可以在此处获取您的 API 密钥，并参阅文档了解如何添加用户偏好。在这种情况下，user_memory 被设置为 mem0 中的 MemoryClient。

代码
import os
from crewai import Crew, Process
from mem0 import MemoryClient

# Set environment variables for Mem0
os.environ["MEM0_API_KEY"] = "m0-xx"

# Step 1: Create a Crew with User Memory

crew = Crew(
    agents=[...],
    tasks=[...],
    verbose=True,
    process=Process.sequential,
    memory=True,
    memory_config={
        "provider": "mem0",
        "config": {"user_id": "john"},
        "user_memory" : {} #Set user_memory explicitly to a dictionary, we are working on this issue.
    },
)

​
附加内存配置选项

如果您想访问特定的组织和项目，可以在内存配置中设置 org_id 和 project_id 参数。

代码
from crewai import Crew

crew = Crew(
    agents=[...],
    tasks=[...],
    verbose=True,
    memory=True,
    memory_config={
        "provider": "mem0",
        "config": {"user_id": "john", "org_id": "my_org_id", "project_id": "my_project_id"},
        "user_memory" : {} #Set user_memory explicitly to a dictionary, we are working on this issue.
    },
)

​
使用本地 Mem0 内存

如果您想使用自定义配置的本地 mem0 内存，可以在配置中设置参数 local_mem0_config。如果同时设置了操作系统环境变量密钥和 local_mem0_config，则 API 平台优先于本地配置。有关更多理解，请查看此 mem0 本地配置文档。在这种情况下，user_memory 被设置为 mem0 中的 Memory。

代码
from crewai import Crew


#local mem0 config
config = {
    "vector_store": {
        "provider": "qdrant",
        "config": {
            "host": "localhost",
            "port": 6333
        }
    },
    "llm": {
        "provider": "openai",
        "config": {
            "api_key": "your-api-key",
            "model": "gpt-4"
        }
    },
    "embedder": {
        "provider": "openai",
        "config": {
            "api_key": "your-api-key",
            "model": "text-embedding-3-small"
        }
    },
    "graph_store": {
        "provider": "neo4j",
        "config": {
            "url": "neo4j+s://your-instance",
            "username": "neo4j",
            "password": "password"
        }
    },
    "history_db_path": "/path/to/history.db",
    "version": "v1.1",
    "custom_fact_extraction_prompt": "Optional custom prompt for fact extraction for memory",
    "custom_update_memory_prompt": "Optional custom prompt for update memory"
}

crew = Crew(
    agents=[...],
    tasks=[...],
    verbose=True,
    memory=True,
    memory_config={
        "provider": "mem0",
        "config": {"user_id": "john", 'local_mem0_config': config},
        "user_memory" : {} #Set user_memory explicitly to a dictionary, we are working on this issue.
    },
)

​
使用外部记忆

外部记忆是一项强大的功能，允许您将外部内存系统与 CrewAI 应用程序集成。当您想使用专门的内存提供商或在不同应用程序中维护内存时，这尤其有用。由于它是外部记忆，我们无法为其添加默认值 - 不像长期记忆和短期记忆那样。

​
与 Mem0 的基本用法

使用外部记忆的最常见方式是将其作为 Mem0 提供商。

import os
from crewai import Agent, Crew, Process, Task
from crewai.memory.external.external_memory import ExternalMemory

os.environ["MEM0_API_KEY"] = "YOUR-API-KEY"

agent = Agent(
    role="You are a helpful assistant",
    goal="Plan a vacation for the user",
    backstory="You are a helpful assistant that can plan a vacation for the user",
    verbose=True,
)
task = Task(
    description="Give things related to the user's vacation",
    expected_output="A plan for the vacation",
    agent=agent,
)

crew = Crew(
    agents=[agent],
    tasks=[task],
    verbose=True,
    process=Process.sequential,
    external_memory=ExternalMemory(
        embedder_config={"provider": "mem0", "config": {"user_id": "U-123"}} # you can provide an entire Mem0 configuration
    ),
)

crew.kickoff(
    inputs={"question": "which destination is better for a beach vacation?"}
)

​
使用自定义存储的外部记忆

您还可以为外部记忆创建自定义存储实现。这是一个如何创建自定义存储的示例：

from crewai import Agent, Crew, Process, Task
from crewai.memory.external.external_memory import ExternalMemory
from crewai.memory.storage.interface import Storage


class CustomStorage(Storage):
    def __init__(self):
        self.memories = []

    def save(self, value, metadata=None, agent=None):
        self.memories.append({"value": value, "metadata": metadata, "agent": agent})

    def search(self, query, limit=10, score_threshold=0.5):
        # Implement your search logic here
        return []

    def reset(self):
        self.memories = []


# Create external memory with custom storage
external_memory = ExternalMemory(
    storage=CustomStorage(),
    embedder_config={"provider": "mem0", "config": {"user_id": "U-123"}},
)

agent = Agent(
    role="You are a helpful assistant",
    goal="Plan a vacation for the user",
    backstory="You are a helpful assistant that can plan a vacation for the user",
    verbose=True,
)
task = Task(
    description="Give things related to the user's vacation",
    expected_output="A plan for the vacation",
    agent=agent,
)

crew = Crew(
    agents=[agent],
    tasks=[task],
    verbose=True,
    process=Process.sequential,
    external_memory=external_memory,
)

crew.kickoff(
    inputs={"question": "which destination is better for a beach vacation?"}
)

​
附加嵌入提供商

​
使用 OpenAI 嵌入（已是默认设置）

代码
from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "openai",
        "config": {
            "model": 'text-embedding-3-small'
        }
    }
)

另外，您可以直接将 OpenAIEmbeddingFunction 传递给 embedder 参数。

示例

代码
from crewai import Crew, Agent, Task, Process
from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "openai",
        "config": {
            "model": 'text-embedding-3-small'
        }
    }
)

​
使用 Ollama 嵌入

代码
from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "ollama",
        "config": {
            "model": "mxbai-embed-large"
        }
    }
)

​
使用 Google AI 嵌入

​
先决条件

在使用 Google AI 嵌入之前，请确保您具备：

  • Gemini API 的访问权限
  • 必要的 API 密钥和权限

您需要更新您的 pyproject.toml 依赖项

dependencies = [
    "google-generativeai>=0.8.4", #main version in January/2025 - crewai v.0.100.0 and crewai-tools 0.33.0
    "crewai[tools]>=0.100.0,<1.0.0"
]
代码
from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "google",
        "config": {
            "api_key": "<YOUR_API_KEY>",
            "model": "<model_name>"
        }
    }
)

​
使用 Azure OpenAI 嵌入

代码
from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction
from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "openai",
        "config": {
            "api_key": "YOUR_API_KEY",
            "api_base": "YOUR_API_BASE_PATH",
            "api_version": "YOUR_API_VERSION",
            "model_name": 'text-embedding-3-small'
        }
    }
)

​
使用 Vertex AI 嵌入

代码
from chromadb.utils.embedding_functions import GoogleVertexEmbeddingFunction
from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "vertexai",
        "config": {
            "project_id"="YOUR_PROJECT_ID",
            "region"="YOUR_REGION",
            "api_key"="YOUR_API_KEY",
            "model_name"="textembedding-gecko"
        }
    }
)

​
使用 Cohere 嵌入

代码
from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "cohere",
        "config": {
            "api_key": "YOUR_API_KEY",
            "model": "<model_name>"
        }
    }
)

​
使用 VoyageAI 嵌入

代码
from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "voyageai",
        "config": {
            "api_key": "YOUR_API_KEY",
            "model": "<model_name>"
        }
    }
)

​
使用 HuggingFace 嵌入

代码
from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "huggingface",
        "config": {
            "api_url": "<api_url>",
        }
    }
)

​
使用 Watson 嵌入

代码
from crewai import Crew, Agent, Task, Process

# Note: Ensure you have installed and imported `ibm_watsonx_ai` for Watson embeddings to work.

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "watson",
        "config": {
            "model": "<model_name>",
            "api_url": "<api_url>",
            "api_key": "<YOUR_API_KEY>",
            "project_id": "<YOUR_PROJECT_ID>",
        }
    }
)

​
使用 Amazon Bedrock 嵌入

代码
# Note: Ensure you have installed `boto3` for Bedrock embeddings to work.

import os
import boto3
from crewai import Crew, Agent, Task, Process

boto3_session = boto3.Session(
    region_name=os.environ.get("AWS_REGION_NAME"),
    aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
    aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY")
)

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    embedder={
    "provider": "bedrock",
        "config":{
            "session": boto3_session,
            "model": "amazon.titan-embed-text-v2:0",
            "vector_dimension": 1024
        }
    }
    verbose=True
)

​
添加自定义嵌入函数

代码
from crewai import Crew, Agent, Task, Process
from chromadb import Documents, EmbeddingFunction, Embeddings

# Create a custom embedding function
class CustomEmbedder(EmbeddingFunction):
    def __call__(self, input: Documents) -> Embeddings:
        # generate embeddings
        return [1, 2, 3] # this is a dummy embedding

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "custom",
        "config": {
            "embedder": CustomEmbedder()
        }
    }
)

​
通过 CLI 重置内存

crewai reset-memories [OPTIONS]

​
重置内存选项

选项描述类型默认值
-l, --long重置长期记忆。标志（布尔）False
-s, --short重置短期记忆。标志（布尔）False
-e, --entities重置实体记忆。标志（布尔）False
-k, --kickoff-outputs重置最新启动的任务输出。标志（布尔）False
-kn, --knowledge重置知识存储。标志（布尔）False
-akn, --agent-knowledge重置代理知识存储。标志（布尔）False
-a, --all重置所有内存。标志（布尔）False

注意：要使用 CLI 命令，您需要在同一目录下创建一个名为 crew.py 的文件，其中包含您的团队定义。

​
通过 crew 对象重置内存


my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "custom",
        "config": {
            "embedder": CustomEmbedder()
        }
    }
)

my_crew.reset_memories(command_type = 'all') # Resets all the memory

​
重置内存选项

命令类型描述
long重置长期记忆。
short重置短期记忆。
entities重置实体记忆。
kickoff_outputs重置最新启动的任务输出。
knowledge重置知识记忆。
agent_knowledge重置代理知识记忆。
all重置所有内存。

​
使用 CrewAI 内存系统的好处

  • 🦾 自适应学习：团队随着时间的推移变得更高效，适应新信息并完善其任务处理方法。
  • 🫡 增强的个性化：内存使代理能够记住用户偏好和历史交互，从而提供个性化体验。
  • 🧠 改进的问题解决：访问丰富的内存存储有助于代理做出更明智的决策，借鉴过去的学习和上下文洞察。

​
结论

将 CrewAI 的内存系统集成到您的项目中非常简单。通过利用提供的内存组件和配置，您可以快速赋予您的代理记忆、推理和从交互中学习的能力，从而解锁新的智能和能力水平。