CrewAI integration

1
import os
2
from zep_cloud.client import Zep
3
from zep_crewai import ZepUserStorage, create_search_tool
4
from crewai import Agent
5
6
zep_client = Zep(api_key=os.getenv("ZEP_API_KEY"))
7
8
# Create the user and thread up front
9
zep_client.user.add(user_id="alice_123", first_name="Alice")
10
zep_client.thread.create(user_id="alice_123", thread_id="project_456")
11
12
# Create user storage
13
user_storage = ZepUserStorage(
14
    client=zep_client,
15
    user_id="alice_123",
16
    thread_id="project_456",
17
)
18
19
# Persist a conversation turn (routes to the thread)
20
user_storage.save(
21
    "How can I help you today?",
22
    metadata={"type": "message", "role": "assistant", "name": "Helper"},
23
)
24
25
# Persist a preference as graph data
26
user_storage.save(
27
    "Alice prefers morning meetings",
28
    metadata={"type": "text"},
29
)
30
31
# Give an agent a Zep search tool so it can retrieve this context on demand
32
assistant = Agent(
33
    role="Personal Assistant",
34
    goal="Help Alice using what you know about her",
35
    backstory="You know Alice's preferences and conversation history.",
36
    tools=[create_search_tool(zep_client, user_id="alice_123")],
37
)

1
# Returns a prompt-ready context block string (or None if empty)
2
context = user_storage.get_context()
3
print(context)

1
from zep_cloud import SearchFilters
2
from zep_crewai import ZepGraphStorage, create_search_tool
3
from crewai import Agent
4
5
# Create the graph
6
zep_client.graph.create(
7
    graph_id="company_knowledge",
8
    name="Company Knowledge Graph",
9
    description="Shared organizational knowledge and insights.",
10
)
11
12
# Create graph storage for shared knowledge
13
graph_storage = ZepGraphStorage(
14
    client=zep_client,
15
    graph_id="company_knowledge",
16
    search_filters=SearchFilters(node_labels=["Technology", "Project"]),
17
)
18
19
# Persist knowledge
20
graph_storage.save(
21
    "Project Alpha uses Python and React",
22
    metadata={"type": "text"},
23
)
24
25
# Let agents search it through a tool
26
knowledge_agent = Agent(
27
    role="Knowledge Assistant",
28
    goal="Answer questions from the shared knowledge graph",
29
    backstory="You maintain and search the team's shared knowledge.",
30
    tools=[create_search_tool(zep_client, graph_id="company_knowledge")],
31
)

1
results = graph_storage.search("project status", limit=5)
2
for item in results:
3
    print(item.get("context", ""))

1
from zep_crewai import create_search_tool, create_add_data_tool
2
from crewai import Agent
3
4
# Tools bound to user storage
5
user_search_tool = create_search_tool(zep_client, user_id="alice_123")
6
user_add_tool = create_add_data_tool(zep_client, user_id="alice_123")
7
8
# Tools bound to graph storage
9
graph_search_tool = create_search_tool(zep_client, graph_id="knowledge_base")
10
graph_add_tool = create_add_data_tool(zep_client, graph_id="knowledge_base")
11
12
curator = Agent(
13
    role="Knowledge Curator",
14
    goal="Search existing knowledge and record new findings",
15
    backstory="You maintain the organization's knowledge base.",
16
    tools=[graph_search_tool, graph_add_tool],
17
    llm="gpt-4o-mini",
18
)

1
from pydantic import Field
2
from zep_cloud import SearchFilters
3
from zep_cloud.external_clients.ontology import EntityModel, EntityText
4
from zep_crewai import ZepGraphStorage
5
6
class ProjectEntity(EntityModel):
7
    """A project tracked in the knowledge graph."""
8
9
    status: EntityText = Field(description="project status")
10
    priority: EntityText = Field(description="priority level")
11
    team_size: EntityText = Field(description="team size")
12
13
# Apply the ontology to one or more graphs
14
zep_client.graph.set_ontology(
15
    graph_ids=["projects"],
16
    entities={"Project": ProjectEntity},
17
    edges={},
18
)
19
20
# Use the graph with filtered search and context limits
21
graph_storage = ZepGraphStorage(
22
    client=zep_client,
23
    graph_id="projects",
24
    search_filters=SearchFilters(node_labels=["Project"]),
25
    facts_limit=20,
26
    entity_limit=5,
27
)

