CrewAI integration

1
import os
2
from zep_cloud.client import Zep
3
from zep_crewai import ZepUserStorage
4
from crewai import Agent, Crew, Task
5
from crewai.memory.external.external_memory import ExternalMemory
6
7
# Initialize Zep client
8
zep_client = Zep(api_key=os.getenv("ZEP_API_KEY"))
9
10
# Create user and thread
11
zep_client.user.add(user_id="alice_123", first_name="Alice")
12
zep_client.thread.create(user_id="alice_123", thread_id="project_456")
13
14
# Set up user storage
15
user_storage = ZepUserStorage(
16
    client=zep_client,
17
    user_id="alice_123",
18
    thread_id="project_456",
19
    mode="summary"  # or "basic" for faster retrieval
20
)
21
22
# Create crew with user memory
23
crew = Crew(
24
    agents=[your_agent],
25
    tasks=[your_task],
26
    external_memory=ExternalMemory(storage=user_storage)
27
)

1
from zep_crewai import ZepGraphStorage
2
3
# Create a graph first
4
graph = zep_client.graph.create(
5
    graph_id="company_knowledge", 
6
    name="Company Knowledge Graph",
7
    description="Shared organizational knowledge and insights."
8
)
9
10
# Create graph storage for shared knowledge
11
graph_storage = ZepGraphStorage(
12
    client=zep_client,
13
    graph_id="company_knowledge",
14
    search_filters={"node_labels": ["Technology", "Project"]}
15
)
16
17
# Create crew with graph memory
18
crew = Crew(
19
    agents=[your_agent],
20
    tasks=[your_task],
21
    external_memory=ExternalMemory(storage=graph_storage)
22
)

1
from zep_crewai import create_search_tool, create_add_data_tool
2
3
# Create tools for user storage
4
user_search_tool = create_search_tool(zep_client, user_id="alice_123")
5
user_add_tool = create_add_data_tool(zep_client, user_id="alice_123")
6
7
# Create tools for graph storage
8
graph_search_tool = create_search_tool(zep_client, graph_id="knowledge_base")
9
graph_add_tool = create_add_data_tool(zep_client, graph_id="knowledge_base")
10
11
# Create agent with tools
12
knowledge_agent = Agent(
13
    role="Knowledge Assistant",
14
    goal="Manage and retrieve information efficiently",
15
    tools=[user_search_tool, graph_add_tool],
16
    backstory="You help maintain and search through relevant knowledge",
17
    llm="gpt-4o-mini"
18
)

1
from pydantic import Field
2
from zep_cloud.external_clients.ontology import EntityModel, EntityText
3
4
class ProjectEntity(EntityModel):
5
    status: EntityText = Field(description="project status")
6
    priority: EntityText = Field(description="priority level")
7
    team_size: EntityText = Field(description="team size")
8
9
# Set graph ontology
10
zep_client.graph.set_ontology(
11
    graph_id="projects",
12
    entities={"Project": ProjectEntity},
13
    edges={}
14
)
15
16
# Use graph with structured entities
17
graph_storage = ZepGraphStorage(
18
    client=zep_client,
19
    graph_id="projects",
20
    search_filters={"node_labels": ["Project"]},
21
    facts_limit=20,
22
    entity_limit=5
23
)

1
# Personal assistant with user-specific memory
2
personal_storage = ZepUserStorage(
3
    client=zep_client,
4
    user_id="alice_123",
5
    thread_id="conversation_456"
6
)
7
8
personal_agent = Agent(
9
    role="Personal Assistant",
10
    tools=[create_search_tool(zep_client, user_id="alice_123")],
11
    backstory="You know Alice's preferences and conversation history"
12
)
13
14
# Team coordinator with shared knowledge
15
team_storage = ZepGraphStorage(
16
    client=zep_client,
17
    graph_id="team_knowledge"
18
)
19
20
team_agent = Agent(
21
    role="Team Coordinator",
22
    tools=[create_search_tool(zep_client, graph_id="team_knowledge")],
23
    backstory="You maintain the team's shared knowledge base"
24
)
25
26
# Create crews with different storage types
27
personal_crew = Crew(
28
    agents=[personal_agent],
29
    tasks=[personal_task],
30
    external_memory=ExternalMemory(storage=personal_storage)
31
)
32
33
team_crew = Crew(
34
    agents=[team_agent],
35
    tasks=[team_task],
36
    external_memory=ExternalMemory(storage=team_storage)
37
)

