crewai-endee 0.1.1b4

Endee vector database integration for CrewAI agent memory

crewai-endee

Endee vector database integration for CrewAI agent memory

crewai-endee connects Endee to CrewAI, giving your agents persistent memory with dense, hybrid, and filtered retrieval.

---

Installation

Requires Python 3.9+.

pip install crewai-endee

This installs endee, endee_model, crewai, and fastembed automatically.

Add your embedding provider (only install the one you use):

pip install cohere       # Cohere
pip install openai       # OpenAI
pip install google-genai # Google Gemini

---

Connect to Endee

First, configure your embedding provider:

embedder_config = {
    "provider": "cohere",
    "config": {"model_name": "small", "api_key": "YOUR_COHERE_API_KEY"},
}

With API token

Sign up at endee.io and get your token. See the Endee docs for details.

from crewai_endee import EndeeStorage

storage = EndeeStorage(
    index_name="my_index",
    api_token="YOUR_ENDEE_API_TOKEN",
)

Without API token (local)

Run the open-source Endee server locally. See github.com/endee-io/endee for setup instructions. Then omit api_token:

storage = EndeeStorage(
    index_name="my_index",
)

---

Dense Mode

from crewai_endee import EndeeStorage
from crewai.memory.unified_memory import Memory

storage = EndeeStorage(
    index_name="demo_dense",
    api_token=ENDEE_API_TOKEN,
)

memory = Memory(storage=storage, embedder=embedder_config, llm="gemini/gemini-2.5-flash")

---

Hybrid Mode

Add sparse_model_name to enable dense + sparse (BM25):

hybrid_storage = EndeeStorage(
    index_name="demo_hybrid",
    api_token=ENDEE_API_TOKEN,
    sparse_model_name="endee/bm25",
)

hybrid_memory = Memory(storage=hybrid_storage, embedder=embedder_config, llm="gemini/gemini-2.5-flash")

---

CrewAI Integration

Pass the Memory object directly to Crew:

from crewai import LLM, Agent, Crew, Process, Task
from crewai.memory.unified_memory import Memory
from crewai_endee import EndeeStorage

# Storage + Memory
storage = EndeeStorage(index_name="crew_memory", api_token=ENDEE_API_TOKEN)
memory = Memory(storage=storage, embedder=embedder_config, llm="gemini/gemini-2.5-flash")

# LLM
llm = LLM(model="gemini/gemini-2.5-flash", api_key=GOOGLE_API_KEY)

# Agents
analyst = Agent(
    role="Software Analyst",
    goal="Extract and interpret programming language characteristics",
    backstory="You study programming language design, typing systems, and paradigms.",
    llm=llm,
)

classifier = Agent(
    role="Language Classifier",
    goal="Categorise languages by paradigm and typing discipline",
    backstory="You classify languages using the analyst's extracted data.",
    llm=llm,
)

validator = Agent(
    role="Quality Validator",
    goal="Cross-check facts and produce a quality report",
    backstory="You verify accuracy of classifications against known facts.",
    llm=llm,
)

# Tasks
analysis_task = Task(
    description="Analyse key characteristics of Python, Java, Go, Rust, and C++.",
    expected_output="Structured summary of each language's typing, paradigm, and key features.",
    agent=analyst,
)

classification_task = Task(
    description="Classify each language by paradigm and typing discipline.",
    expected_output="A table mapping each language to its paradigm and typing.",
    agent=classifier,
)

validation_task = Task(
    description="Cross-check the classification against known facts and flag errors.",
    expected_output="Quality report with accuracy score and corrections.",
    agent=validator,
)

# Crew with Endee-backed memory
crew = Crew(
    agents=[analyst, classifier, validator],
    tasks=[analysis_task, classification_task, validation_task],
    process=Process.sequential,
    memory=memory,
    verbose=True,
)

result = crew.kickoff()
print(result)

Querying After Crew Execution

After the crew runs, query memory directly — no LLM call needed:

matches = memory.recall("Who created Go?", limit=2)
for match in matches:
    print(f"  [{match.score:.3f}] {match.record.content[:100]}")

Memory Recall Across Sessions

A new crew can recall everything the previous crew stored — no re-processing:

recall_agent = Agent(
    role="Tech Knowledge Tester",
    goal="Answer technical questions using previously stored memory",
    backstory="You answer questions by recalling stored knowledge.",
    llm=llm,
)

recall_task = Task(
    description="Using stored memory, answer: What concurrency mechanism does Go provide?",
    expected_output="A concise factual answer.",
    agent=recall_agent,
)

recall_crew = Crew(
    agents=[recall_agent],
    tasks=[recall_task],
    memory=memory,  # Same Memory object -> same Endee index
)

result = recall_crew.kickoff()
print(result)

Hybrid Mode with CrewAI Memory

One-line change — add sparse_model_name. The CrewAI wiring is identical:

hybrid_storage = EndeeStorage(
    index_name="crew_hybrid_memory",
    api_token=ENDEE_API_TOKEN,
    sparse_model_name="endee/bm25",
)

hybrid_memory = Memory(storage=hybrid_storage, embedder=embedder_config, llm="gemini/gemini-2.5-flash")

crew = Crew(
    agents=[analyst],
    tasks=[analysis_task],
    memory=hybrid_memory,
)

---

API Reference

Constructor Parameters

Parameter	Type	Required	Default	Description
index_name	str	Yes	—	Unique Endee index name
api_token	str	No	None	Endee API token (omit for local mode)
vector_dim	int	No	None	Dense vector dimension (auto-detected from first embedding if omitted)
sparse_model_name	str	No	None	"endee/bm25" for BM25, "splade_pp" for SPLADE, or any supported model key

For space_type, precision, ef_con, and other Endee index options, see the Endee docs.

StorageBackend Methods

Method	Description
save(records)	Save a list of MemoryRecord objects to the index
search(query_embedding, ...)	Vector similarity search returning [(MemoryRecord, score)]
get_record(record_id)	Retrieve a single record by ID
update(record)	Update an existing record
delete(...)	Delete records by IDs, scope, categories, or metadata
count(scope_prefix)	Count records in the index
reset(scope_prefix)	Delete the entire index
list_records(...)	List records in a scope
list_scopes(parent)	List child scopes
list_categories(scope_prefix)	List categories and counts
get_scope_info(scope)	Get scope metadata

Utility Functions

from crewai_endee import list_supported_models

for name, config in list_supported_models().items():
    print(f"  {name} — {config['description']}")

---

Full Endee documentation: docs.endee.io | GitHub: endee-io/endee | CrewAI docs: docs.crewai.com