client.py

# This file was auto-generated by Fern from our API Definition.

import typing

from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper
from ..core.request_options import RequestOptions
from ..types.add_triple_response import AddTripleResponse
from ..types.clone_graph_response import CloneGraphResponse
from ..types.custom_instruction import CustomInstruction
from ..types.detect_config import DetectConfig
from ..types.detect_patterns_response import DetectPatternsResponse
from ..types.edge_type import EdgeType
from ..types.entity_type import EntityType
from ..types.entity_type_response import EntityTypeResponse
from ..types.episode import Episode
from ..types.episode_data import EpisodeData
from ..types.graph import Graph
from ..types.graph_data_type import GraphDataType
from ..types.graph_list_response import GraphListResponse
from ..types.graph_search_results import GraphSearchResults
from ..types.graph_search_scope import GraphSearchScope
from ..types.list_custom_instructions_response import ListCustomInstructionsResponse
from ..types.pattern_seeds import PatternSeeds
from ..types.recency_weight import RecencyWeight
from ..types.reranker import Reranker
from ..types.search_filters import SearchFilters
from ..types.success_response import SuccessResponse
from .community.client import AsyncCommunityClient, CommunityClient
from .edge.client import AsyncEdgeClient, EdgeClient
from .episode.client import AsyncEpisodeClient, EpisodeClient
from .node.client import AsyncNodeClient, NodeClient
from .raw_client import AsyncRawGraphClient, RawGraphClient
from .saga.client import AsyncSagaClient, SagaClient
from .theme.client import AsyncThemeClient, ThemeClient

# this is used as the default value for optional parameters
OMIT = typing.cast(typing.Any, ...)


