fleet-memory/hindsight-integrations/crewai/test_manual.py
Ben 41db2960c5
feat: add CrewAI integration for persistent crew memory (#319)
* feat: add CrewAI integration for persistent crew memory

Implements a CrewAI ExternalMemory storage backend that maps CrewAI's
Storage interface (save/search/reset) to Hindsight's retain/recall/delete
APIs, giving crews long-term memory with fact extraction, entity tracking,
and temporal awareness across runs.

Key features:
- HindsightStorage: drop-in Storage backend for CrewAI ExternalMemory
- HindsightReflectTool: BaseTool exposing Hindsight's reflect API
- Per-agent memory banks with customizable bank resolver
- Async compatibility layer for CrewAI's threading model
- 35 unit tests, docs site page, example script

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* refactor: move CrewAI example to hindsight-cookbook

Move research_crew.py example from hindsight-integrations/crewai/examples/
to the cookbook repo and update the integration README to link there instead.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* ci: add GitHub Actions test job for CrewAI integration

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* ci: add uv.lock for frozen installs in CI

The test-crewai-integration CI job uses `uv sync --frozen` which
requires a committed lock file.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-18 17:05:52 +01:00

116 lines
2.9 KiB
Python

"""Manual integration test for hindsight-crewai.
Prerequisites:
1. Hindsight API running on localhost:8888 (./scripts/dev/start-api.sh)
2. OPENAI_API_KEY set (or configure CrewAI for another LLM provider)
3. uv pip install -e . (from this directory)
Usage:
uv run python test_manual.py
"""
from hindsight_crewai import configure, HindsightStorage, HindsightReflectTool
from crewai.memory.external.external_memory import ExternalMemory
from crewai import Agent, Crew, Task
BANK_ID = "crewai-test"
HINDSIGHT_URL = "http://localhost:8888"
# --- Configure ---
configure(hindsight_api_url=HINDSIGHT_URL, verbose=True)
storage = HindsightStorage(
bank_id=BANK_ID,
mission="Track research findings and summaries for a software team.",
)
reflect_tool = HindsightReflectTool(bank_id=BANK_ID, budget="mid")
# --- Smoke test (no LLM needed) ---
print("=== SMOKE TEST: save/search/reset ===\n")
storage.save("Python is great for data science", metadata={"task": "research"}, agent="Tester")
print("Saved memory.")
results = storage.search("What programming languages are useful?")
print(f"Search returned {len(results)} result(s):")
for r in results:
print(f" - [{r['score']}] {r['context']}")
print()
# --- Full crew test ---
print("=== RUN 1: Initial research ===\n")
researcher = Agent(
role="Researcher",
goal="Research topics and remember findings",
backstory="You are a diligent researcher who remembers everything.",
tools=[reflect_tool],
verbose=True,
)
writer = Agent(
role="Writer",
goal="Write summaries based on research",
backstory="You write clear, concise summaries.",
tools=[reflect_tool],
verbose=True,
)
research_task = Task(
description=(
"Research the benefits of functional programming. "
"List at least 3 key benefits with examples."
),
expected_output="A list of functional programming benefits with examples.",
agent=researcher,
)
summary_task = Task(
description="Write a one-paragraph summary of the research findings.",
expected_output="A concise summary paragraph.",
agent=writer,
)
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, summary_task],
external_memory=ExternalMemory(storage=storage),
verbose=True,
)
result = crew.kickoff()
print(f"\nRun 1 result:\n{result}\n")
# --- Second run: recall from memory ---
print("=== RUN 2: Recall from memory ===\n")
recall_task = Task(
description=(
"What do you already know about functional programming from previous research? "
"Use the hindsight_reflect tool to check your memories."
),
expected_output="A summary of what was previously learned.",
agent=researcher,
)
crew2 = Crew(
agents=[researcher],
tasks=[recall_task],
external_memory=ExternalMemory(storage=storage),
verbose=True,
)
result2 = crew2.kickoff()
print(f"\nRun 2 result:\n{result2}\n")
# --- Cleanup ---
print("=== CLEANUP ===\n")
storage.reset()
print("Bank reset. Done.")