Zep Python SDK

repository·main·Indexed 41 days ago

https://github.com/getzep/zep-python

A context engineering platform for AI assistants that manages long-term memory through chat history persistence, summarization, and semantic search. The SDK provides tools for document collections, fact extraction, dialog classification, and structured data retrieval. It includes support for Zep Cloud, Zep Community Edition, and LangChain integration via ZepChatMessageHistory and ZepVectorStore.

Tokens
37K
Snippets
138
Records
158
Agent score
53%

What's inside Zep

  1. Manage UserGroups with `client.user_group`

    main
    The user_group namespace in the Zep client allows you to manage groups of users, their memberships, and associated policy sets within a project. You can create, update, delete, and list UserGroups, as well as manage which users belong to which group and which policy sets are attached to them.
  2. How Zep works: Context Engineering and Document Collections

    main

    Zep is a context engineering platform designed to provide AI assistants with long-term memory.

    Core Capabilities

    • Chat History Management: Persists and recalls chat histories, automatically generating summaries and artifacts asynchronously to avoid impacting latency.
    • Searchable Context: Embeds messages and summaries, allowing for semantic search of relevant context from past conversations.
    • Document Collections: A specialized abstraction for document vector search designed to complement core context features (not intended as a general-purpose vector database).

    Prompt Construction Patterns

    Zep enables intentional prompt construction by allowing you to include:

    1. Recent messages (customizable count).
    2. Summaries of conversations preceding the recent messages.
    3. Contextually relevant summaries or messages surfaced from the entire session.
    4. Relevant business data from Zep Document Collections.

    Zep Cloud Specific Features

    • Fact Extraction: Automatically builds fact tables from conversations without upfront schema definition.
    • Dialog Classification: Classifies chat dialog to understand intent, emotion, and user segments for routing or event triggering.
    • Structured Data Extraction: Extracts business data from conversations based on a user-defined schema.
  3. Configure graph ingestion with ignore_roles

    main

    When creating a batch using create(), you can provide an ignore_roles parameter. This is a sequence of RoleType values.

    If a thread_message item in the batch has a role included in ignore_roles, Zep will still store and retain the message as context, but it will skip graph extraction for that specific message. This parameter has no effect on graph_episode items.

  4. Manage graph nodes with NodeClient

    main

    The NodeClient (and its asynchronous counterpart AsyncNodeClient) provides methods to interact with entity nodes within a Zep graph. You can retrieve nodes by graph ID or user ID, fetch specific nodes by UUID, update node properties, or delete them.

    Both clients support a .with_raw_response property to access the underlying RawNodeClient for direct response handling.

    from zep_cloud import Zep
    
    client = Zep(api_key="YOUR_API_KEY")
    # Access node management via client.graph.node
    node = client.graph.node.get(uuid_="node_uuid")
  5. Manage observation steering (Experimental)

    main

    Observation steering allows you to provide instructions or specify observation types for how Zep processes data. This API is experimental and may change.

    Get Steering Configuration

    Use get_observation_steering() to retrieve the current project steering or the effective user/graph steering with project fallback.

    Parameters:

    • user_id: Optional ID for user-specific steering.
    • graph_id: Optional ID for graph-specific steering.
    • request_options: Optional RequestOptions.

    Set Steering Configuration

    Use set_observation_steering() to replace project, user, or graph steering. An empty configuration clears the project default or removes the user/graph override. Changes affect later materializer runs only.

    Parameters:

    • user_id: Optional ID for user-specific steering.
    • graph_id: Optional ID for graph-specific steering.
    • instruction: Optional string containing steering instructions.
    • types: Optional sequence of ObservationType.
    • request_options: Optional RequestOptions.

    Returns:

    • ObservationSteeringConfig object.
    from zep_cloud import Zep
    
    client = Zep(api_key="YOUR_API_KEY")
    
    # Get steering for a specific user
    steering = client.project.get_observation_steering(user_id="user_123")
    
    # Set steering instructions
    client.project.set_observation_steering(
        user_id="user_123",
        instruction="Focus on extracting technical entities",
        types=["ENTITY_EXTRACTION"]
    )
  6. Manage graph episodes with EpisodeClient

    main

    The EpisodeClient (sync) and AsyncEpisodeClient (async) allow you to manage episodes within a Zep graph. You can retrieve episodes by graph ID, user ID, or specific UUID, as well as list, update, or delete them.

    Key operations include:

    • Retrieving episodes: Use get_by_graph_id, get_by_user_id, or get (by UUID).
    • Listing episodes: Use list_by_graph_id or list_by_user_id for paginated, filterable results.
    • Updating metadata: Use update to modify episode metadata using merge semantics.
    • Deleting episodes: Use delete to remove an episode by its UUID.
    from zep_cloud import Zep
    
    client = Zep(api_key="YOUR_API_KEY")
    
    # Sync example
    episode = client.graph.episode.get(uuid_="uuid")
    
    # Async example
    import asyncio
    from zep_cloud import AsyncZep
    
    async_client = AsyncZep(api_key="YOUR_API_KEY")
    
    async def main():
        episode = await async_client.graph.episode.get(uuid_="uuid")
    
    asyncio.run(main())
  7. Paginate edge results using cursors

    main

    When retrieving lists of edges via get_by_graph_id or get_by_user_id, you can use pagination parameters to handle large datasets.

    • cursor: An opaque string used for pagination. It should be obtained from the Zep-Next-Cursor response header of a previous page. This is the preferred method for pagination.
    • uuid_cursor: (Deprecated) A UUID-based cursor used for pagination. It should be the UUID of the last item in the previous page. Prefer using cursor instead.
    • limit: The maximum number of items to return in a single request.
    • direction: The sort direction, either "asc" or "desc" (defaults to "desc").
    • order_by: The field to sort by. Options are "created_at", "valid_at", or "uuid" (defaults to "uuid").
  8. Define custom Entity and Edge models for Zep Ontology

    main

    To define custom entities and edges in the Zep ontology, you should subclass EntityModel or EdgeModel and use the provided annotated types for field properties. These annotated types ensure that your Python models are correctly serialized into the JSON schema format expected by the Zep API.

    Supported Property Types

    When defining fields, use these specific annotated types to ensure correct type mapping (Text, Int, Float, or Boolean):

    • EntityText: For string values.
    • EntityInt: For integer values.
    • EntityFloat: For floating-point values.
    • EntityBoolean: For boolean values.

    Conversion to API Schema

    Once your models are defined, use entity_model_to_api_schema or edge_model_to_api_schema to convert your Pydantic models into the dictionary format required by the Zep API for EntityType or EntityEdge definitions.

    from zep_cloud.external_clients.ontology import EntityModel, EdgeModel, EntityText, EntityInt
    
    class UserEntity(EntityModel):
        """A representation of a user in the system"""
        user_id: EntityText
        age: EntityInt
    
    class FollowsEdge(EdgeModel):
        """Represents a follow relationship between users"""
        strength: EntityInt
    
    # Convert to API-ready schema
    user_schema = entity_model_to_api_schema(UserEntity, "User")
    edge_schema = edge_model_to_api_schema(FollowsEdge, "Follows")