class GraphClient:
    def __init__(self, *, client_wrapper: SyncClientWrapper):
        self._raw_client = RawGraphClient(client_wrapper=client_wrapper)
        self.community = CommunityClient(client_wrapper=client_wrapper)

        self.edge = EdgeClient(client_wrapper=client_wrapper)

        self.episode = EpisodeClient(client_wrapper=client_wrapper)

        self.node = NodeClient(client_wrapper=client_wrapper)

        self.saga = SagaClient(client_wrapper=client_wrapper)

        self.theme = ThemeClient(client_wrapper=client_wrapper)

    @property
    def with_raw_response(self) -> RawGraphClient:
        """
        Retrieves a raw implementation of this client that returns raw responses.

        Returns
        -------
        RawGraphClient
        """
        return self._raw_client

    def list_custom_instructions(
        self,
        *,
        user_id: typing.Optional[str] = None,
        graph_id: typing.Optional[str] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> ListCustomInstructionsResponse:
        """
        Lists all custom instructions for a project, user, or graph.

        Parameters
        ----------
        user_id : typing.Optional[str]
            User ID to get user-specific instructions

        graph_id : typing.Optional[str]
            Graph ID to get graph-specific instructions

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ListCustomInstructionsResponse
            The list of instructions.

        Examples
        --------
        from zep_cloud import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.list_custom_instructions(
            user_id="user_id",
            graph_id="graph_id",
        )
        """
        _response = self._raw_client.list_custom_instructions(
            user_id=user_id, graph_id=graph_id, request_options=request_options
        )
        return _response.data

    def add_custom_instructions(
        self,
        *,
        instructions: typing.Sequence[CustomInstruction],
        graph_ids: typing.Optional[typing.Sequence[str]] = OMIT,
        user_ids: typing.Optional[typing.Sequence[str]] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> SuccessResponse:
        """
        Adds new custom instructions for graphs without removing existing ones. If user_ids or graph_ids is empty, adds to project-wide default instructions.

        Parameters
        ----------
        instructions : typing.Sequence[CustomInstruction]
            Instructions to add to the graph.

        graph_ids : typing.Optional[typing.Sequence[str]]
            Graph IDs to add the instructions to. If empty, the instructions are added to the project-wide default.

        user_ids : typing.Optional[typing.Sequence[str]]
            User IDs to add the instructions to. If empty, the instructions are added to the project-wide default.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        SuccessResponse
            Instructions added successfully

        Examples
        --------
        from zep_cloud import CustomInstruction, Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.add_custom_instructions(
            instructions=[
                CustomInstruction(
                    name="name",
                    text="text",
                )
            ],
        )
        """
        _response = self._raw_client.add_custom_instructions(
            instructions=instructions, graph_ids=graph_ids, user_ids=user_ids, request_options=request_options
        )
        return _response.data

    def delete_custom_instructions(
        self,
        *,
        graph_ids: typing.Optional[typing.Sequence[str]] = OMIT,
        instruction_names: typing.Optional[typing.Sequence[str]] = OMIT,
        user_ids: typing.Optional[typing.Sequence[str]] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> SuccessResponse:
        """
        Deletes custom instructions for graphs or project wide defaults.

        Parameters
        ----------
        graph_ids : typing.Optional[typing.Sequence[str]]
            Determines which group graphs will have their custom instructions deleted. If no graphs are provided, the project-wide custom instructions will be affected.

        instruction_names : typing.Optional[typing.Sequence[str]]
            Unique identifier for the instructions to be deleted. If empty deletes all instructions.

        user_ids : typing.Optional[typing.Sequence[str]]
            Determines which user graphs will have their custom instructions deleted. If no users are provided, the project-wide custom instructions will be affected.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        SuccessResponse
            Instructions deleted successfully

        Examples
        --------
        from zep_cloud import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.delete_custom_instructions()
        """
        _response = self._raw_client.delete_custom_instructions(
            graph_ids=graph_ids, instruction_names=instruction_names, user_ids=user_ids, request_options=request_options
        )
        return _response.data

    def list_entity_types(
        self,
        *,
        user_id: typing.Optional[str] = None,
        graph_id: typing.Optional[str] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> EntityTypeResponse:
        """
        Returns all entity types for a project, user, or graph.

        Parameters
        ----------
        user_id : typing.Optional[str]
            User ID to get user-specific entity types

        graph_id : typing.Optional[str]
            Graph ID to get graph-specific entity types

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        EntityTypeResponse
            The list of entity types.

        Examples
        --------
        from zep_cloud import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.list_entity_types(
            user_id="user_id",
            graph_id="graph_id",
        )
        """
        _response = self._raw_client.list_entity_types(
            user_id=user_id, graph_id=graph_id, request_options=request_options
        )
        return _response.data

    def set_entity_types_internal(
        self,
        *,
        edge_types: typing.Optional[typing.Sequence[EdgeType]] = OMIT,
        entity_types: typing.Optional[typing.Sequence[EntityType]] = OMIT,
        graph_ids: typing.Optional[typing.Sequence[str]] = OMIT,
        user_ids: typing.Optional[typing.Sequence[str]] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> SuccessResponse:
        """
        Sets the entity types for multiple users and graphs, replacing any existing ones.

        Parameters
        ----------
        edge_types : typing.Optional[typing.Sequence[EdgeType]]

        entity_types : typing.Optional[typing.Sequence[EntityType]]

        graph_ids : typing.Optional[typing.Sequence[str]]

        user_ids : typing.Optional[typing.Sequence[str]]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        SuccessResponse
            Entity types set successfully

        Examples
        --------
        from zep_cloud import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.set_entity_types_internal()
        """
        _response = self._raw_client.set_entity_types_internal(
            edge_types=edge_types,
            entity_types=entity_types,
            graph_ids=graph_ids,
            user_ids=user_ids,
            request_options=request_options,
        )
        return _response.data

    def add(
        self,
        *,
        data: str,
        type: GraphDataType,
        created_at: typing.Optional[str] = OMIT,
        graph_id: typing.Optional[str] = OMIT,
        metadata: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT,
        source_description: typing.Optional[str] = OMIT,
        user_id: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> Episode:
        """
        Add data to the graph.

        Parameters
        ----------
        data : str

        type : GraphDataType

        created_at : typing.Optional[str]

        graph_id : typing.Optional[str]
            graph_id is the ID of the graph to which the data will be added. If adding to the user graph, please use user_id field instead.

        metadata : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]]
            Optional metadata key-value pairs. Max 10 keys. Values must be strings, numbers, booleans, or arrays of scalars.

        source_description : typing.Optional[str]

        user_id : typing.Optional[str]
            User ID is the ID of the user to which the data will be added. If not adding to a user graph, please use graph_id field instead.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        Episode
            Added episode

        Examples
        --------
        from zep_cloud import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.add(
            data="data",
            type="text",
        )
        """
        _response = self._raw_client.add(
            data=data,
            type=type,
            created_at=created_at,
            graph_id=graph_id,
            metadata=metadata,
            source_description=source_description,
            user_id=user_id,
            request_options=request_options,
        )
        return _response.data

    def add_batch(
        self,
        *,
        episodes: typing.Sequence[EpisodeData],
        graph_id: typing.Optional[str] = OMIT,
        user_id: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> typing.List[Episode]:
        """
        Add data to the graph in batch mode. Episodes are processed sequentially in the order provided.

        Parameters
        ----------
        episodes : typing.Sequence[EpisodeData]

        graph_id : typing.Optional[str]
            graph_id is the ID of the graph to which the data will be added. If adding to the user graph, please use user_id field instead.

        user_id : typing.Optional[str]
            User ID is the ID of the user to which the data will be added. If not adding to a user graph, please use graph_id field instead.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        typing.List[Episode]
            Added episodes

        Examples
        --------
        from zep_cloud import EpisodeData, Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.add_batch(
            episodes=[
                EpisodeData(
                    data="data",
                    type="text",
                )
            ],
        )
        """
        _response = self._raw_client.add_batch(
            episodes=episodes, graph_id=graph_id, user_id=user_id, request_options=request_options
        )
        return _response.data

    def add_fact_triple(
        self,
        *,
        fact: str,
        fact_name: str,
        created_at: typing.Optional[str] = OMIT,
        edge_attributes: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT,
        expired_at: typing.Optional[str] = OMIT,
        fact_uuid: typing.Optional[str] = OMIT,
        graph_id: typing.Optional[str] = OMIT,
        invalid_at: typing.Optional[str] = OMIT,
        metadata: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT,
        source_node_attributes: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT,
        source_node_labels: typing.Optional[typing.Sequence[str]] = OMIT,
        source_node_name: typing.Optional[str] = OMIT,
        source_node_summary: typing.Optional[str] = OMIT,
        source_node_uuid: typing.Optional[str] = OMIT,
        target_node_attributes: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT,
        target_node_labels: typing.Optional[typing.Sequence[str]] = OMIT,
        target_node_name: typing.Optional[str] = OMIT,
        target_node_summary: typing.Optional[str] = OMIT,
        target_node_uuid: typing.Optional[str] = OMIT,
        user_id: typing.Optional[str] = OMIT,
        valid_at: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> AddTripleResponse:
        """
        Add a fact triple for a user or group

        Parameters
        ----------
        fact : str
            The fact relating the two nodes that this edge represents

        fact_name : str
            The name of the edge to add. Should be all caps using snake case (eg RELATES_TO)

        created_at : typing.Optional[str]
            The timestamp of the message

        edge_attributes : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]]
            Additional attributes of the edge. Values must be scalar types (string, number, boolean, or null).
            Nested objects and arrays are not allowed.

        expired_at : typing.Optional[str]
            The time (if any) at which the edge expires

        fact_uuid : typing.Optional[str]
            The uuid of the edge to add

        graph_id : typing.Optional[str]

        invalid_at : typing.Optional[str]
            The time (if any) at which the fact stops being true

        metadata : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]]
            Optional metadata key-value pairs for the shadow episode created for this fact triple.
            Max 10 keys. Values must be strings, numbers, or booleans.

        source_node_attributes : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]]
            Additional attributes of the source node. Values must be scalar types (string, number, boolean, or null).
            Nested objects and arrays are not allowed.

        source_node_labels : typing.Optional[typing.Sequence[str]]
            The labels for the source node

        source_node_name : typing.Optional[str]
            The name of the source node to add

        source_node_summary : typing.Optional[str]
            The summary of the source node to add

        source_node_uuid : typing.Optional[str]
            The source node uuid

        target_node_attributes : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]]
            Additional attributes of the target node. Values must be scalar types (string, number, boolean, or null).
            Nested objects and arrays are not allowed.

        target_node_labels : typing.Optional[typing.Sequence[str]]
            The labels for the target node

        target_node_name : typing.Optional[str]
            The name of the target node to add

        target_node_summary : typing.Optional[str]
            The summary of the target node to add

        target_node_uuid : typing.Optional[str]
            The target node uuid

        user_id : typing.Optional[str]

        valid_at : typing.Optional[str]
            The time at which the fact becomes true

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        AddTripleResponse
            Resulting triple

        Examples
        --------
        from zep_cloud import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.add_fact_triple(
            fact="fact",
            fact_name="fact_name",
        )
        """
        _response = self._raw_client.add_fact_triple(
            fact=fact,
            fact_name=fact_name,
            created_at=created_at,
            edge_attributes=edge_attributes,
            expired_at=expired_at,
            fact_uuid=fact_uuid,
            graph_id=graph_id,
            invalid_at=invalid_at,
            metadata=metadata,
            source_node_attributes=source_node_attributes,
            source_node_labels=source_node_labels,
            source_node_name=source_node_name,
            source_node_summary=source_node_summary,
            source_node_uuid=source_node_uuid,
            target_node_attributes=target_node_attributes,
            target_node_labels=target_node_labels,
            target_node_name=target_node_name,
            target_node_summary=target_node_summary,
            target_node_uuid=target_node_uuid,
            user_id=user_id,
            valid_at=valid_at,
            request_options=request_options,
        )
        return _response.data

    def clone(
        self,
        *,
        source_graph_id: typing.Optional[str] = OMIT,
        source_user_id: typing.Optional[str] = OMIT,
        target_graph_id: typing.Optional[str] = OMIT,
        target_user_id: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> CloneGraphResponse:
        """
        Clone a user or group graph.

        Parameters
        ----------
        source_graph_id : typing.Optional[str]
            source_graph_id is the ID of the graph to be cloned. Required if source_user_id is not provided

        source_user_id : typing.Optional[str]
            user_id of the user whose graph is being cloned. Required if source_graph_id is not provided

        target_graph_id : typing.Optional[str]
            target_graph_id is the ID to be set on the cloned graph. Must not point to an existing graph. Required if target_user_id is not provided.

        target_user_id : typing.Optional[str]
            user_id to be set on the cloned user. Must not point to an existing user. Required if target_graph_id is not provided.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        CloneGraphResponse
            Response object containing graph_id or user_id pointing to the new graph

        Examples
        --------
        from zep_cloud import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.clone()
        """
        _response = self._raw_client.clone(
            source_graph_id=source_graph_id,
            source_user_id=source_user_id,
            target_graph_id=target_graph_id,
            target_user_id=target_user_id,
            request_options=request_options,
        )
        return _response.data

    def create(
        self,
        *,
        graph_id: str,
        description: typing.Optional[str] = OMIT,
        name: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> Graph:
        """
        Creates a new graph.

        Parameters
        ----------
        graph_id : str

        description : typing.Optional[str]

        name : typing.Optional[str]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        Graph
            The added graph

        Examples
        --------
        from zep_cloud import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.create(
            graph_id="graph_id",
        )
        """
        _response = self._raw_client.create(
            graph_id=graph_id, description=description, name=name, request_options=request_options
        )
        return _response.data

    def list_all(
        self,
        *,
        page_number: typing.Optional[int] = None,
        page_size: typing.Optional[int] = None,
        search: typing.Optional[str] = None,
        order_by: typing.Optional[str] = None,
        asc: typing.Optional[bool] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> GraphListResponse:
        """
        Returns all graphs. In order to list users, use user.list_ordered instead

        Parameters
        ----------
        page_number : typing.Optional[int]
            Page number for pagination, starting from 1.

        page_size : typing.Optional[int]
            Number of graphs to retrieve per page.

        search : typing.Optional[str]
            Search term for filtering graphs by graph_id.

        order_by : typing.Optional[str]
            Column to sort by (created_at, group_id, name).

        asc : typing.Optional[bool]
            Sort in ascending order.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GraphListResponse
            Successfully retrieved list of graphs.

        Examples
        --------
        from zep_cloud import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.list_all(
            page_number=1,
            page_size=1,
            search="search",
            order_by="order_by",
            asc=True,
        )
        """
        _response = self._raw_client.list_all(
            page_number=page_number,
            page_size=page_size,
            search=search,
            order_by=order_by,
            asc=asc,
            request_options=request_options,
        )
        return _response.data

    def detect_patterns(
        self,
        *,
        detect: typing.Optional[DetectConfig] = OMIT,
        edge_limit: typing.Optional[int] = OMIT,
        graph_id: typing.Optional[str] = OMIT,
        limit: typing.Optional[int] = OMIT,
        min_occurrences: typing.Optional[int] = OMIT,
        query: typing.Optional[str] = OMIT,
        query_limit: typing.Optional[int] = OMIT,
        recency_weight: typing.Optional[RecencyWeight] = OMIT,
        search_filters: typing.Optional[SearchFilters] = OMIT,
        seeds: typing.Optional[PatternSeeds] = OMIT,
        user_id: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> DetectPatternsResponse:
        """
        Detects structural patterns in a knowledge graph including relationship frequencies,
        multi-hop paths, co-occurrences, hubs, and clusters.
        When a query is provided, uses hybrid search to discover seed nodes,
        detects triple-frequency patterns, and returns resolved edges ranked by relevance.

        Parameters
        ----------
        detect : typing.Optional[DetectConfig]
            Which pattern types to detect with type-specific configuration.
            Omit to detect all types with defaults. Ignored when query is set.

        edge_limit : typing.Optional[int]
            Max resolved edges per pattern. Default: 10, Max: 100. Only used with query.

        graph_id : typing.Optional[str]
            Graph ID when detecting patterns on a named graph

        limit : typing.Optional[int]
            Max patterns to return. Default: 50, Max: 200

        min_occurrences : typing.Optional[int]
            Minimum occurrence count to report a pattern. Default: 2

        query : typing.Optional[str]
            Search query for discovering seed nodes via hybrid search.
            When set, forces triple-frequency detection only and enables edge resolution
            with cross-encoder reranking. Mutually exclusive with seeds.

        query_limit : typing.Optional[int]
            Max seed nodes from search. Default: 10, Max: 50. Only used with query.

        recency_weight : typing.Optional[RecencyWeight]
            Exponential half-life decay applied to edge created_at timestamps.
            Valid values: none, 7_days, 30_days, 90_days. Default: none

        search_filters : typing.Optional[SearchFilters]
            Filters which edges/nodes participate in pattern detection.
            Reuses the same filter format as /graph/search.

        seeds : typing.Optional[PatternSeeds]
            Seed selection. If omitted, analyzes the entire graph. Mutually exclusive with query.

        user_id : typing.Optional[str]
            User ID when detecting patterns on a user graph

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        DetectPatternsResponse
            Detected patterns

        Examples
        --------
        from zep_cloud import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.detect_patterns()
        """
        _response = self._raw_client.detect_patterns(
            detect=detect,
            edge_limit=edge_limit,
            graph_id=graph_id,
            limit=limit,
            min_occurrences=min_occurrences,
            query=query,
            query_limit=query_limit,
            recency_weight=recency_weight,
            search_filters=search_filters,
            seeds=seeds,
            user_id=user_id,
            request_options=request_options,
        )
        return _response.data

    def search(
        self,
        *,
        query: str,
        bfs_origin_node_uuids: typing.Optional[typing.Sequence[str]] = OMIT,
        center_node_uuid: typing.Optional[str] = OMIT,
        graph_id: typing.Optional[str] = OMIT,
        limit: typing.Optional[int] = OMIT,
        max_characters: typing.Optional[int] = OMIT,
        mmr_lambda: typing.Optional[float] = OMIT,
        reranker: typing.Optional[Reranker] = OMIT,
        return_raw_results: typing.Optional[bool] = OMIT,
        scope: typing.Optional[GraphSearchScope] = OMIT,
        search_filters: typing.Optional[SearchFilters] = OMIT,
        user_id: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> GraphSearchResults:
        """
        Perform a graph search query.

        Parameters
        ----------
        query : str
            The string to search for (required)

        bfs_origin_node_uuids : typing.Optional[typing.Sequence[str]]
            Nodes that are the origins of the BFS searches

        center_node_uuid : typing.Optional[str]
            Node to rerank around for node distance reranking

        graph_id : typing.Optional[str]
            The graph_id to search in. When searching user graph, please use user_id instead.

        limit : typing.Optional[int]
            The maximum number of facts to retrieve. Defaults to 10. Limited to 50.

        max_characters : typing.Optional[int]
            Maximum total characters across all selected results when scope=auto. Defaults to 2000. Limited to 50000.

        mmr_lambda : typing.Optional[float]
            weighting for maximal marginal relevance

        reranker : typing.Optional[Reranker]
            Defaults to RRF

        return_raw_results : typing.Optional[bool]
            When scope=auto, include the selected raw graph results alongside the materialized context block.

        scope : typing.Optional[GraphSearchScope]
            Defaults to Edges.

        search_filters : typing.Optional[SearchFilters]
            Search filters to apply to the search

        user_id : typing.Optional[str]
            The user_id when searching user graph. If not searching user graph, please use graph_id instead.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GraphSearchResults
            Graph search results or auto-context block

        Examples
        --------
        from zep_cloud import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.search(
            query="query",
        )
        """
        _response = self._raw_client.search(
            query=query,
            bfs_origin_node_uuids=bfs_origin_node_uuids,
            center_node_uuid=center_node_uuid,
            graph_id=graph_id,
            limit=limit,
            max_characters=max_characters,
            mmr_lambda=mmr_lambda,
            reranker=reranker,
            return_raw_results=return_raw_results,
            scope=scope,
            search_filters=search_filters,
            user_id=user_id,
            request_options=request_options,
        )
        return _response.data

    def get(self, graph_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> Graph:
        """
        Returns a graph.

        Parameters
        ----------
        graph_id : str
            The graph_id of the graph to get.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        Graph
            The graph that was retrieved.

        Examples
        --------
        from zep_cloud import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.get(
            graph_id="graphId",
        )
        """
        _response = self._raw_client.get(graph_id, request_options=request_options)
        return _response.data

    def delete(self, graph_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> SuccessResponse:
        """
        Deletes a graph. If you would like to delete a user graph, make sure to use user.delete instead.

        Parameters
        ----------
        graph_id : str
            Graph ID

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        SuccessResponse
            Deleted

        Examples
        --------
        from zep_cloud import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.delete(
            graph_id="graphId",
        )
        """
        _response = self._raw_client.delete(graph_id, request_options=request_options)
        return _response.data

    def update(
        self,
        graph_id: str,
        *,
        description: typing.Optional[str] = OMIT,
        name: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> Graph:
        """
        Updates information about a graph.

        Parameters
        ----------
        graph_id : str
            Graph ID

        description : typing.Optional[str]

        name : typing.Optional[str]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        Graph
            The updated graph object

        Examples
        --------
        from zep_cloud import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.update(
            graph_id="graphId",
        )
        """
        _response = self._raw_client.update(
            graph_id, description=description, name=name, request_options=request_options
        )
        return _response.data


class AsyncGraphClient:
    def __init__(self, *, client_wrapper: AsyncClientWrapper):
        self._raw_client = AsyncRawGraphClient(client_wrapper=client_wrapper)
        self.community = AsyncCommunityClient(client_wrapper=client_wrapper)

        self.edge = AsyncEdgeClient(client_wrapper=client_wrapper)

        self.episode = AsyncEpisodeClient(client_wrapper=client_wrapper)

        self.node = AsyncNodeClient(client_wrapper=client_wrapper)

        self.saga = AsyncSagaClient(client_wrapper=client_wrapper)

        self.theme = AsyncThemeClient(client_wrapper=client_wrapper)

    @property
    def with_raw_response(self) -> AsyncRawGraphClient:
        """
        Retrieves a raw implementation of this client that returns raw responses.

        Returns
        -------
        AsyncRawGraphClient
        """
        return self._raw_client

    async def list_custom_instructions(
        self,
        *,
        user_id: typing.Optional[str] = None,
        graph_id: typing.Optional[str] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> ListCustomInstructionsResponse:
        """
        Lists all custom instructions for a project, user, or graph.

        Parameters
        ----------
        user_id : typing.Optional[str]
            User ID to get user-specific instructions

        graph_id : typing.Optional[str]
            Graph ID to get graph-specific instructions

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ListCustomInstructionsResponse
            The list of instructions.

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.list_custom_instructions(
                user_id="user_id",
                graph_id="graph_id",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.list_custom_instructions(
            user_id=user_id, graph_id=graph_id, request_options=request_options
        )
        return _response.data

    async def add_custom_instructions(
        self,
        *,
        instructions: typing.Sequence[CustomInstruction],
        graph_ids: typing.Optional[typing.Sequence[str]] = OMIT,
        user_ids: typing.Optional[typing.Sequence[str]] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> SuccessResponse:
        """
        Adds new custom instructions for graphs without removing existing ones. If user_ids or graph_ids is empty, adds to project-wide default instructions.

        Parameters
        ----------
        instructions : typing.Sequence[CustomInstruction]
            Instructions to add to the graph.

        graph_ids : typing.Optional[typing.Sequence[str]]
            Graph IDs to add the instructions to. If empty, the instructions are added to the project-wide default.

        user_ids : typing.Optional[typing.Sequence[str]]
            User IDs to add the instructions to. If empty, the instructions are added to the project-wide default.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        SuccessResponse
            Instructions added successfully

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep, CustomInstruction

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.add_custom_instructions(
                instructions=[
                    CustomInstruction(
                        name="name",
                        text="text",
                    )
                ],
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.add_custom_instructions(
            instructions=instructions, graph_ids=graph_ids, user_ids=user_ids, request_options=request_options
        )
        return _response.data

    async def delete_custom_instructions(
        self,
        *,
        graph_ids: typing.Optional[typing.Sequence[str]] = OMIT,
        instruction_names: typing.Optional[typing.Sequence[str]] = OMIT,
        user_ids: typing.Optional[typing.Sequence[str]] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> SuccessResponse:
        """
        Deletes custom instructions for graphs or project wide defaults.

        Parameters
        ----------
        graph_ids : typing.Optional[typing.Sequence[str]]
            Determines which group graphs will have their custom instructions deleted. If no graphs are provided, the project-wide custom instructions will be affected.

        instruction_names : typing.Optional[typing.Sequence[str]]
            Unique identifier for the instructions to be deleted. If empty deletes all instructions.

        user_ids : typing.Optional[typing.Sequence[str]]
            Determines which user graphs will have their custom instructions deleted. If no users are provided, the project-wide custom instructions will be affected.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        SuccessResponse
            Instructions deleted successfully

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.delete_custom_instructions()


        asyncio.run(main())
        """
        _response = await self._raw_client.delete_custom_instructions(
            graph_ids=graph_ids, instruction_names=instruction_names, user_ids=user_ids, request_options=request_options
        )
        return _response.data

    async def list_entity_types(
        self,
        *,
        user_id: typing.Optional[str] = None,
        graph_id: typing.Optional[str] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> EntityTypeResponse:
        """
        Returns all entity types for a project, user, or graph.

        Parameters
        ----------
        user_id : typing.Optional[str]
            User ID to get user-specific entity types

        graph_id : typing.Optional[str]
            Graph ID to get graph-specific entity types

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        EntityTypeResponse
            The list of entity types.

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.list_entity_types(
                user_id="user_id",
                graph_id="graph_id",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.list_entity_types(
            user_id=user_id, graph_id=graph_id, request_options=request_options
        )
        return _response.data

    async def set_entity_types_internal(
        self,
        *,
        edge_types: typing.Optional[typing.Sequence[EdgeType]] = OMIT,
        entity_types: typing.Optional[typing.Sequence[EntityType]] = OMIT,
        graph_ids: typing.Optional[typing.Sequence[str]] = OMIT,
        user_ids: typing.Optional[typing.Sequence[str]] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> SuccessResponse:
        """
        Sets the entity types for multiple users and graphs, replacing any existing ones.

        Parameters
        ----------
        edge_types : typing.Optional[typing.Sequence[EdgeType]]

        entity_types : typing.Optional[typing.Sequence[EntityType]]

        graph_ids : typing.Optional[typing.Sequence[str]]

        user_ids : typing.Optional[typing.Sequence[str]]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        SuccessResponse
            Entity types set successfully

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.set_entity_types_internal()


        asyncio.run(main())
        """
        _response = await self._raw_client.set_entity_types_internal(
            edge_types=edge_types,
            entity_types=entity_types,
            graph_ids=graph_ids,
            user_ids=user_ids,
            request_options=request_options,
        )
        return _response.data

    async def add(
        self,
        *,
        data: str,
        type: GraphDataType,
        created_at: typing.Optional[str] = OMIT,
        graph_id: typing.Optional[str] = OMIT,
        metadata: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT,
        source_description: typing.Optional[str] = OMIT,
        user_id: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> Episode:
        """
        Add data to the graph.

        Parameters
        ----------
        data : str

        type : GraphDataType

        created_at : typing.Optional[str]

        graph_id : typing.Optional[str]
            graph_id is the ID of the graph to which the data will be added. If adding to the user graph, please use user_id field instead.

        metadata : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]]
            Optional metadata key-value pairs. Max 10 keys. Values must be strings, numbers, booleans, or arrays of scalars.

        source_description : typing.Optional[str]

        user_id : typing.Optional[str]
            User ID is the ID of the user to which the data will be added. If not adding to a user graph, please use graph_id field instead.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        Episode
            Added episode

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.add(
                data="data",
                type="text",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.add(
            data=data,
            type=type,
            created_at=created_at,
            graph_id=graph_id,
            metadata=metadata,
            source_description=source_description,
            user_id=user_id,
            request_options=request_options,
        )
        return _response.data

    async def add_batch(
        self,
        *,
        episodes: typing.Sequence[EpisodeData],
        graph_id: typing.Optional[str] = OMIT,
        user_id: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> typing.List[Episode]:
        """
        Add data to the graph in batch mode. Episodes are processed sequentially in the order provided.

        Parameters
        ----------
        episodes : typing.Sequence[EpisodeData]

        graph_id : typing.Optional[str]
            graph_id is the ID of the graph to which the data will be added. If adding to the user graph, please use user_id field instead.

        user_id : typing.Optional[str]
            User ID is the ID of the user to which the data will be added. If not adding to a user graph, please use graph_id field instead.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        typing.List[Episode]
            Added episodes

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep, EpisodeData

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.add_batch(
                episodes=[
                    EpisodeData(
                        data="data",
                        type="text",
                    )
                ],
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.add_batch(
            episodes=episodes, graph_id=graph_id, user_id=user_id, request_options=request_options
        )
        return _response.data

    async def add_fact_triple(
        self,
        *,
        fact: str,
        fact_name: str,
        created_at: typing.Optional[str] = OMIT,
        edge_attributes: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT,
        expired_at: typing.Optional[str] = OMIT,
        fact_uuid: typing.Optional[str] = OMIT,
        graph_id: typing.Optional[str] = OMIT,
        invalid_at: typing.Optional[str] = OMIT,
        metadata: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT,
        source_node_attributes: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT,
        source_node_labels: typing.Optional[typing.Sequence[str]] = OMIT,
        source_node_name: typing.Optional[str] = OMIT,
        source_node_summary: typing.Optional[str] = OMIT,
        source_node_uuid: typing.Optional[str] = OMIT,
        target_node_attributes: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT,
        target_node_labels: typing.Optional[typing.Sequence[str]] = OMIT,
        target_node_name: typing.Optional[str] = OMIT,
        target_node_summary: typing.Optional[str] = OMIT,
        target_node_uuid: typing.Optional[str] = OMIT,
        user_id: typing.Optional[str] = OMIT,
        valid_at: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> AddTripleResponse:
        """
        Add a fact triple for a user or group

        Parameters
        ----------
        fact : str
            The fact relating the two nodes that this edge represents

        fact_name : str
            The name of the edge to add. Should be all caps using snake case (eg RELATES_TO)

        created_at : typing.Optional[str]
            The timestamp of the message

        edge_attributes : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]]
            Additional attributes of the edge. Values must be scalar types (string, number, boolean, or null).
            Nested objects and arrays are not allowed.

        expired_at : typing.Optional[str]
            The time (if any) at which the edge expires

        fact_uuid : typing.Optional[str]
            The uuid of the edge to add

        graph_id : typing.Optional[str]

        invalid_at : typing.Optional[str]
            The time (if any) at which the fact stops being true

        metadata : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]]
            Optional metadata key-value pairs for the shadow episode created for this fact triple.
            Max 10 keys. Values must be strings, numbers, or booleans.

        source_node_attributes : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]]
            Additional attributes of the source node. Values must be scalar types (string, number, boolean, or null).
            Nested objects and arrays are not allowed.

        source_node_labels : typing.Optional[typing.Sequence[str]]
            The labels for the source node

        source_node_name : typing.Optional[str]
            The name of the source node to add

        source_node_summary : typing.Optional[str]
            The summary of the source node to add

        source_node_uuid : typing.Optional[str]
            The source node uuid

        target_node_attributes : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]]
            Additional attributes of the target node. Values must be scalar types (string, number, boolean, or null).
            Nested objects and arrays are not allowed.

        target_node_labels : typing.Optional[typing.Sequence[str]]
            The labels for the target node

        target_node_name : typing.Optional[str]
            The name of the target node to add

        target_node_summary : typing.Optional[str]
            The summary of the target node to add

        target_node_uuid : typing.Optional[str]
            The target node uuid

        user_id : typing.Optional[str]

        valid_at : typing.Optional[str]
            The time at which the fact becomes true

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        AddTripleResponse
            Resulting triple

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.add_fact_triple(
                fact="fact",
                fact_name="fact_name",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.add_fact_triple(
            fact=fact,
            fact_name=fact_name,
            created_at=created_at,
            edge_attributes=edge_attributes,
            expired_at=expired_at,
            fact_uuid=fact_uuid,
            graph_id=graph_id,
            invalid_at=invalid_at,
            metadata=metadata,
            source_node_attributes=source_node_attributes,
            source_node_labels=source_node_labels,
            source_node_name=source_node_name,
            source_node_summary=source_node_summary,
            source_node_uuid=source_node_uuid,
            target_node_attributes=target_node_attributes,
            target_node_labels=target_node_labels,
            target_node_name=target_node_name,
            target_node_summary=target_node_summary,
            target_node_uuid=target_node_uuid,
            user_id=user_id,
            valid_at=valid_at,
            request_options=request_options,
        )
        return _response.data

    async def clone(
        self,
        *,
        source_graph_id: typing.Optional[str] = OMIT,
        source_user_id: typing.Optional[str] = OMIT,
        target_graph_id: typing.Optional[str] = OMIT,
        target_user_id: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> CloneGraphResponse:
        """
        Clone a user or group graph.

        Parameters
        ----------
        source_graph_id : typing.Optional[str]
            source_graph_id is the ID of the graph to be cloned. Required if source_user_id is not provided

        source_user_id : typing.Optional[str]
            user_id of the user whose graph is being cloned. Required if source_graph_id is not provided

        target_graph_id : typing.Optional[str]
            target_graph_id is the ID to be set on the cloned graph. Must not point to an existing graph. Required if target_user_id is not provided.

        target_user_id : typing.Optional[str]
            user_id to be set on the cloned user. Must not point to an existing user. Required if target_graph_id is not provided.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        CloneGraphResponse
            Response object containing graph_id or user_id pointing to the new graph

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.clone()


        asyncio.run(main())
        """
        _response = await self._raw_client.clone(
            source_graph_id=source_graph_id,
            source_user_id=source_user_id,
            target_graph_id=target_graph_id,
            target_user_id=target_user_id,
            request_options=request_options,
        )
        return _response.data

    async def create(
        self,
        *,
        graph_id: str,
        description: typing.Optional[str] = OMIT,
        name: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> Graph:
        """
        Creates a new graph.

        Parameters
        ----------
        graph_id : str

        description : typing.Optional[str]

        name : typing.Optional[str]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        Graph
            The added graph

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.create(
                graph_id="graph_id",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.create(
            graph_id=graph_id, description=description, name=name, request_options=request_options
        )
        return _response.data

    async def list_all(
        self,
        *,
        page_number: typing.Optional[int] = None,
        page_size: typing.Optional[int] = None,
        search: typing.Optional[str] = None,
        order_by: typing.Optional[str] = None,
        asc: typing.Optional[bool] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> GraphListResponse:
        """
        Returns all graphs. In order to list users, use user.list_ordered instead

        Parameters
        ----------
        page_number : typing.Optional[int]
            Page number for pagination, starting from 1.

        page_size : typing.Optional[int]
            Number of graphs to retrieve per page.

        search : typing.Optional[str]
            Search term for filtering graphs by graph_id.

        order_by : typing.Optional[str]
            Column to sort by (created_at, group_id, name).

        asc : typing.Optional[bool]
            Sort in ascending order.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GraphListResponse
            Successfully retrieved list of graphs.

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.list_all(
                page_number=1,
                page_size=1,
                search="search",
                order_by="order_by",
                asc=True,
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.list_all(
            page_number=page_number,
            page_size=page_size,
            search=search,
            order_by=order_by,
            asc=asc,
            request_options=request_options,
        )
        return _response.data

    async def detect_patterns(
        self,
        *,
        detect: typing.Optional[DetectConfig] = OMIT,
        edge_limit: typing.Optional[int] = OMIT,
        graph_id: typing.Optional[str] = OMIT,
        limit: typing.Optional[int] = OMIT,
        min_occurrences: typing.Optional[int] = OMIT,
        query: typing.Optional[str] = OMIT,
        query_limit: typing.Optional[int] = OMIT,
        recency_weight: typing.Optional[RecencyWeight] = OMIT,
        search_filters: typing.Optional[SearchFilters] = OMIT,
        seeds: typing.Optional[PatternSeeds] = OMIT,
        user_id: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> DetectPatternsResponse:
        """
        Detects structural patterns in a knowledge graph including relationship frequencies,
        multi-hop paths, co-occurrences, hubs, and clusters.
        When a query is provided, uses hybrid search to discover seed nodes,
        detects triple-frequency patterns, and returns resolved edges ranked by relevance.

        Parameters
        ----------
        detect : typing.Optional[DetectConfig]
            Which pattern types to detect with type-specific configuration.
            Omit to detect all types with defaults. Ignored when query is set.

        edge_limit : typing.Optional[int]
            Max resolved edges per pattern. Default: 10, Max: 100. Only used with query.

        graph_id : typing.Optional[str]
            Graph ID when detecting patterns on a named graph

        limit : typing.Optional[int]
            Max patterns to return. Default: 50, Max: 200

        min_occurrences : typing.Optional[int]
            Minimum occurrence count to report a pattern. Default: 2

        query : typing.Optional[str]
            Search query for discovering seed nodes via hybrid search.
            When set, forces triple-frequency detection only and enables edge resolution
            with cross-encoder reranking. Mutually exclusive with seeds.

        query_limit : typing.Optional[int]
            Max seed nodes from search. Default: 10, Max: 50. Only used with query.

        recency_weight : typing.Optional[RecencyWeight]
            Exponential half-life decay applied to edge created_at timestamps.
            Valid values: none, 7_days, 30_days, 90_days. Default: none

        search_filters : typing.Optional[SearchFilters]
            Filters which edges/nodes participate in pattern detection.
            Reuses the same filter format as /graph/search.

        seeds : typing.Optional[PatternSeeds]
            Seed selection. If omitted, analyzes the entire graph. Mutually exclusive with query.

        user_id : typing.Optional[str]
            User ID when detecting patterns on a user graph

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        DetectPatternsResponse
            Detected patterns

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.detect_patterns()


        asyncio.run(main())
        """
        _response = await self._raw_client.detect_patterns(
            detect=detect,
            edge_limit=edge_limit,
            graph_id=graph_id,
            limit=limit,
            min_occurrences=min_occurrences,
            query=query,
            query_limit=query_limit,
            recency_weight=recency_weight,
            search_filters=search_filters,
            seeds=seeds,
            user_id=user_id,
            request_options=request_options,
        )
        return _response.data

    async def search(
        self,
        *,
        query: str,
        bfs_origin_node_uuids: typing.Optional[typing.Sequence[str]] = OMIT,
        center_node_uuid: typing.Optional[str] = OMIT,
        graph_id: typing.Optional[str] = OMIT,
        limit: typing.Optional[int] = OMIT,
        max_characters: typing.Optional[int] = OMIT,
        mmr_lambda: typing.Optional[float] = OMIT,
        reranker: typing.Optional[Reranker] = OMIT,
        return_raw_results: typing.Optional[bool] = OMIT,
        scope: typing.Optional[GraphSearchScope] = OMIT,
        search_filters: typing.Optional[SearchFilters] = OMIT,
        user_id: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> GraphSearchResults:
        """
        Perform a graph search query.

        Parameters
        ----------
        query : str
            The string to search for (required)

        bfs_origin_node_uuids : typing.Optional[typing.Sequence[str]]
            Nodes that are the origins of the BFS searches

        center_node_uuid : typing.Optional[str]
            Node to rerank around for node distance reranking

        graph_id : typing.Optional[str]
            The graph_id to search in. When searching user graph, please use user_id instead.

        limit : typing.Optional[int]
            The maximum number of facts to retrieve. Defaults to 10. Limited to 50.

        max_characters : typing.Optional[int]
            Maximum total characters across all selected results when scope=auto. Defaults to 2000. Limited to 50000.

        mmr_lambda : typing.Optional[float]
            weighting for maximal marginal relevance

        reranker : typing.Optional[Reranker]
            Defaults to RRF

        return_raw_results : typing.Optional[bool]
            When scope=auto, include the selected raw graph results alongside the materialized context block.

        scope : typing.Optional[GraphSearchScope]
            Defaults to Edges.

        search_filters : typing.Optional[SearchFilters]
            Search filters to apply to the search

        user_id : typing.Optional[str]
            The user_id when searching user graph. If not searching user graph, please use graph_id instead.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GraphSearchResults
            Graph search results or auto-context block

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.search(
                query="query",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.search(
            query=query,
            bfs_origin_node_uuids=bfs_origin_node_uuids,
            center_node_uuid=center_node_uuid,
            graph_id=graph_id,
            limit=limit,
            max_characters=max_characters,
            mmr_lambda=mmr_lambda,
            reranker=reranker,
            return_raw_results=return_raw_results,
            scope=scope,
            search_filters=search_filters,
            user_id=user_id,
            request_options=request_options,
        )
        return _response.data

    async def get(self, graph_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> Graph:
        """
        Returns a graph.

        Parameters
        ----------
        graph_id : str
            The graph_id of the graph to get.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        Graph
            The graph that was retrieved.

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.get(
                graph_id="graphId",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.get(graph_id, request_options=request_options)
        return _response.data

    async def delete(
        self, graph_id: str, *, request_options: typing.Optional[RequestOptions] = None
    ) -> SuccessResponse:
        """
        Deletes a graph. If you would like to delete a user graph, make sure to use user.delete instead.

        Parameters
        ----------
        graph_id : str
            Graph ID

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        SuccessResponse
            Deleted

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.delete(
                graph_id="graphId",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.delete(graph_id, request_options=request_options)
        return _response.data

    async def update(
        self,
        graph_id: str,
        *,
        description: typing.Optional[str] = OMIT,
        name: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> Graph:
        """
        Updates information about a graph.

        Parameters
        ----------
        graph_id : str
            Graph ID

        description : typing.Optional[str]

        name : typing.Optional[str]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        Graph
            The updated graph object

        Examples
        --------
        import asyncio

        from zep_cloud import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.graph.update(
                graph_id="graphId",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.update(
            graph_id, description=description, name=name, request_options=request_options
        )
        return _response.data