Add long-term memory to CrewAI agents and crews.

CrewAI Integration

CortexDB gives a CrewAI crew shared long-term memory — agents in the crew read from and write to the same scope, so context accumulated by one task is available to every later task.

Install

pip install cortexdbai crewai

Wire it in via task callbacks

CrewAI exposes per-task callback hooks. Use them to capture inputs/outputs into CortexDB and recall context before execution.

import os
from datetime import datetime, timezone
from uuid import uuid4
from crewai import Agent, Task, Crew, Process
from cortexdb.v1 import V1Client

client = V1Client(
    api_url="https://api-v1.cortexdb.ai",
    actor="agent:crew_runner",
    bearer=os.environ["CORTEX_TOKEN"],
)
SCOPE = "org:acme/crew:research-team"


def with_memory(description: str) -> str:
    """Prepend CortexDB recall context to a task description."""
    pack = client.recall(
        scope=SCOPE, view="holistic", query=description,
        include=["beliefs", "facts", "episodes"],
        budgets={"max_tokens": 3000},
    )
    if pack["context_block"]:
        return f"PRIOR CONTEXT:\n{pack['context_block']}\n\n---\n\n{description}"
    return description


def capture_result(task_output) -> None:
    client.experience(
        scope=SCOPE,
        text=str(task_output),
        modality="tool_result",
        observed_at=datetime.now(timezone.utc).isoformat(),
        idempotency_key=f"crew-{uuid4()}",
    )


researcher = Agent(
    role="Senior Researcher",
    goal="Find relevant technical context",
    backstory="Deep institutional knowledge of the team's prior decisions.",
)

writer = Agent(
    role="Technical Writer",
    goal="Produce clear documentation.",
    backstory="Skilled at turning notes into shippable docs.",
)

research = Task(
    description=with_memory("Find prior decisions about the database migration."),
    agent=researcher,
    callback=capture_result,
)

write_up = Task(
    description=with_memory("Write a summary doc based on the research."),
    agent=writer,
    callback=capture_result,
)

crew = Crew(agents=[researcher, writer], tasks=[research, write_up], process=Process.sequential)
crew.kickoff()

Tips

  • Crew-scoped vs user-scoped. Use org:acme/crew:<name> for memory the whole crew shares; org:acme/user:<id> for end-user-scoped memory.
  • Read-after-write. If a downstream task in the crew needs to recall what an upstream task just captured, pass wait="indexed" to client.experience so the BM25/HNSW indexes are up to date before the next task runs.

See also