RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: (crewai) update crew ai package and examples for latest protocol

This commit is contained in:
Vineeth Voruganti 2026-04-28 14:54:27 -04:00
parent e659b6b31f
commit 06af6559d2
14 changed files with 1334 additions and 1481 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

242
docs/v3/guides/integrations/crewai.mdx
View File

@ -5,28 +5,24 @@ description: "Build AI agents with persistent memory using CrewAI and Honcho"
sidebarTitle: 'CrewAI'
---
Integrate Honcho with CrewAI to build AI agents that maintain memory across sessions. This guide shows you how to use Honcho's memory layer with CrewAI's agent orchestration framework.
Integrate Honcho with CrewAI to build agents that maintain memory across sessions. This guide uses CrewAI's unified `Memory` API with Honcho as a custom storage backend.
<Note>
The full code is available on [GitHub](https://github.com/plastic-labs/honcho/tree/main/examples/crewai) with examples in [Python](https://github.com/plastic-labs/honcho/tree/main/examples/crewai/python/examples)
The full code is available on [GitHub](https://github.com/plastic-labs/honcho/tree/main/examples/crewai) with examples in [Python](https://github.com/plastic-labs/honcho/tree/main/examples/crewai/python/examples).
</Note>
## What We're Building
We'll create AI agents that remember and reason over past conversations. Here's how the pieces fit together:
- **CrewAI** orchestrates agent behavior and task execution
- **Honcho** stores messages and retrieves relevant context
The key benefit: CrewAI automatically retrieves relevant conversation history from Honcho without you needing to manually manage context, token limits, or message formatting.
- **CrewAI** orchestrates agents, tasks, and memory recall.
- **Honcho** persists CrewAI memory records and exposes additional context, search, and reasoning tools.
<Note>
This tutorial demonstrates single-agent setup to show how Honcho integrates with CrewAI. For production applications, you can extend this to multi-agent crews with shared or individual memory using Honcho's `peer` system.
CrewAI currently supports Python `>=3.10,<3.14`; use one of those interpreters when installing this integration.
</Note>
## Setup
Install required packages:
Install the packages:
<CodeGroup>
```bash Python (uv)
@ -38,243 +34,153 @@ pip install honcho-crewai crewai python-dotenv
```
</CodeGroup>
Use any LLM provider for your Crew. Create a `.env` file with your API keys:
Set your model provider keys and Honcho configuration:
```bash
OPENAI_API_KEY=your_openai_key
HONCHO_API_KEY=your_honcho_key
HONCHO_WORKSPACE_ID=crewai-demo
```
<Note>
This tutorial uses the Honcho demo server at https://demo.honcho.dev which runs a small instance of Honcho on the latest version. For production, get your Honcho API key at [app.honcho.dev](https://app.honcho.dev). For local development, use `environment="local"`.
</Note>
For local development, initialize the Honcho client with `environment="local"`.
## CrewAI Honcho Storage
## CrewAI Memory Storage
The `honcho_crewai` package provides `HonchoStorage`, a storage provider that implements CrewAI's `Storage` interface using Honcho's session-based memory.
<Note>
Before proceeding, it's important to understand Honcho's core concepts (`Peers` and `Sessions`). Review the [Honcho Architecture](/v3/documentation/core-concepts/architecture) to familiarize yourself with these primitives.
</Note>
`HonchoStorage` implements CrewAI's `Storage` interface using Honcho's `peer` and `session` primitives.
`HonchoMemoryStorage` implements CrewAI's current `StorageBackend` protocol and can be passed directly to `Memory(storage=...)`.
```python
storage = HonchoStorage(
user_id="demo-user", # Required: Honcho `peer` ID for the user
session_id=None, # Optional: Specific `session` ID (auto-generated UUID if None)
honcho_client=None, # Optional: Pre-configured Honcho client instance
from crewai import Memory
from honcho import Honcho
from honcho_crewai import HonchoMemoryStorage
honcho = Honcho(workspace_id="crewai-demo")
storage = HonchoMemoryStorage(
peer_id="user-123",
session_id="session-123",
honcho_client=honcho,
)
memory = Memory(storage=storage)
```
CrewAI embeds memory records before storing them. The Honcho backend stores those records as Honcho messages, keeps CrewAI metadata in message metadata, and performs vector search over the stored embeddings.
```python
memory.remember(
"The user is learning Python and wants to build web applications.",
scope="/users/user-123",
categories=["preferences"],
metadata={"source": "onboarding"},
)
```
The `HonchoStorage` class implements three key methods:
- **`save()`** - Stores messages in Honcho's `session`, associating them with the appropriate `peer` (user or assistant)
- **`search()`** - Performs semantic vector search using `session.search()` to find messages most relevant to the query. Supports optional `filters` parameter for fine-grained scoping.
- **`reset()`** - Creates a new `session` to start fresh conversations
CrewAI automatically calls these methods when agents need to store or retrieve memory, creating a seamless integration.
### Search with Filters
The `search()` method supports an optional `filters` parameter for fine-grained scoping of search results:
```python
# Search with peer_id filter (only messages from a specific peer)
results = storage.search("query", filters={"peer_id": "user123"})
# Search with metadata filter
results = storage.search("query", filters={"metadata": {"priority": "high"}})
# Search with time range filter
results = storage.search("query", filters={"created_at": {"gte": "2024-01-01"}})
# Complex filter with logical operators
results = storage.search("query", filters={
"AND": [
{"peer_id": "user123"},
{"metadata": {"topic": "python"}}
]
})
```
For the full filter syntax including logical operators (AND, OR, NOT), comparison operators, and metadata filtering, see the [Using Filters](https://docs.honcho.dev/v3/documentation/features/advanced/using-filters) documentation.
<Note>
For comprehensive details about CrewAI's memory system, see the [official CrewAI Memory documentation](https://docs.crewai.com/en/concepts/memory).
</Note>
Let's create a basic example showing how CrewAI agents use Honcho's memory automatically:
Use the memory instance with a crew:
```python Python
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from crewai.memory.external.external_memory import ExternalMemory
from honcho_crewai import HonchoStorage
load_dotenv()
storage = HonchoStorage(user_id="simple-demo-user")
external_memory = ExternalMemory(storage=storage)
messages = [
("user", "I'm learning Python programming"),
("assistant", "Great! Python is an excellent language to learn."),
("user", "I'm particularly interested in web development"),
]
for role, message in messages:
external_memory.save(message, metadata={"agent": role})
from crewai import Agent, Crew, Process, Task
agent = Agent(
role="Programming Mentor",
goal="Help users learn programming by remembering their interests and progress",
backstory=(
"You are a patient programming mentor who remembers what students "
"have told you about their learning journey and interests."
),
verbose=True,
allow_delegation=False
backstory="You are a patient programming mentor.",
)
task = Task(
description=(
"Based on what you know about the user's interests, "
"suggest a simple web development project they could build to practice Python."
),
expected_output="A specific project suggestion with brief explanation",
agent=agent
description="Suggest a Python web project that matches the user's interests.",
expected_output="A specific project suggestion with a brief explanation",
agent=agent,
)
crew = Crew(
agents=[agent],
tasks=[task],
process=Process.sequential,
external_memory=external_memory,
verbose=True
memory=memory,
verbose=True,
)
result = crew.kickoff()
print(result.raw)
```
<Note>
`HonchoStorage` is still available as a compatibility adapter for older CrewAI `ExternalMemory` integrations, but new projects should use `HonchoMemoryStorage`.
</Note>
## CrewAI Tool Integration
Honcho provides specialized tools that give CrewAI agents explicit control over memory retrieval:
Honcho also provides tools that let agents explicitly retrieve memory:
- **`HonchoGetContextTool`** - Retrieves comprehensive conversation history with token limits. Use for tasks needing broad conversation understanding.
- **`HonchoDialecticTool`** - Queries representations about `peer`s. Use for understanding user preferences and characteristics without full message history.
- **`HonchoSearchTool`** - Performs semantic search for specific information. Supports optional `filters` parameter for fine-grained scoping. Use for targeted queries like "what did the user say about budget?"
<Tip>
Agents can use multiple tools in sequence: search for topics, query dialectic for preferences, then get full context for generation.
</Tip>
Here's an example demonstrating all three tools:
- **`HonchoGetContextTool`** retrieves session context with token limits.
- **`HonchoDialecticTool`** queries Honcho's representation of a peer.
- **`HonchoSearchTool`** performs semantic search over session messages.
```python Python
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from crewai import Agent, Crew, Process, Task
from honcho import Honcho
from honcho_crewai import (
HonchoGetContextTool,
HonchoDialecticTool,
HonchoGetContextTool,
HonchoSearchTool,
)
load_dotenv()
honcho = Honcho()
user_id = "demo-user-45"
honcho = Honcho(workspace_id="crewai-demo")
user_id = "demo-user"
session_id = "tools-demo-session"
user = honcho.peer(user_id)
session = honcho.session(session_id)
messages = [
for message in [
"I'm planning a trip to Japan in March",
"I love trying authentic local cuisine, especially ramen and sushi",
"I love authentic local cuisine, especially ramen and sushi",
"My budget is around $3000 for a 10-day trip",
"I'm interested in visiting both Tokyo and Kyoto",
"I prefer staying in traditional ryokans over hotels",
]
for msg in messages:
session.add_messages([user.message(msg)])
]:
session.add_messages([user.message(message)])
context_tool = HonchoGetContextTool(
honcho=honcho, session_id=session_id, peer_id=user_id
honcho=honcho,
session_id=session_id,
peer_id=user_id,
)
dialectic_tool = HonchoDialecticTool(
honcho=honcho, session_id=session_id, peer_id=user_id
honcho=honcho,
session_id=session_id,
peer_id=user_id,
)
search_tool = HonchoSearchTool(honcho=honcho, session_id=session_id)
# Note: The search tool supports optional filters for fine-grained scoping
# Agents can use filters like {"peer_id": "user123"} or {"metadata": {"priority": "high"}}
travel_agent = Agent(
role="Travel Planning Specialist",
goal="Create personalized travel recommendations using memory tools",
backstory=(
"You are an expert travel planner with access to conversation memory tools. "
"Use the tools to understand the user's preferences before making recommendations."
),
backstory="You are an expert travel planner with access to memory tools.",
tools=[context_tool, dialectic_tool, search_tool],
verbose=True,
allow_delegation=False
allow_delegation=False,
)
task = Task(
description=(
"Create a personalized 3-day Tokyo itinerary. "
"Use the memory tools to understand:\n"
" • Food preferences (use search_tool for 'cuisine' or 'food')\n"
" • Travel style and budget (use dialectic_tool to query user knowledge)\n"
" • Recent context (use context_tool to get conversation history)\n"
"Then create a detailed plan matching their interests."
),
expected_output=(
"A 3-day Tokyo itinerary with:\n"
" • Daily activities matching user interests\n"
" • Restaurant recommendations\n"
" • Accommodation suggestions\n"
" • Budget considerations"
),
agent=travel_agent
description="Create a personalized 3-day Tokyo itinerary using the memory tools.",
expected_output="A 3-day Tokyo itinerary with activities, restaurants, and budget notes",
agent=travel_agent,
)
crew = Crew(
agents=[travel_agent],
tasks=[task],
process=Process.sequential,
verbose=True
verbose=True,
)
crew.kickoff()
```
## Tool-Based vs Automatic Memory
## When To Use Each
**Use `HonchoStorage`** for automatic memory - CrewAI handles everything transparently. Best for simple conversational flows.
Use `HonchoMemoryStorage` when you want CrewAI to handle recall automatically through the unified memory system.
**Use Honcho Tools** for strategic control - agents decide when and how to query memory. Best for multi-step reasoning, when different query types are needed, or multi-agent systems.
Use the Honcho tools when the agent should decide when and how to query memory, search messages, or ask Honcho for a peer-level representation.
You can combine both: automatic memory for baseline context, tools for specific queries. See the [hybrid memory example](https://github.com/plastic-labs/honcho/blob/main/examples/crewai/python/examples/hybrid_memory_example.py) for a complete implementation.
<Note>
**Multi-Agent Memory:** Use Honcho tools with different `peer_id` values to give each agent distinct memory and identity.
</Note>
## Next Steps
Now that you have a working CrewAI integration with Honcho, you can:
- **Create specialized agents** with domain-specific memory and context
- **Use CrewAI's advanced features** like hierarchical processes, tool delegation, and conditional task execution
- **Leverage logical reasoning** via the Dialectic API for deep `peer` understanding
- **Implement custom tools** to give agents explicit control over memory retrieval
You can combine both: unified memory for baseline context, tools for targeted retrieval. See the [hybrid memory example](https://github.com/plastic-labs/honcho/blob/main/examples/crewai/python/examples/hybrid_memory_example.py) for a complete implementation.
## Related Resources
@ -286,7 +192,7 @@ Now that you have a working CrewAI integration with Honcho, you can:
Learn about retrieving and formatting conversation context
</Card>
<Card title="Chat API" icon="brain" href="/v3/documentation/features/chat">
Query `peer` representations for deeper understanding
Query peer representations for deeper understanding
</Card>
<Card title="LangGraph Integration" icon="diagram-project" href="/v3/guides/integrations/langgraph">
Build stateful agents with LangGraph and Honcho

71
examples/crewai/python/README.md
View File

@ -5,58 +5,67 @@ Build CrewAI agents with persistent memory and reasoning capabilities powered by
## Installation
```bash
pip install honcho-crewai
uv add honcho-crewai crewai python-dotenv
```
CrewAI currently supports Python `>=3.10,<3.14`; this package follows the same range.
## Quick Start
```python
from crewai import Agent, Task, Crew, Process
from crewai.memory.external.external_memory import ExternalMemory
from honcho_crewai import HonchoStorage
from crewai import Agent, Crew, Memory, Process, Task
from honcho import Honcho
from honcho_crewai import HonchoMemoryStorage
# Initialize Honcho storage
storage = HonchoStorage(user_id="user-123")
external_memory = ExternalMemory(storage=storage)
honcho = Honcho(workspace_id="crewai-demo")
storage = HonchoMemoryStorage(
peer_id="user-123",
session_id="session-123",
honcho_client=honcho,
)
memory = Memory(storage=storage)
# Create agent with memory
agent = Agent(
role="AI Assistant",
goal="Help users with persistent memory",
backstory="You remember past conversations.",
memory.remember(
"The user is learning Python and wants to build web applications.",
scope="/users/user-123",
categories=["preferences"],
metadata={"source": "onboarding"},
)
agent = Agent(
role="Programming Mentor",
goal="Help users learn programming by remembering their interests and progress",
backstory="You are a patient programming mentor.",
)
task = Task(
description="Suggest a Python web project that matches the user's interests.",
expected_output="A specific project suggestion with a brief explanation",
agent=agent,
)
# Create crew with external memory
crew = Crew(
agents=[agent],
tasks=[task],
external_memory=external_memory
process=Process.sequential,
memory=memory,
)
result = crew.kickoff()
print(result.raw)
```
## Features
- **Automatic Memory**: CrewAI agents automatically store and retrieve conversation context
- **Semantic Search**: Find relevant past messages using vector similarity
- **Logical Reasoning**: Query what the system knows about users via the Dialectic API
- **Multi-Agent Support**: Give each agent distinct memory and identity
- **Tools Integration**: `HonchoGetContextTool`, `HonchoDialecticTool`, and `HonchoSearchTool` for explicit memory control
- `HonchoMemoryStorage`: CrewAI unified `Memory` storage backend.
- `HonchoStorage`: compatibility adapter for older CrewAI `ExternalMemory` usage.
- `HonchoGetContextTool`, `HonchoDialecticTool`, and `HonchoSearchTool` for explicit Honcho memory retrieval.
- Lazy Honcho peer/session handles, matching the latest Honcho SDK get-or-create behavior.
## Documentation
For comprehensive guides, examples, and API reference, visit:
**[https://docs.honcho.dev/v3/integrations/crewai](https://docs.honcho.dev/v3/integrations/crewai)**
## Examples
Check out complete examples in the [GitHub repository](https://github.com/plastic-labs/honcho/tree/main/examples/crewai/python/examples).
For guides and API reference, visit [docs.honcho.dev](https://docs.honcho.dev/v3/guides/integrations/crewai).
## License
AGPL-3.0-or-later
## Support
- Report issues: [GitHub Issues](https://github.com/plastic-labs/honcho/issues)
- Documentation: [docs.honcho.dev](https://docs.honcho.dev)
- Website: [honcho.dev](https://honcho.dev)

34
examples/crewai/python/examples/hybrid_memory_example.py
View File

@ -1,18 +1,17 @@
"""
Hybrid Memory Example: Combining Automatic Memory + Explicit Tools
Demonstrates combining automatic memory (HonchoStorage) with explicit memory tools.
Demonstrates combining automatic memory (HonchoMemoryStorage) with explicit memory tools.
The agent gets baseline context automatically but can also make targeted queries.
"""
from crewai import Agent, Crew, Memory, Process, Task
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from crewai.memory.external.external_memory import ExternalMemory
from honcho import Honcho
from honcho_crewai import (
HonchoStorage,
HonchoSearchTool,
HonchoDialecticTool,
HonchoMemoryStorage,
HonchoSearchTool,
)
load_dotenv()
@ -25,13 +24,13 @@ def main():
user_id = "hybrid-demo-user"
session_id = "hybrid-demo-session"
# Setup automatic memory
storage = HonchoStorage(
user_id=user_id,
# Setup unified CrewAI memory
storage = HonchoMemoryStorage(
peer_id=user_id,
session_id=session_id,
honcho_client=honcho
honcho_client=honcho,
)
external_memory = ExternalMemory(storage=storage)
memory = Memory(storage=storage)
# Add conversation history
messages = [
@ -45,7 +44,12 @@ def main():
]
for role, message in messages:
external_memory.save(message, metadata={"agent": role})
memory.remember(
message,
scope=f"/users/{user_id}/conversation",
categories=["conversation"],
metadata={"role": role},
)
# Create memory tools for targeted queries
search_tool = HonchoSearchTool(honcho=honcho, session_id=session_id)
@ -63,7 +67,7 @@ def main():
),
tools=[search_tool, dialectic_tool],
verbose=True,
allow_delegation=False
allow_delegation=False,
)
# Create task
@ -75,7 +79,7 @@ def main():
"Then create a personalized itinerary with activities and restaurant recommendations."
),
expected_output="A 3-day Tokyo itinerary with daily activities and dining suggestions",
agent=travel_agent
agent=travel_agent,
)
# Execute with hybrid memory: automatic baseline + explicit tools
@ -83,8 +87,8 @@ def main():
agents=[travel_agent],
tasks=[task],
process=Process.sequential,
external_memory=external_memory, # Automatic memory!
verbose=True
memory=memory,
verbose=True,
)
result = crew.kickoff()

47
examples/crewai/python/examples/interactive_chat.py
View File

@ -6,11 +6,9 @@ CrewAI for agent orchestration, OpenAI for the AI model, and Honcho for memory
management via the honcho_crewai package.
"""
from typing import Optional
from crewai import Agent, Crew, Memory, Process, Task
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from crewai.memory.external.external_memory import ExternalMemory
from honcho_crewai import HonchoStorage
from honcho_crewai import HonchoMemoryStorage
load_dotenv()
@ -18,9 +16,9 @@ load_dotenv()
def run_conversation_turn(
user_id: str,
user_input: str,
session_id: Optional[str] = None,
storage: Optional[HonchoStorage] = None
) -> tuple[str, HonchoStorage]:
session_id: str | None = None,
storage: HonchoMemoryStorage | None = None,
) -> tuple[str, HonchoMemoryStorage]:
"""
Run a single conversation turn with the CrewAI agent.
@ -28,7 +26,7 @@ def run_conversation_turn(
user_id: Unique identifier for the user
user_input: User's message
session_id: Optional session ID for conversation continuity
storage: Optional existing HonchoStorage instance
storage: Optional existing HonchoMemoryStorage instance
Returns:
Tuple of (agent_response, storage_instance)
@ -37,13 +35,17 @@ def run_conversation_turn(
if storage is None:
if not session_id:
session_id = f"session_{user_id}"
storage = HonchoStorage(user_id=user_id, session_id=session_id)
storage = HonchoMemoryStorage(peer_id=user_id, session_id=session_id)
# Create ExternalMemory wrapper for automatic context retrieval
external_memory = ExternalMemory(storage=storage)
memory = Memory(storage=storage)
# Save user input to memory
external_memory.save(user_input, metadata={"agent": "user"})
memory.remember(
user_input,
scope=f"/users/{user_id}/conversation",
categories=["conversation"],
metadata={"role": "user"},
)
# Create an agent with memory
agent = Agent(
@ -54,23 +56,23 @@ def run_conversation_turn(
"You use context from previous interactions to provide personalized and relevant responses."
),
verbose=False,
allow_delegation=False
allow_delegation=False,
)
# Create task for the agent
task = Task(
description=f"Respond to the user's message: {user_input}",
expected_output="A helpful and contextually relevant response that considers conversation history",
agent=agent
agent=agent,
)
# Create crew with external memory - enables automatic context retrieval
# Create crew with unified memory - enables automatic context retrieval
crew = Crew(
agents=[agent],
tasks=[task],
process=Process.sequential,
external_memory=external_memory,
verbose=False
memory=memory,
verbose=False,
)
# Execute - CrewAI automatically retrieves relevant context from Honcho
@ -78,7 +80,12 @@ def run_conversation_turn(
# Save assistant response back to memory
response_text = str(result.raw)
external_memory.save(response_text, metadata={"agent": "assistant"})
memory.remember(
response_text,
scope=f"/users/{user_id}/conversation",
categories=["conversation"],
metadata={"role": "assistant"},
)
return response_text, storage
@ -93,7 +100,7 @@ def main():
while True:
user_input = input("You: ")
if user_input.lower() in ['quit', 'exit']:
if user_input.lower() in ["quit", "exit"]:
print("Goodbye!")
break
@ -104,7 +111,7 @@ def main():
response, storage = run_conversation_turn(
user_id=user_id,
user_input=user_input,
storage=storage
storage=storage,
)
print(f"Assistant: {response}\n")
except Exception as e:

35
examples/crewai/python/examples/simple_example.py
View File

@ -1,23 +1,27 @@
"""
Simple Honcho + CrewAI Example
A minimal example showing how to use Honcho's ExternalMemory with CrewAI agents.
A minimal example showing how to use Honcho-backed unified Memory with CrewAI agents.
This demonstrates the basic pattern for persistent conversation memory.
"""
from crewai import Agent, Crew, Memory, Process, Task
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from crewai.memory.external.external_memory import ExternalMemory
from honcho_crewai import HonchoStorage
from honcho_crewai import HonchoMemoryStorage
load_dotenv()
def main():
"""Simple example of CrewAI agent with Honcho memory."""
# Initialize Honcho storage
storage = HonchoStorage(user_id="simple-demo-user")
external_memory = ExternalMemory(storage=storage)
user_id = "simple-demo-user"
# Initialize CrewAI unified memory backed by Honcho
storage = HonchoMemoryStorage(
peer_id=user_id,
session_id="simple-demo-session",
)
memory = Memory(storage=storage)
# Add some conversation history
messages = [
@ -27,7 +31,12 @@ def main():
]
for role, message in messages:
external_memory.save(message, metadata={"agent": role})
memory.remember(
message,
scope=f"/users/{user_id}/conversation",
categories=["conversation"],
metadata={"role": role},
)
# Create agent with memory
agent = Agent(
@ -38,7 +47,7 @@ def main():
"have told you about their learning journey and interests."
),
verbose=True,
allow_delegation=False
allow_delegation=False,
)
# Create task
@ -48,16 +57,16 @@ def main():
"suggest a simple web development project they could build to practice Python."
),
expected_output="A specific project suggestion with brief explanation",
agent=agent
agent=agent,
)
# Execute with memory - CrewAI automatically retrieves relevant context!
# Execute with memory - CrewAI automatically retrieves relevant context.
crew = Crew(
agents=[agent],
tasks=[task],
process=Process.sequential,
external_memory=external_memory,
verbose=True
memory=memory,
verbose=True,
)
result = crew.kickoff()

13
examples/crewai/python/examples/tools_example.py
View File

@ -7,15 +7,15 @@ Demonstrates how to equip CrewAI agents with Honcho's memory tools:
- HonchoSearchTool: Perform semantic search across session messages
These tools give agents explicit control over memory retrieval, beyond the
automatic memory provided by ExternalMemory.
automatic memory provided by CrewAI's unified Memory API.
"""
from crewai import Agent, Crew, Process, Task
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from honcho import Honcho
from honcho_crewai import (
HonchoGetContextTool,
HonchoDialecticTool,
HonchoGetContextTool,
HonchoSearchTool,
)
@ -81,7 +81,7 @@ def main():
),
tools=[context_tool, dialectic_tool, search_tool],
verbose=True,
allow_delegation=False
allow_delegation=False,
)
print(" ✓ Agent created with 3 Honcho tools\n")
@ -104,7 +104,7 @@ def main():
" • Accommodation suggestions\n"
" • Budget considerations"
),
agent=travel_agent
agent=travel_agent,
)
print(" ✓ Task created\n")
@ -116,10 +116,11 @@ def main():
agents=[travel_agent],
tasks=[task],
process=Process.sequential,
verbose=True
verbose=True,
)
crew.kickoff()
if __name__ == "__main__":
main()

20
examples/crewai/python/pyproject.toml
View File

@ -1,9 +1,9 @@
[project]
name = "honcho-crewai"
version = "0.2.0"
version = "0.3.0"
description = "CrewAI integration with Honcho for persistent agent memory"
readme = "README.md"
requires-python = ">=3.10"
requires-python = ">=3.10,<3.14"
license = {text = "AGPL-3.0-or-later"}
authors = [
{name = "Plastic Labs", email = "hello@plasticlabs.ai"}
@ -33,15 +33,25 @@ classifiers = [
"Framework :: Pydantic",
]
dependencies = [
"crewai>=0.134.0",
"honcho-ai>=2.0.0",
"crewai>=1.14.3,<2.0.0",
"honcho-ai>=2.1.1,<3.0.0",
"openai>=1.0.0",
"python-dotenv>=1.0.0",
]
[project.urls]
Homepage = "https://honcho.dev"
Documentation = "https://docs.honcho.dev/v3/integrations/crewai"
Documentation = "https://docs.honcho.dev/v3/guides/integrations/crewai"
Repository = "https://github.com/plastic-labs/honcho"
"Bug Tracker" = "https://github.com/plastic-labs/honcho/issues"
Changelog = "https://github.com/plastic-labs/honcho/blob/main/CHANGELOG.md"
[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
where = ["src"]
[tool.pytest.ini_options]
testpaths = ["tests"]

24
examples/crewai/python/src/honcho_crewai/__init__.py
View File

@ -6,15 +6,18 @@ enabling AI agents to maintain persistent memory across conversations.
Example:
```python
from honcho_crewai import HonchoStorage, HonchoSearchTool, HonchoGetContextTool, HonchoDialecticTool
from crewai.memory.external.external_memory import ExternalMemory
from crewai import Agent, Task, Crew
from honcho_crewai import HonchoMemoryStorage, HonchoSearchTool, HonchoGetContextTool, HonchoDialecticTool
from crewai import Agent, Task, Crew, Memory
from honcho import Honcho
# Initialize Honcho client and storage
# Initialize Honcho client and CrewAI memory
honcho = Honcho()
storage = HonchoStorage(user_id="user123", honcho_client=honcho)
external_memory = ExternalMemory(storage=storage)
storage = HonchoMemoryStorage(
peer_id="user123",
session_id="session123",
honcho_client=honcho,
)
memory = Memory(storage=storage)
# Create tools for agents
search_tool = HonchoSearchTool(honcho=honcho, session_id=storage.session_id)
@ -36,28 +39,29 @@ Example:
agent=agent,
)
# Create crew with external memory
# Create crew with unified memory
crew = Crew(
agents=[agent],
tasks=[task],
external_memory=external_memory
memory=memory
)
```
"""
from honcho_crewai.exceptions import HonchoDependencyError
from honcho_crewai.storage import HonchoStorage
from honcho_crewai.storage import HonchoMemoryStorage, HonchoStorage
from honcho_crewai.tools import (
HonchoDialecticTool,
HonchoGetContextTool,
HonchoSearchTool,
)
__version__ = "0.2.0"
__version__ = "0.3.0"
__all__ = [
"HonchoDependencyError",
"HonchoDialecticTool",
"HonchoGetContextTool",
"HonchoMemoryStorage",
"HonchoSearchTool",
"HonchoStorage",
]

569
examples/crewai/python/src/honcho_crewai/storage.py
View File

@ -1,104 +1,457 @@
"""
Honcho Storage for CrewAI External Memory
Honcho storage adapters for CrewAI memory.
This module provides a Honcho-backed storage provider for CrewAI's external memory
system, enabling AI agents to maintain persistent conversation memory across sessions.
`HonchoMemoryStorage` implements CrewAI's current unified memory
`StorageBackend` protocol. `HonchoStorage` is kept as a compatibility adapter
for older CrewAI `ExternalMemory` usage.
"""
from __future__ import annotations
import asyncio
import logging
import math
import uuid
from typing import Any, Optional
from collections.abc import Iterable
from datetime import UTC, datetime
from typing import Any
from crewai.memory.storage.interface import Storage
from honcho import Honcho
from honcho_crewai.exceptions import HonchoDependencyError
try: # CrewAI <= 1.9 external memory interface.
from crewai.memory.storage.interface import Storage as LegacyStorage
except ModuleNotFoundError: # CrewAI >= 1.10 unified memory only.
class LegacyStorage: # type: ignore[no-redef]
pass
try: # CrewAI >= 1.10 unified memory types.
from crewai.memory.types import MemoryRecord, ScopeInfo
except ModuleNotFoundError:
MemoryRecord = None # type: ignore[assignment]
ScopeInfo = None # type: ignore[assignment]
logger = logging.getLogger(__name__)
_MEMORY_KIND = "crewai_memory_record"
_KIND_KEY = "honcho_crewai_kind"
_DELETED_KEY = "honcho_crewai_deleted"
_RECORD_ID_KEY = "crewai_record_id"
_SCOPE_KEY = "crewai_scope"
_CATEGORIES_KEY = "crewai_categories"
_MEMORY_METADATA_KEY = "crewai_metadata"
_IMPORTANCE_KEY = "crewai_importance"
_CREATED_AT_KEY = "crewai_created_at"
_LAST_ACCESSED_KEY = "crewai_last_accessed"
_EMBEDDING_KEY = "crewai_embedding"
_SOURCE_KEY = "crewai_source"
_PRIVATE_KEY = "crewai_private"
class HonchoStorage(Storage):
def _require_unified_memory() -> None:
if MemoryRecord is None or ScopeInfo is None:
raise HonchoDependencyError("CrewAI unified memory", "uv add crewai>=1.14.3")
def _iso(value: datetime | None) -> str | None:
return value.isoformat() if value else None
def _parse_datetime(value: Any, fallback: datetime | None = None) -> datetime:
if isinstance(value, datetime):
return value
if isinstance(value, str):
try:
return datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
logger.debug("Could not parse datetime %r", value)
return fallback or datetime.now(UTC)
def _scope_matches(scope: str, scope_prefix: str | None) -> bool:
if scope_prefix in (None, "", "/"):
return True
normalized = scope_prefix.rstrip("/")
return scope == normalized or scope.startswith(f"{normalized}/")
def _category_matches(
record_categories: list[str], categories: list[str] | None
) -> bool:
if not categories:
return True
return bool(set(record_categories).intersection(categories))
def _metadata_matches(
metadata: dict[str, Any], metadata_filter: dict[str, Any] | None
) -> bool:
if not metadata_filter:
return True
return all(metadata.get(key) == value for key, value in metadata_filter.items())
def _cosine_similarity(left: list[float] | None, right: list[float] | None) -> float:
if not left or not right or len(left) != len(right):
return 0.0
dot_product = sum(a * b for a, b in zip(left, right, strict=True))
left_norm = math.sqrt(sum(a * a for a in left))
right_norm = math.sqrt(sum(b * b for b in right))
if left_norm == 0.0 or right_norm == 0.0:
return 0.0
return dot_product / (left_norm * right_norm)
class HonchoMemoryStorage:
"""
Honcho-backed storage provider for CrewAI external memory.
CrewAI unified memory storage backend backed by Honcho messages.
Implements CrewAI's Storage interface using Honcho's session-based memory,
allowing agents to maintain context across conversations.
CrewAI's current memory system embeds records before passing them to custom
storage. This adapter stores those embeddings in Honcho message metadata and
performs vector search locally over the session's active memory records.
"""
Attributes:
honcho: The Honcho client instance
user: Peer representing the user
assistant: Peer representing the AI assistant
session: The conversation session
session_id: Unique identifier for the session
def __init__(
self,
*,
session_id: str | None = None,
peer_id: str = "crewai-memory",
honcho_client: Honcho | None = None,
) -> None:
_require_unified_memory()
self.honcho = honcho_client or Honcho()
self.session_id = session_id or str(uuid.uuid4())
self.peer_id = peer_id
self._session: Any | None = None
self._peer: Any | None = None
Example:
```python
from honcho_crewai import HonchoStorage
from crewai.memory.external.external_memory import ExternalMemory
@property
def session(self) -> Any:
if self._session is None:
self._session = self.honcho.session(self.session_id)
return self._session
# Initialize storage
storage = HonchoStorage(user_id="user123")
@property
def peer(self) -> Any:
if self._peer is None:
self._peer = self.honcho.peer(self.peer_id)
return self._peer
# Use with CrewAI's external memory
external_memory = ExternalMemory(storage=storage)
```
def save(self, records: list[Any]) -> None:
"""Save CrewAI memory records to Honcho."""
if not records:
return
messages = [
self.peer.message(
record.content,
metadata=self._record_metadata(record),
created_at=record.created_at,
)
for record in records
]
self.session.add_messages(messages)
def search(
self,
query_embedding: list[float],
scope_prefix: str | None = None,
categories: list[str] | None = None,
metadata_filter: dict[str, Any] | None = None,
limit: int = 10,
min_score: float = 0.0,
) -> list[tuple[Any, float]]:
"""Search records by cosine similarity over CrewAI-provided embeddings."""
matches: list[tuple[Any, float]] = []
for _, record in self._active_record_messages():
if not self._record_matches(
record, scope_prefix, categories, metadata_filter
):
continue
score = _cosine_similarity(query_embedding, record.embedding)
if score >= min_score:
matches.append((record, score))
matches.sort(key=lambda item: item[1], reverse=True)
return matches[:limit]
def delete(
self,
scope_prefix: str | None = None,
categories: list[str] | None = None,
record_ids: list[str] | None = None,
older_than: datetime | None = None,
metadata_filter: dict[str, Any] | None = None,
) -> int:
"""Tombstone records that match the delete criteria."""
deleted = 0
record_id_set = set(record_ids or [])
for message, record in self._active_record_messages():
if record_id_set and record.id not in record_id_set:
continue
if not self._record_matches(
record, scope_prefix, categories, metadata_filter
):
continue
if older_than is not None and record.created_at >= older_than:
continue
metadata = dict(message.metadata)
metadata[_DELETED_KEY] = True
self.session.update_message(message, metadata=metadata)
deleted += 1
return deleted
def update(self, record: Any) -> None:
"""Replace an existing record by tombstoning old copies and saving the new one."""
self.delete(record_ids=[record.id])
self.save([record])
def get_record(self, record_id: str) -> Any | None:
"""Return the newest active record with the given ID."""
records = [
record
for _, record in self._active_record_messages()
if record.id == record_id
]
if not records:
return None
return max(records, key=lambda record: record.created_at)
def list_records(
self,
scope_prefix: str | None = None,
limit: int = 200,
offset: int = 0,
) -> list[Any]:
"""List active records, newest first."""
records = [
record
for _, record in self._active_record_messages()
if _scope_matches(record.scope, scope_prefix)
]
records.sort(key=lambda record: record.created_at, reverse=True)
return records[offset : offset + limit]
def get_scope_info(self, scope: str) -> Any:
"""Build CrewAI scope metadata from active Honcho-backed records."""
_require_unified_memory()
records = self.list_records(scope_prefix=scope, limit=10_000)
categories = sorted(
{category for record in records for category in record.categories}
)
created_at_values = [record.created_at for record in records]
return ScopeInfo( # type: ignore[operator]
path=scope,
record_count=len(records),
categories=categories,
oldest_record=min(created_at_values) if created_at_values else None,
newest_record=max(created_at_values) if created_at_values else None,
child_scopes=self.list_scopes(scope),
)
def list_scopes(self, parent: str = "/") -> list[str]:
"""List immediate child scopes below `parent`."""
children: set[str] = set()
parent = parent.rstrip("/") or "/"
for record in self.list_records(scope_prefix=parent, limit=10_000):
scope = record.scope.rstrip("/") or "/"
if scope == parent:
continue
if parent == "/":
parts = [part for part in scope.split("/") if part]
if parts:
children.add(f"/{parts[0]}")
else:
remainder = scope.removeprefix(parent).strip("/")
if remainder:
children.add(f"{parent}/{remainder.split('/')[0]}")
return sorted(children)
def list_categories(self, scope_prefix: str | None = None) -> dict[str, int]:
"""Count categories in active records."""
counts: dict[str, int] = {}
for record in self.list_records(scope_prefix=scope_prefix, limit=10_000):
for category in record.categories:
counts[category] = counts.get(category, 0) + 1
return counts
def count(self, scope_prefix: str | None = None) -> int:
"""Count active records in a scope."""
return len(self.list_records(scope_prefix=scope_prefix, limit=10_000))
def reset(self, scope_prefix: str | None = None) -> None:
"""Tombstone all records in a scope, or all records when no scope is given."""
self.delete(scope_prefix=scope_prefix)
async def asave(self, records: list[Any]) -> None:
await asyncio.to_thread(self.save, records)
async def asearch(
self,
query_embedding: list[float],
scope_prefix: str | None = None,
categories: list[str] | None = None,
metadata_filter: dict[str, Any] | None = None,
limit: int = 10,
min_score: float = 0.0,
) -> list[tuple[Any, float]]:
return await asyncio.to_thread(
self.search,
query_embedding,
scope_prefix,
categories,
metadata_filter,
limit,
min_score,
)
async def adelete(
self,
scope_prefix: str | None = None,
categories: list[str] | None = None,
record_ids: list[str] | None = None,
older_than: datetime | None = None,
metadata_filter: dict[str, Any] | None = None,
) -> int:
return await asyncio.to_thread(
self.delete,
scope_prefix,
categories,
record_ids,
older_than,
metadata_filter,
)
def _record_metadata(self, record: Any) -> dict[str, Any]:
return {
_KIND_KEY: _MEMORY_KIND,
_DELETED_KEY: False,
_RECORD_ID_KEY: record.id,
_SCOPE_KEY: record.scope,
_CATEGORIES_KEY: list(record.categories),
_MEMORY_METADATA_KEY: dict(record.metadata),
_IMPORTANCE_KEY: record.importance,
_CREATED_AT_KEY: _iso(record.created_at),
_LAST_ACCESSED_KEY: _iso(record.last_accessed),
_EMBEDDING_KEY: record.embedding,
_SOURCE_KEY: record.source,
_PRIVATE_KEY: record.private,
}
def _active_record_messages(self) -> Iterable[tuple[Any, Any]]:
for message in self._record_messages():
metadata = message.metadata or {}
if metadata.get(_DELETED_KEY):
continue
yield message, self._message_to_record(message)
def _record_messages(self) -> Iterable[Any]:
filters = {"metadata": {_KIND_KEY: _MEMORY_KIND}}
for message in self.session.messages(filters=filters, size=100, reverse=True):
if (message.metadata or {}).get(_KIND_KEY) == _MEMORY_KIND:
yield message
def _message_to_record(self, message: Any) -> Any:
_require_unified_memory()
metadata = message.metadata or {}
return MemoryRecord( # type: ignore[operator]
id=metadata[_RECORD_ID_KEY],
content=message.content,
scope=metadata.get(_SCOPE_KEY, "/"),
categories=list(metadata.get(_CATEGORIES_KEY) or []),
metadata=dict(metadata.get(_MEMORY_METADATA_KEY) or {}),
importance=metadata.get(_IMPORTANCE_KEY, 0.5),
created_at=_parse_datetime(
metadata.get(_CREATED_AT_KEY), message.created_at
),
last_accessed=_parse_datetime(
metadata.get(_LAST_ACCESSED_KEY), message.created_at
),
embedding=metadata.get(_EMBEDDING_KEY),
source=metadata.get(_SOURCE_KEY),
private=bool(metadata.get(_PRIVATE_KEY, False)),
)
def _record_matches(
self,
record: Any,
scope_prefix: str | None,
categories: list[str] | None,
metadata_filter: dict[str, Any] | None,
) -> bool:
return (
_scope_matches(record.scope, scope_prefix)
and _category_matches(record.categories, categories)
and _metadata_matches(record.metadata, metadata_filter)
)
class HonchoStorage(LegacyStorage):
"""
Backwards-compatible Honcho storage for CrewAI `ExternalMemory`.
New CrewAI projects should prefer `HonchoMemoryStorage` with
`crewai.Memory(storage=...)`.
"""
def __init__(
self,
user_id: str,
session_id: Optional[str] = None,
honcho_client: Optional[Honcho] = None,
session_id: str | None = None,
honcho_client: Honcho | None = None,
assistant_id: str = "assistant",
) -> None:
"""
Initialize Honcho storage for a specific user and session.
Args:
user_id: Unique identifier for the user
session_id: Optional session ID. If not provided, a UUID will be generated
honcho_client: Optional Honcho client instance. If not provided, creates one
using the demo environment (https://demo.honcho.dev)
"""
self.honcho = honcho_client or Honcho()
self.user_id = user_id
self.assistant_id = assistant_id
self.session_id = session_id or str(uuid.uuid4())
self._user: Any | None = None
self._assistant: Any | None = None
self._session: Any | None = None
# Initialize user and assistant peers
self.user = self.honcho.peer(user_id)
self.assistant = self.honcho.peer("assistant")
@property
def user(self) -> Any:
if self._user is None:
self._user = self.honcho.peer(self.user_id)
return self._user
# Create or use existing session
if not session_id:
session_id = str(uuid.uuid4())
self.session = self.honcho.session(session_id)
self.session_id = session_id
@property
def assistant(self) -> Any:
if self._assistant is None:
self._assistant = self.honcho.peer(self.assistant_id)
return self._assistant
@property
def session(self) -> Any:
if self._session is None:
self._session = self.honcho.session(self.session_id)
return self._session
def save(self, value: Any, metadata: dict[str, Any]) -> None:
"""
Save a message to Honcho session.
This method is called by CrewAI to store messages and context. Messages
are associated with the appropriate peer (user or assistant) based on
the metadata.
Args:
value: Message content to save
metadata: Metadata dict that may contain 'role', 'agent', or 'type' info
Common keys: 'role', 'agent', 'type'
"""
"""Save a CrewAI external-memory message to a Honcho session."""
try:
# Determine if this is from user or assistant based on metadata
# Check various metadata keys that might indicate the role
role = metadata.get("role", metadata.get("agent", "assistant"))
is_user = role == "user"
peer = self.user if is_user else self.assistant
content_str = str(value)
# Add message to session
self.session.add_messages([peer.message(content_str, metadata=metadata)])
logger.debug(
"Saved message from %s: %s...",
metadata.get("name", role),
content_str[:100],
role = str(metadata.get("role", metadata.get("agent", "assistant"))).lower()
peer = (
self.user if role in {"user", "human", self.user_id} else self.assistant
)
content = str(value)
self.session.add_messages([peer.message(content, metadata=metadata)])
logger.debug("Saved %s message to Honcho session %s", role, self.session_id)
except Exception:
logger.exception("Error saving to Honcho")
@ -109,62 +462,34 @@ class HonchoStorage(Storage):
query: str,
limit: int = 10,
score_threshold: float = 0.5,
filters: Optional[dict[str, Any]] = None,
filters: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
"""
Search for relevant messages using semantic search.
This method uses Honcho's semantic vector search to find messages most
relevant to the query.
Args:
query: Search query used for semantic matching
limit: Maximum number of messages to retrieve
score_threshold: Minimum relevance score (not currently used by Honcho API)
filters: Optional filters to scope the search. Supports Honcho's filter syntax
including logical operators (AND, OR, NOT), comparison operators
(gt, gte, lt, lte, eq, ne), and metadata filtering.
Example: {"peer_id": "user123"} or {"metadata": {"type": "important"}}
See: https://docs.honcho.dev/v3/documentation/core-concepts/features/using-filters
Returns:
List of message dictionaries in CrewAI expected format.
Each dict contains:
- content: The message content
- memory: The message content (required by CrewAI)
- context: The message content (for compatibility)
- metadata: Message metadata including peer_id, created_at, and custom metadata
"""
"""Search session messages and return CrewAI external-memory records."""
try:
results = []
# Use semantic search to find relevant messages
# This performs vector similarity search on message content
_ = score_threshold
messages = self.session.search(query=query, filters=filters, limit=limit)
results = []
# Convert to CrewAI expected format
for msg in messages:
# Build base metadata with peer_id and created_at
for message in messages:
metadata = {
"peer_id": msg.peer_id,
"created_at": str(msg.created_at)
if hasattr(msg, "created_at")
"peer_id": message.peer_id,
"created_at": str(message.created_at)
if hasattr(message, "created_at")
else None,
}
# Merge custom metadata if present
if hasattr(msg, "metadata") and msg.metadata:
metadata.update(msg.metadata)
if getattr(message, "metadata", None):
metadata.update(message.metadata)
results.append(
{
"content": msg.content,
"memory": msg.content,
"context": msg.content,
"content": message.content,
"memory": message.content,
"context": message.content,
"metadata": metadata,
}
)
logger.debug("Search for '%s' returned %d results", query, len(results))
logger.debug("Search for %r returned %d results", query, len(results))
return results
except Exception:
@ -172,19 +497,7 @@ class HonchoStorage(Storage):
raise
def reset(self) -> None:
"""
Create a new session, effectively resetting memory.
This creates a new Honcho session with a fresh UUID, allowing the agent
to start a new conversation without the previous context.
"""
try:
new_session_id = str(uuid.uuid4())
self.session = self.honcho.session(new_session_id)
self.session_id = new_session_id
logger.debug("Reset session. New session ID: %s", new_session_id)
except Exception:
logger.exception("Error resetting Honcho session")
raise
"""Start writing to a fresh Honcho session."""
self.session_id = str(uuid.uuid4())
self._session = None
logger.debug("Reset HonchoStorage to session %s", self.session_id)

87
examples/crewai/python/src/honcho_crewai/tools.py
View File

@ -6,7 +6,7 @@ session context, dialectic API, and semantic search capabilities.
"""
import logging
from typing import Any, Optional
from typing import Any
from crewai.tools import BaseTool
from honcho import Honcho
@ -19,38 +19,49 @@ logger = logging.getLogger(__name__)
class GetContextInput(BaseModel):
"""Input schema for context tool."""
tokens: Optional[int] = Field(
default=None, gt=0, description="Maximum number of tokens to include in the context"
tokens: int | None = Field(
default=None,
gt=0,
description="Maximum number of tokens to include in the context",
)
peer_target: Optional[str] = Field(
default=None, description="A peer ID to get context for (retrieves representation and peer card)"
peer_target: str | None = Field(
default=None,
description="A peer ID to get context for (retrieves representation and peer card)",
)
summary: bool = Field(
default=True, description="Whether to include session summary in the context"
)
peer_perspective: Optional[str] = Field(
default=None, description="Peer ID to use as the perspective for context retrieval"
peer_perspective: str | None = Field(
default=None,
description="Peer ID to use as the perspective for context retrieval",
)
class DialecticInput(BaseModel):
"""Input schema for dialectic (chat) tool."""
query: str = Field(..., min_length=1, description="Natural language question to ask")
target: Optional[str] = Field(
query: str = Field(
..., min_length=1, description="Natural language question to ask"
)
target: str | None = Field(
default=None, description="Optional target peer for local representation query"
)
session_id: Optional[str] = Field(
default=None, description="Optional session ID to scope query to specific session"
session_id: str | None = Field(
default=None,
description="Optional session ID to scope query to specific session",
)
class SearchInput(BaseModel):
"""Input schema for search tool."""
query: str = Field(..., min_length=1, description="Search query for semantic matching")
limit: int = Field(default=10, ge=1, le=100, description="Number of results to return (1-100)")
filters: Optional[dict[str, Any]] = Field(
query: str = Field(
..., min_length=1, description="Search query for semantic matching"
)
limit: int = Field(
default=10, ge=1, le=100, description="Number of results to return (1-100)"
)
filters: dict[str, Any] | None = Field(
default=None,
description=(
"Optional filters to scope the search. Supports Honcho's filter syntax including "
@ -81,6 +92,7 @@ class HonchoGetContextTool(BaseTool):
_honcho: Honcho = PrivateAttr()
_session_id: str = PrivateAttr()
_peer_id: str = PrivateAttr()
_session: Any = PrivateAttr(default=None)
def __init__(self, honcho: Honcho, session_id: str, peer_id: str) -> None:
"""
@ -96,13 +108,19 @@ class HonchoGetContextTool(BaseTool):
self._session_id = session_id
self._peer_id = peer_id
@property
def _honcho_session(self) -> Any:
if self._session is None:
self._session = self._honcho.session(self._session_id)
return self._session
def _run(
self,
tokens: Optional[int] = None,
peer_target: Optional[str] = None,
tokens: int | None = None,
peer_target: str | None = None,
*,
summary: bool = True,
peer_perspective: Optional[str] = None,
peer_perspective: str | None = None,
) -> str:
"""
Execute context retrieval and format results.
@ -117,8 +135,7 @@ class HonchoGetContextTool(BaseTool):
Formatted string containing context information
"""
try:
session = self._honcho.session(self._session_id)
context = session.context(
context = self._honcho_session.context(
summary=summary,
tokens=tokens,
peer_target=peer_target,
@ -179,6 +196,7 @@ class HonchoDialecticTool(BaseTool):
_honcho: Honcho = PrivateAttr()
_session_id: str = PrivateAttr()
_peer_id: str = PrivateAttr()
_peer: Any = PrivateAttr(default=None)
def __init__(self, honcho: Honcho, session_id: str, peer_id: str) -> None:
"""
@ -194,11 +212,17 @@ class HonchoDialecticTool(BaseTool):
self._session_id = session_id
self._peer_id = peer_id
@property
def _honcho_peer(self) -> Any:
if self._peer is None:
self._peer = self._honcho.peer(self._peer_id)
return self._peer
def _run(
self,
query: str,
target: Optional[str] = None,
session_id: Optional[str] = None,
target: str | None = None,
session_id: str | None = None,
) -> str:
"""
Execute dialectic query.
@ -212,13 +236,11 @@ class HonchoDialecticTool(BaseTool):
String response from the dialectic API
"""
try:
peer = self._honcho.peer(self._peer_id)
# Use provided session_id or fall back to default
scope_session_id = session_id or self._session_id
# Query the dialectic API (non-streaming)
response = peer.chat(
response = self._honcho_peer.chat(
query=query,
target=target,
session=scope_session_id,
@ -254,6 +276,7 @@ class HonchoSearchTool(BaseTool):
_honcho: Honcho = PrivateAttr()
_session_id: str = PrivateAttr()
_session: Any = PrivateAttr(default=None)
def __init__(self, honcho: Honcho, session_id: str) -> None:
"""
@ -267,7 +290,15 @@ class HonchoSearchTool(BaseTool):
self._honcho = honcho
self._session_id = session_id
def _run(self, query: str, limit: int = 10, filters: Optional[dict[str, Any]] = None) -> str:
@property
def _honcho_session(self) -> Any:
if self._session is None:
self._session = self._honcho.session(self._session_id)
return self._session
def _run(
self, query: str, limit: int = 10, filters: dict[str, Any] | None = None
) -> str:
"""
Execute semantic search.
@ -280,10 +311,10 @@ class HonchoSearchTool(BaseTool):
Formatted string with search results
"""
try:
session = self._honcho.session(self._session_id)
# Perform semantic search
messages = session.search(query=query, limit=limit, filters=filters)
messages = self._honcho_session.search(
query=query, limit=limit, filters=filters
)
if not messages:
return f"No messages found matching '{query}'"

12
examples/crewai/python/tests/test_basic.py
View File

@ -4,8 +4,6 @@ Basic tests for honcho_crewai package
Validates package structure, imports, and basic functionality.
"""
import pytest
def test_package_import():
"""Test that honcho_crewai imports successfully."""
@ -21,11 +19,18 @@ def test_storage_import():
assert HonchoStorage is not None
def test_memory_storage_import():
"""Test that HonchoMemoryStorage can be imported."""
from honcho_crewai import HonchoMemoryStorage
assert HonchoMemoryStorage is not None
def test_tools_import():
"""Test that tool classes can be imported."""
from honcho_crewai import (
HonchoGetContextTool,
HonchoDialecticTool,
HonchoGetContextTool,
HonchoSearchTool,
)
@ -52,6 +57,7 @@ class TestPackageMetadata:
assert hasattr(honcho_crewai, "__all__")
expected_exports = [
"HonchoStorage",
"HonchoMemoryStorage",
"HonchoGetContextTool",
"HonchoDialecticTool",
"HonchoSearchTool",

394
examples/crewai/python/tests/test_storage.py
View File

@ -1,174 +1,256 @@
"""
Tests for HonchoStorage
Tests the CrewAI-Honcho integration layer, focusing on:
- CrewAI Storage interface compliance
- Metadata mapping (agent/role -> peer_id)
- Format conversion (Honcho -> CrewAI format)
Tests for Honcho CrewAI storage adapters.
"""
from honcho_crewai import HonchoStorage
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from crewai.memory.types import MemoryRecord
from honcho_crewai import HonchoMemoryStorage, HonchoStorage
class FakeMessageCreate:
def __init__(self, peer_id, content, metadata=None, created_at=None):
self.peer_id = peer_id
self.content = content
self.metadata = metadata or {}
self.created_at = created_at
class FakeMessage:
def __init__(self, id, peer_id, content, metadata=None, created_at=None):
self.id = id
self.peer_id = peer_id
self.content = content
self.metadata = metadata or {}
self.created_at = created_at or datetime.now(UTC)
class FakePeer:
def __init__(self, id):
self.id = id
def message(self, content, *, metadata=None, created_at=None):
return FakeMessageCreate(self.id, content, metadata, created_at)
def chat(self, query, **kwargs):
return f"answer: {query} {kwargs}"
class FakeSession:
def __init__(self, id):
self.id = id
self._messages = []
def add_messages(self, messages):
saved = []
for message in messages:
saved_message = FakeMessage(
id=f"msg-{len(self._messages) + 1}",
peer_id=message.peer_id,
content=message.content,
metadata=dict(message.metadata),
created_at=message.created_at,
)
self._messages.append(saved_message)
saved.append(saved_message)
return saved
def search(self, query, filters=None, limit=10):
return self._messages[:limit]
def messages(self, filters=None, size=100, reverse=False):
messages = list(self._messages)
if reverse:
messages.reverse()
return messages
def update_message(self, message, metadata):
message.metadata = metadata
return message
def context(self, **kwargs):
return type(
"FakeContext",
(),
{
"summary": None,
"peer_representation": None,
"peer_card": None,
"messages": self._messages,
},
)()
class FakeHoncho:
def __init__(self):
self.peer_calls = []
self.session_calls = []
self._peers = {}
self._sessions = {}
def peer(self, id):
self.peer_calls.append(id)
self._peers.setdefault(id, FakePeer(id))
return self._peers[id]
def session(self, id):
self.session_calls.append(id)
self._sessions.setdefault(id, FakeSession(id))
return self._sessions[id]
class TestHonchoMemoryStorage:
def test_initialization_is_lazy(self):
honcho = FakeHoncho()
storage = HonchoMemoryStorage(
peer_id="user-1",
session_id="session-1",
honcho_client=honcho,
)
assert storage.session_id == "session-1"
assert storage.peer_id == "user-1"
assert honcho.peer_calls == []
assert honcho.session_calls == []
def test_save_and_search_memory_records(self):
honcho = FakeHoncho()
storage = HonchoMemoryStorage(
peer_id="user-1",
session_id="session-1",
honcho_client=honcho,
)
record = MemoryRecord(
id="record-1",
content="User likes ramen",
scope="/users/user-1",
categories=["preferences"],
metadata={"topic": "food"},
embedding=[1.0, 0.0],
created_at=datetime.now(UTC),
last_accessed=datetime.now(UTC),
)
storage.save([record])
matches = storage.search(
[1.0, 0.0],
scope_prefix="/users",
categories=["preferences"],
metadata_filter={"topic": "food"},
)
assert len(matches) == 1
assert matches[0][0].id == "record-1"
assert matches[0][1] == 1.0
assert honcho.peer_calls == ["user-1"]
assert honcho.session_calls == ["session-1"]
def test_delete_update_and_discovery_methods(self):
honcho = FakeHoncho()
storage = HonchoMemoryStorage(
peer_id="user-1",
session_id="session-1",
honcho_client=honcho,
)
old_record = MemoryRecord(
id="record-1",
content="Old preference",
scope="/users/user-1/preferences",
categories=["preferences"],
metadata={"topic": "food"},
embedding=[1.0, 0.0],
created_at=datetime.now(UTC) - timedelta(days=1),
last_accessed=datetime.now(UTC) - timedelta(days=1),
)
new_record = MemoryRecord(
id="record-1",
content="New preference",
scope="/users/user-1/preferences",
categories=["preferences"],
metadata={"topic": "food"},
embedding=[0.0, 1.0],
created_at=datetime.now(UTC),
last_accessed=datetime.now(UTC),
)
storage.save([old_record])
storage.update(new_record)
assert storage.get_record("record-1").content == "New preference"
assert storage.count("/users") == 1
assert storage.list_categories("/users") == {"preferences": 1}
assert storage.list_scopes("/") == ["/users"]
assert storage.get_scope_info("/users").record_count == 1
assert storage.delete(record_ids=["record-1"]) == 1
assert storage.get_record("record-1") is None
class TestHonchoStorage:
"""Tests for HonchoStorage integration layer."""
def test_legacy_initialization_is_lazy(self):
honcho = FakeHoncho()
def test_initialization(self):
"""Test that HonchoStorage initializes with correct peers and session."""
storage = HonchoStorage(user_id="test_user")
storage = HonchoStorage(
user_id="user-1",
session_id="session-1",
honcho_client=honcho,
)
assert storage is not None
assert storage.session_id is not None
assert storage.user is not None
assert storage.assistant is not None
assert storage.session is not None
assert storage.session_id == "session-1"
assert honcho.peer_calls == []
assert honcho.session_calls == []
def test_initialization_with_custom_session(self):
"""Test that custom session_id is preserved."""
custom_session_id = "my_custom_session"
storage = HonchoStorage(user_id="test_user", session_id=custom_session_id)
def test_legacy_save_maps_roles_to_peers(self):
honcho = FakeHoncho()
storage = HonchoStorage(
user_id="user-1",
session_id="session-1",
honcho_client=honcho,
)
assert storage.session_id == custom_session_id
storage.save("User message", metadata={"role": "user"})
storage.save("Assistant message", metadata={"role": "assistant"})
def test_save_with_different_roles(self):
"""Test that save handles different agent/role metadata."""
storage = HonchoStorage(user_id="test_user_roles")
messages = honcho._sessions["session-1"]._messages
assert [message.peer_id for message in messages] == ["user-1", "assistant"]
# Save with different metadata patterns
storage.save("User via agent", metadata={"agent": "user"})
storage.save("User via role", metadata={"role": "user"})
storage.save("Assistant via agent", metadata={"agent": "assistant"})
storage.save("Default (no metadata)", metadata={})
def test_legacy_search_returns_crewai_external_memory_format(self):
honcho = FakeHoncho()
storage = HonchoStorage(
user_id="user-1",
session_id="session-1",
honcho_client=honcho,
)
# If no exceptions raised, metadata mapping works
storage.save("User likes ramen", metadata={"role": "user", "topic": "food"})
results = storage.search("ramen")
def test_search_returns_crewai_format(self):
"""Test that search returns results in CrewAI format."""
storage = HonchoStorage(user_id="test_user_search")
assert results == [
{
"content": "User likes ramen",
"memory": "User likes ramen",
"context": "User likes ramen",
"metadata": {
"peer_id": "user-1",
"created_at": str(results[0]["metadata"]["created_at"]),
"role": "user",
"topic": "food",
},
}
]
# Add a message
storage.save("Test message", metadata={"agent": "user"})
def test_legacy_reset_is_lazy(self):
honcho = FakeHoncho()
storage = HonchoStorage(
user_id="user-1",
session_id="session-1",
honcho_client=honcho,
)
# Search
results = storage.search("test", limit=10)
# Verify CrewAI format
assert isinstance(results, list)
for result in results:
# Required keys for CrewAI
assert "memory" in result
assert "context" in result
assert "content" in result
assert "metadata" in result
def test_search_includes_all_required_fields(self):
"""Test that all search results have required CrewAI fields."""
storage = HonchoStorage(user_id="test_user_format")
# Add a message
storage.save("Test message", metadata={"agent": "user"})
# Search
results = storage.search("test", limit=5)
# Verify all results have required fields with correct types
for result in results:
assert isinstance(result["content"], str)
assert isinstance(result["memory"], str)
assert isinstance(result["context"], str)
assert isinstance(result["metadata"], dict)
def test_search_formats_summaries_correctly(self):
"""Test that session summaries are formatted with [Session Summary] prefix."""
storage = HonchoStorage(user_id="test_user_summaries")
# Add enough messages to potentially trigger summaries
for i in range(25):
storage.save(
f"Message {i}",
metadata={"agent": "user" if i % 2 == 0 else "assistant"},
)
# Search
results = storage.search("message", limit=10)
# Check summary formatting (if summaries exist)
summary_results = [r for r in results if r["metadata"].get("type") == "summary"]
for summary in summary_results:
# Verify our formatting logic
assert "summary_type" in summary["metadata"]
assert "[Session Summary]" in summary["context"] # Our formatting
def test_reset_creates_new_session_id(self):
"""Test that reset() creates a new session with different ID."""
storage = HonchoStorage(user_id="test_user_reset")
original_session_id = storage.session_id
# Reset
storage.reset()
# Verify new session ID was created
assert storage.session_id != original_session_id
def test_search_with_filters(self):
"""Test that search accepts and uses filters parameter."""
storage = HonchoStorage(user_id="test_user_filters")
# Add messages with different metadata
storage.save("User question about Python", metadata={"agent": "user", "topic": "python"})
storage.save("Assistant answer about Python", metadata={"agent": "assistant", "topic": "python"})
storage.save("User question about JavaScript", metadata={"agent": "user", "topic": "javascript"})
# Search with peer_id filter - filter to only user messages
results = storage.search(
"programming",
limit=10,
filters={"peer_id": storage.user.id}
)
# Verify results are returned and in correct format
assert isinstance(results, list)
for result in results:
assert "memory" in result
assert "content" in result
assert "context" in result
assert "metadata" in result
def test_search_with_metadata_filters(self):
"""Test that search works with metadata filters."""
storage = HonchoStorage(user_id="test_user_metadata_filters")
# Add messages with specific metadata
storage.save("Important message", metadata={"agent": "user", "priority": "high"})
storage.save("Regular message", metadata={"agent": "user", "priority": "low"})
# Search with metadata filter
results = storage.search(
"message",
limit=10,
filters={"metadata": {"priority": "high"}}
)
# Verify results are in correct format
assert isinstance(results, list)
for result in results:
assert "memory" in result
assert "metadata" in result
def test_search_without_filters(self):
"""Test that search works without filters."""
storage = HonchoStorage(user_id="test_user_no_filters")
# Add a message
storage.save("Test message for search", metadata={"agent": "user"})
# Search without filters
results = storage.search("test", limit=5)
# Verify it works and returns correct format
assert isinstance(results, list)
for result in results:
assert "memory" in result
assert "content" in result
assert storage.session_id != "session-1"
assert honcho.session_calls == []

189
examples/crewai/python/tests/test_tools.py
View File

@ -1,194 +1,117 @@
"""
Tests for Honcho CrewAI Tools
Tests the CrewAI-Honcho tool integration layer using real Honcho SDK.
Focuses on tool interface compliance and result formatting.
Tests for Honcho CrewAI tools.
"""
from honcho import Honcho
from honcho_crewai import (
HonchoDialecticTool,
HonchoGetContextTool,
HonchoSearchTool,
)
from test_storage import FakeHoncho
class TestGetContextTool:
"""Tests for HonchoGetContextTool."""
def test_initialization_is_lazy(self):
honcho = FakeHoncho()
def test_initialization(self):
"""Test that tool initializes with correct attributes."""
honcho = Honcho()
tool = HonchoGetContextTool(
honcho=honcho, session_id="test_session", peer_id="test_peer"
honcho=honcho,
session_id="session-1",
peer_id="user-1",
)
assert tool is not None
assert tool.name == "get_session_context"
assert tool.description is not None
assert tool.args_schema is not None
assert honcho.session_calls == []
def test_returns_formatted_context(self):
"""Test that tool returns formatted context string."""
honcho = Honcho()
peer = honcho.peer("context_test_user")
session_id = "context_test_session"
session = honcho.session(session_id)
# Add test message
honcho = FakeHoncho()
peer = honcho.peer("user-1")
session = honcho.session("session-1")
session.add_messages([peer.message("Test message for context")])
# Create and execute tool
honcho.session_calls.clear()
tool = HonchoGetContextTool(
honcho=honcho, session_id=session_id, peer_id="context_test_user"
honcho=honcho,
session_id="session-1",
peer_id="user-1",
)
result = tool._run()
# Verify result is a formatted string
assert isinstance(result, str)
assert len(result) > 0
assert "Messages (1)" in result
assert "user-1: Test message for context" in result
assert honcho.session_calls == ["session-1"]
class TestDialecticTool:
"""Tests for HonchoDialecticTool."""
def test_initialization_is_lazy(self):
honcho = FakeHoncho()
def test_initialization(self):
"""Test that tool initializes with correct attributes."""
honcho = Honcho()
tool = HonchoDialecticTool(
honcho=honcho, session_id="test_session", peer_id="test_peer"
honcho=honcho,
session_id="session-1",
peer_id="user-1",
)
assert tool is not None
assert tool.name == "query_peer_knowledge"
assert tool.description is not None
assert honcho.peer_calls == []
def test_returns_response(self):
"""Test that tool returns a response string."""
honcho = Honcho()
peer = honcho.peer("dialectic_test_user")
session_id = "dialectic_test_session"
session = honcho.session(session_id)
# Add test messages
session.add_messages([peer.message("I love pizza and Italian food")])
# Create and execute tool
honcho = FakeHoncho()
tool = HonchoDialecticTool(
honcho=honcho, session_id=session_id, peer_id="dialectic_test_user"
honcho=honcho,
session_id="session-1",
peer_id="user-1",
)
result = tool._run(query="What does the user like?")
# Verify result is a string
assert isinstance(result, str)
assert len(result) > 0
assert "answer: What does the user like?" in result
assert honcho.peer_calls == ["user-1"]
class TestSearchTool:
"""Tests for HonchoSearchTool."""
def test_initialization_is_lazy(self):
honcho = FakeHoncho()
def test_initialization(self):
"""Test that tool initializes with correct attributes."""
honcho = Honcho()
tool = HonchoSearchTool(honcho=honcho, session_id="test_session")
tool = HonchoSearchTool(honcho=honcho, session_id="session-1")
assert tool is not None
assert tool.name == "search_session_messages"
assert tool.description is not None
assert honcho.session_calls == []
def test_returns_formatted_results(self):
"""Test that tool returns formatted search results."""
honcho = Honcho()
peer = honcho.peer("search_test_user")
session_id = "search_test_session"
session = honcho.session(session_id)
# Add test messages
honcho = FakeHoncho()
peer = honcho.peer("user-1")
session = honcho.session("session-1")
session.add_messages([peer.message("I love pizza and pasta")])
# Create and execute tool
tool = HonchoSearchTool(honcho=honcho, session_id=session_id)
honcho.session_calls.clear()
tool = HonchoSearchTool(honcho=honcho, session_id="session-1")
result = tool._run(query="food", limit=5)
# Verify result is a formatted string
assert isinstance(result, str)
assert len(result) > 0
# Should have either results or "No messages found"
assert "Search Results" in result or "No messages found" in result
def test_search_with_filters(self):
"""Test that search tool accepts and uses filters parameter."""
honcho = Honcho()
peer = honcho.peer("search_filter_test_user")
session_id = "search_filter_test_session"
session = honcho.session(session_id)
# Add test messages
session.add_messages([peer.message("Important message about Python")])
# Create and execute tool with filters
tool = HonchoSearchTool(honcho=honcho, session_id=session_id)
result = tool._run(
query="Python",
limit=5,
filters={"peer_id": peer.id}
)
# Verify result is a formatted string
assert isinstance(result, str)
assert len(result) > 0
def test_search_with_metadata_filters(self):
"""Test that search tool works with metadata filters."""
honcho = Honcho()
peer = honcho.peer("search_metadata_filter_user")
session_id = "search_metadata_filter_session"
session = honcho.session(session_id)
# Add test messages with metadata
session.add_messages([peer.message("High priority task", metadata={"priority": "high"})])
# Create and execute tool with metadata filter
tool = HonchoSearchTool(honcho=honcho, session_id=session_id)
result = tool._run(
query="task",
limit=5,
filters={"metadata": {"priority": "high"}}
)
# Verify result is a formatted string
assert isinstance(result, str)
assert len(result) > 0
assert "Search Results" in result
assert "[user-1] I love pizza and pasta" in result
assert honcho.session_calls == ["session-1"]
class TestToolsWorkTogether:
"""Test that all tools can work together."""
def test_all_tools_in_same_session(self):
"""Test that all three tools can be used in the same session."""
honcho = Honcho()
peer = honcho.peer("combo_test_user")
session_id = "combo_test_session"
session = honcho.session(session_id)
# Add messages
honcho = FakeHoncho()
peer = honcho.peer("user-1")
session = honcho.session("session-1")
session.add_messages([peer.message("I enjoy coding in Python")])
# Create all tools
context_tool = HonchoGetContextTool(
honcho=honcho, session_id=session_id, peer_id="combo_test_user"
honcho=honcho,
session_id="session-1",
peer_id="user-1",
)
dialectic_tool = HonchoDialecticTool(
honcho=honcho, session_id=session_id, peer_id="combo_test_user"
honcho=honcho,
session_id="session-1",
peer_id="user-1",
)
search_tool = HonchoSearchTool(honcho=honcho, session_id=session_id)
search_tool = HonchoSearchTool(honcho=honcho, session_id="session-1")
# Execute all tools
context_result = context_tool._run()
dialectic_result = dialectic_tool._run(query="What does the user like?")
search_result = search_tool._run(query="coding", limit=5)
# Verify all return valid strings
assert isinstance(context_result, str) and len(context_result) > 0
assert isinstance(dialectic_result, str) and len(dialectic_result) > 0
assert isinstance(search_result, str) and len(search_result) > 0
assert context_tool._run()
assert dialectic_tool._run(query="What does the user like?")
assert search_tool._run(query="coding", limit=5)

1078
examples/crewai/python/uv.lock
View File

File diff suppressed because it is too large Load Diff