1
import os
2
import sys
3
import time
4
import uuid
5
6
from crewai import Agent, Crew, Process, Task
7
from zep_cloud.client import Zep
8
9
from zep_crewai import ZepUserStorage, create_search_tool
10
11
12
def main():
13
    api_key = os.environ.get("ZEP_API_KEY")
14
    if not api_key:
15
        print("Error: set your ZEP_API_KEY environment variable")
16
        print("Get your API key from: https://app.getzep.com")
17
        sys.exit(1)
18
19
    zep_client = Zep(api_key=api_key)
20
21
    # Set up a unique user and thread
22
    user_id = "demo_user_" + str(uuid.uuid4())
23
    thread_id = "demo_thread_" + str(uuid.uuid4())
24
25
    zep_client.user.add(
26
        user_id=user_id,
27
        first_name="John",
28
        last_name="Doe",
29
        email="[email protected]",
30
    )
31
    zep_client.thread.create(user_id=user_id, thread_id=thread_id)
32
33
    # Initialize the Zep storage adapter
34
    user_storage = ZepUserStorage(client=zep_client, user_id=user_id, thread_id=thread_id)
35
36
    # Persist context with metadata-based routing
37
    # JSON data routes to the graph
38
    user_storage.save(
39
        '{"trip_type": "business", "destination": "New York", "duration": "3 days", '
40
        '"budget": 2000, "accommodation_preference": "mid-range hotels"}',
41
        metadata={"type": "json"},
42
    )
43
44
    # Messages route to the thread
45
    user_storage.save(
46
        "Hi, I need help planning a business trip to New York. I'll be there for 3 "
47
        "days and prefer mid-range hotels.",
48
        metadata={"type": "message", "role": "user", "name": "John Doe"},
49
    )
50
    user_storage.save(
51
        "I'd be happy to help you plan your New York business trip!",
52
        metadata={"type": "message", "role": "assistant", "name": "Travel Planning Assistant"},
53
    )
54
55
    # Text data routes to the graph
56
    user_storage.save(
57
        "John Doe prefers mid-range hotels with business amenities, enjoys local "
58
        "cuisine, and values convenient locations near business districts.",
59
        metadata={"type": "text"},
60
    )
61
    user_storage.save(
62
        "John Doe's budget constraint: around $2000 total for the trip including "
63
        "flights and accommodation. Looking for good value rather than luxury.",
64
        metadata={"type": "text"},
65
    )
66
67
    # Allow time for indexing before the agent searches
68
    time.sleep(20)
69
70
    # Give the agent a Zep search tool bound to this user
71
    search_tool = create_search_tool(zep_client, user_id=user_id)
72
73
    travel_agent = Agent(
74
        role="Travel Planning Assistant",
75
        goal="Help plan business trips efficiently and within budget",
76
        backstory="""You are an experienced travel planner who specializes in business
77
        trips. You always consider the user's preferences, budget, and trip context.
78
        Use the Zep memory search tool to recall what you know about the user before
79
        answering.""",
80
        tools=[search_tool],
81
        verbose=True,
82
        llm="gpt-4.1-mini",
83
    )
84
85
    planning_task = Task(
86
        description="""First, search Zep memory for the user's saved preferences and
87
        trip context. Then provide 3 specific hotel recommendations in New York that
88
        would be good for a business traveler. Include hotel names and locations, price
89
        range per night, why each fits the user's preferences, and any business
90
        amenities.""",
91
        expected_output="A list of 3 hotel recommendations with detailed explanations",
92
        agent=travel_agent,
93
    )
94
95
    crew = Crew(
96
        agents=[travel_agent],
97
        tasks=[planning_task],
98
        process=Process.sequential,
99
        verbose=True,
100
    )
101
102
    result = crew.kickoff()
103
    print(result)
104
105
    # Optionally persist the result for future runs
106
    user_storage.save(str(result), metadata={"type": "message", "role": "assistant"})
107
108
109
if __name__ == "__main__":
110
    main()