CrewAI integration

1
from zep_crewai import ensure_user, ensure_thread
2
3
def setup_new_user(client, user_id):
4
    client.graph.set_ontology(...)  # one-time per-user configuration
5
6
ensure_user(zep_client, user_id="alice_123", first_name="Alice", on_created=setup_new_user)
7
ensure_thread(zep_client, thread_id="project_456", user_id="alice_123")

1
import os
2
from zep_cloud.client import Zep
3
from zep_crewai import ZepUserStorage, create_search_tool, ensure_user, ensure_thread
4
from crewai import Agent
5
6
zep_client = Zep(api_key=os.getenv("ZEP_API_KEY"))
7
8
# Provision the user and thread (idempotent; genuine failures raise)
9
ensure_user(zep_client, user_id="alice_123", first_name="Alice")
10
ensure_thread(zep_client, thread_id="project_456", user_id="alice_123")
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", ""))  # <ZEP_CONTEXT> ... </ZEP_CONTEXT>

1
from zep_crewai import ZepUserStorage, ContextInput
2
3
def my_builder(ctx: ContextInput) -> str | None:
4
    results = ctx.zep.graph.search(user_id=ctx.user_id, query=ctx.user_message, scope="edges")
5
    if not results.edges:
6
        return None
7
    return "\n".join(edge.fact for edge in results.edges)
8
9
storage = ZepUserStorage(
10
    client=zep_client,
11
    user_id="alice_123",
12
    thread_id="project_456",
13
    context_builder=my_builder,
14
)

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

1
# Pin scope and limit (hidden from the model, always sent); hide reranker entirely
2
search_tool = create_search_tool(
3
    zep_client,
4
    user_id="alice_123",
5
    pinned_params={"scope": "edges", "limit": 5},
6
    hidden_params={"reranker"},
7
)
8
9
# Constructor-only parameters are never exposed to the model
10
search_tool = create_search_tool(
11
    zep_client,
12
    graph_id="knowledge_base",
13
    search_filters=SearchFilters(node_labels=["Project"]),
14
    bfs_origin_node_uuids=["node-uuid-1"],
15
)

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