1
# Research agent with search capabilities
2
researcher = Agent(
3
    role="Research Analyst",
4
    goal="Analyze information from multiple knowledge sources",
5
    tools=[
6
        create_search_tool(zep_client, user_id="alice_123"),
7
        create_search_tool(zep_client, graph_id="research_data")
8
    ],
9
    backstory="You analyze both personal and organizational data"
10
)
11
12
# Knowledge curator with write access
13
curator = Agent(
14
    role="Knowledge Curator",
15
    goal="Maintain the organization's knowledge base",
16
    tools=[
17
        create_search_tool(zep_client, graph_id="knowledge_base"),
18
        create_add_data_tool(zep_client, graph_id="knowledge_base")
19
    ],
20
    backstory="You maintain and organize company knowledge"
21
)
22
23
# Task that demonstrates search and add workflow
24
research_task = Task(
25
    description="""Research current trends in AI frameworks:
26
    1. Search existing knowledge about AI frameworks
27
    2. Identify gaps in current knowledge
28
    3. Add new insights to the knowledge base""",
29
    expected_output="Summary of research findings and new knowledge added",
30
    agent=curator
31
)

1
# Messages go to thread (if thread_id is configured)
2
external_memory.save(
3
    "How can I help you today?",
4
    metadata={"type": "message", "role": "assistant", "name": "Helper"}
5
)
6
7
# JSON data goes to graph
8
external_memory.save(
9
    '{"project": "Alpha", "status": "active", "budget": 50000}',
10
    metadata={"type": "json"}
11
)
12
13
# Text data goes to graph
14
external_memory.save(
15
    "Project Alpha requires Python and React expertise",
16
    metadata={"type": "text"}
17
)

1
import os
2
import time
3
from zep_cloud.client import Zep
4
from zep_crewai import ZepUserStorage
5
from crewai import Agent, Crew, Task, Process
6
from crewai.memory.external.external_memory import ExternalMemory
7
8
# Initialize Zep
9
zep_client = Zep(api_key=os.getenv("ZEP_API_KEY"))
10
11
# Set up user and thread
12
user_id = "alice_123"
13
thread_id = "project_planning"
14
15
zep_client.user.add(user_id=user_id, first_name="Alice")
16
zep_client.thread.create(user_id=user_id, thread_id=thread_id)
17
18
# Configure user storage
19
user_storage = ZepUserStorage(
20
    client=zep_client,
21
    user_id=user_id,
22
    thread_id=thread_id,
23
    mode="summary"
24
)
25
external_memory = ExternalMemory(storage=user_storage)
26
27
# Store conversation context
28
external_memory.save(
29
    "I need help planning our Q4 product roadmap with focus on mobile features",
30
    metadata={"type": "message", "role": "user", "name": "Alice"}
31
)
32
33
external_memory.save(
34
    '{"budget": 500000, "team_size": 12, "deadline": "Q4 2024"}',
35
    metadata={"type": "json"}
36
)
37
38
# Allow time for indexing
39
time.sleep(20)
40
41
# Create agent with memory
42
planning_agent = Agent(
43
    role="Product Planning Assistant",
44
    goal="Help create data-driven product roadmaps",
45
    backstory="You understand Alice's preferences and project context",
46
    llm="gpt-4o-mini"
47
)
48
49
planning_task = Task(
50
    description="Create a Q4 roadmap based on Alice's requirements and context",
51
    expected_output="Detailed roadmap with timeline and resource allocation",
52
    agent=planning_agent
53
)
54
55
# Execute with automatic memory retrieval
56
crew = Crew(
57
    agents=[planning_agent],
58
    tasks=[planning_task],
59
    process=Process.sequential,
60
    external_memory=external_memory
61
)
62
63
result = crew.kickoff()
64
65
# Save results back to memory
66
external_memory.save(
67
    str(result),
68
    metadata={"type": "message", "role": "assistant", "name": "Planning Agent"}
69
)