client.py

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

import typing
from json.decoder import JSONDecodeError

from ..core.api_error import ApiError as core_api_error_ApiError
from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper
from ..core.pydantic_utilities import pydantic_v1
from ..core.request_options import RequestOptions
from ..errors.bad_request_error import BadRequestError
from ..errors.internal_server_error import InternalServerError
from ..errors.not_found_error import NotFoundError
from ..types.add_triple_response import AddTripleResponse
from ..types.api_error import ApiError as types_api_error_ApiError
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_data_type import GraphDataType
from ..types.graph_search_results import GraphSearchResults
from ..types.graph_search_scope import GraphSearchScope
from ..types.reranker import Reranker
from ..types.search_filters import SearchFilters
from ..types.success_response import SuccessResponse
from .edge.client import AsyncEdgeClient, EdgeClient
from .episode.client import AsyncEpisodeClient, EpisodeClient
from .node.client import AsyncNodeClient, NodeClient

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


class GraphClient:
    def __init__(self, *, client_wrapper: SyncClientWrapper):
        self._client_wrapper = client_wrapper
        self.edge = EdgeClient(client_wrapper=self._client_wrapper)
        self.episode = EpisodeClient(client_wrapper=self._client_wrapper)
        self.node = NodeClient(client_wrapper=self._client_wrapper)

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

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

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

        Examples
        --------
        from zep_cloud.client import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.list_entity_types()
        """
        _response = self._client_wrapper.httpx_client.request(
            "entity-types", method="GET", request_options=request_options
        )
        try:
            if 200 <= _response.status_code < 300:
                return pydantic_v1.parse_obj_as(EntityTypeResponse, _response.json())  # type: ignore
            if _response.status_code == 404:
                raise NotFoundError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            if _response.status_code == 500:
                raise InternalServerError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            _response_json = _response.json()
        except JSONDecodeError:
            raise core_api_error_ApiError(status_code=_response.status_code, body=_response.text)
        raise core_api_error_ApiError(status_code=_response.status_code, body=_response_json)

    def set_entity_types_internal(
        self,
        *,
        entity_types: typing.Optional[typing.Sequence[EntityType]] = OMIT,
        request_options: typing.Optional[RequestOptions] = None
    ) -> SuccessResponse:
        """
        Sets the entity types for a project, replacing any existing ones.

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

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

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

        Examples
        --------
        from zep_cloud.client import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.set_entity_types_internal()
        """
        _response = self._client_wrapper.httpx_client.request(
            "entity-types",
            method="PUT",
            json={"entity_types": entity_types},
            request_options=request_options,
            omit=OMIT,
        )
        try:
            if 200 <= _response.status_code < 300:
                return pydantic_v1.parse_obj_as(SuccessResponse, _response.json())  # type: ignore
            if _response.status_code == 400:
                raise BadRequestError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            if _response.status_code == 500:
                raise InternalServerError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            _response_json = _response.json()
        except JSONDecodeError:
            raise core_api_error_ApiError(status_code=_response.status_code, body=_response.text)
        raise core_api_error_ApiError(status_code=_response.status_code, body=_response_json)

    def add(
        self,
        *,
        data: str,
        type: GraphDataType,
        group_id: typing.Optional[str] = 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. Note: each subscription tier has different limits on the amount of data that can be added to the graph please refer to the pricing page for more information.

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

        type : GraphDataType

        group_id : typing.Optional[str]

        source_description : typing.Optional[str]

        user_id : typing.Optional[str]

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

        Returns
        -------
        Episode
            Added episode

        Examples
        --------
        from zep_cloud.client import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.add(
            data="data",
            type="text",
        )
        """
        _response = self._client_wrapper.httpx_client.request(
            "graph",
            method="POST",
            json={
                "data": data,
                "group_id": group_id,
                "source_description": source_description,
                "type": type,
                "user_id": user_id,
            },
            request_options=request_options,
            omit=OMIT,
        )
        try:
            if 200 <= _response.status_code < 300:
                return pydantic_v1.parse_obj_as(Episode, _response.json())  # type: ignore
            if _response.status_code == 400:
                raise BadRequestError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            if _response.status_code == 500:
                raise InternalServerError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            _response_json = _response.json()
        except JSONDecodeError:
            raise core_api_error_ApiError(status_code=_response.status_code, body=_response.text)
        raise core_api_error_ApiError(status_code=_response.status_code, body=_response_json)

    def add_batch(
        self,
        *,
        episodes: typing.Sequence[EpisodeData],
        group_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 (each episode processed concurrently). Note: each subscription tier has different limits on the amount of data that can be added to the graph please refer to the pricing page for more information.

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

        group_id : typing.Optional[str]

        user_id : typing.Optional[str]

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

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

        Examples
        --------
        from zep_cloud import EpisodeData
        from zep_cloud.client import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.add_batch(
            episodes=[
                EpisodeData(
                    data="data",
                    type="text",
                )
            ],
        )
        """
        _response = self._client_wrapper.httpx_client.request(
            "graph-batch",
            method="POST",
            json={"episodes": episodes, "group_id": group_id, "user_id": user_id},
            request_options=request_options,
            omit=OMIT,
        )
        try:
            if 200 <= _response.status_code < 300:
                return pydantic_v1.parse_obj_as(typing.List[Episode], _response.json())  # type: ignore
            if _response.status_code == 400:
                raise BadRequestError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            if _response.status_code == 500:
                raise InternalServerError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            _response_json = _response.json()
        except JSONDecodeError:
            raise core_api_error_ApiError(status_code=_response.status_code, body=_response.text)
        raise core_api_error_ApiError(status_code=_response.status_code, body=_response_json)

    def add_fact_triple(
        self,
        *,
        fact: str,
        fact_name: str,
        target_node_name: str,
        created_at: typing.Optional[str] = OMIT,
        expired_at: typing.Optional[str] = OMIT,
        fact_uuid: typing.Optional[str] = OMIT,
        group_id: typing.Optional[str] = OMIT,
        invalid_at: typing.Optional[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_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)

        target_node_name : str
            The name of the target node to add

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

        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

        group_id : typing.Optional[str]

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

        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_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.client import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.add_fact_triple(
            fact="fact",
            fact_name="fact_name",
            target_node_name="target_node_name",
        )
        """
        _response = self._client_wrapper.httpx_client.request(
            "graph/add-fact-triple",
            method="POST",
            json={
                "created_at": created_at,
                "expired_at": expired_at,
                "fact": fact,
                "fact_name": fact_name,
                "fact_uuid": fact_uuid,
                "group_id": group_id,
                "invalid_at": invalid_at,
                "source_node_name": source_node_name,
                "source_node_summary": source_node_summary,
                "source_node_uuid": source_node_uuid,
                "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,
            omit=OMIT,
        )
        try:
            if 200 <= _response.status_code < 300:
                return pydantic_v1.parse_obj_as(AddTripleResponse, _response.json())  # type: ignore
            if _response.status_code == 400:
                raise BadRequestError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            if _response.status_code == 500:
                raise InternalServerError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            _response_json = _response.json()
        except JSONDecodeError:
            raise core_api_error_ApiError(status_code=_response.status_code, body=_response.text)
        raise core_api_error_ApiError(status_code=_response.status_code, body=_response_json)

    def search(
        self,
        *,
        query: str,
        center_node_uuid: typing.Optional[str] = OMIT,
        group_id: typing.Optional[str] = OMIT,
        limit: typing.Optional[int] = OMIT,
        min_fact_rating: typing.Optional[float] = OMIT,
        min_score: typing.Optional[float] = OMIT,
        mmr_lambda: typing.Optional[float] = OMIT,
        reranker: typing.Optional[Reranker] = 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)

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

        group_id : typing.Optional[str]
            one of user_id or group_id must be provided

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

        min_fact_rating : typing.Optional[float]
            The minimum rating by which to filter relevant facts

        min_score : typing.Optional[float]
            Deprecated

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

        reranker : typing.Optional[Reranker]
            Defaults to RRF

        scope : typing.Optional[GraphSearchScope]
            Defaults to Edges. Communities will be added in the future.

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

        user_id : typing.Optional[str]
            one of user_id or group_id must be provided

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

        Returns
        -------
        GraphSearchResults
            Graph search results

        Examples
        --------
        from zep_cloud.client import Zep

        client = Zep(
            api_key="YOUR_API_KEY",
        )
        client.graph.search(
            query="query",
        )
        """
        _response = self._client_wrapper.httpx_client.request(
            "graph/search",
            method="POST",
            json={
                "center_node_uuid": center_node_uuid,
                "group_id": group_id,
                "limit": limit,
                "min_fact_rating": min_fact_rating,
                "min_score": min_score,
                "mmr_lambda": mmr_lambda,
                "query": query,
                "reranker": reranker,
                "scope": scope,
                "search_filters": search_filters,
                "user_id": user_id,
            },
            request_options=request_options,
            omit=OMIT,
        )
        try:
            if 200 <= _response.status_code < 300:
                return pydantic_v1.parse_obj_as(GraphSearchResults, _response.json())  # type: ignore
            if _response.status_code == 400:
                raise BadRequestError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            if _response.status_code == 500:
                raise InternalServerError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            _response_json = _response.json()
        except JSONDecodeError:
            raise core_api_error_ApiError(status_code=_response.status_code, body=_response.text)
        raise core_api_error_ApiError(status_code=_response.status_code, body=_response_json)


class AsyncGraphClient:
    def __init__(self, *, client_wrapper: AsyncClientWrapper):
        self._client_wrapper = client_wrapper
        self.edge = AsyncEdgeClient(client_wrapper=self._client_wrapper)
        self.episode = AsyncEpisodeClient(client_wrapper=self._client_wrapper)
        self.node = AsyncNodeClient(client_wrapper=self._client_wrapper)

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

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

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

        Examples
        --------
        import asyncio

        from zep_cloud.client import AsyncZep

        client = AsyncZep(
            api_key="YOUR_API_KEY",
        )


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


        asyncio.run(main())
        """
        _response = await self._client_wrapper.httpx_client.request(
            "entity-types", method="GET", request_options=request_options
        )
        try:
            if 200 <= _response.status_code < 300:
                return pydantic_v1.parse_obj_as(EntityTypeResponse, _response.json())  # type: ignore
            if _response.status_code == 404:
                raise NotFoundError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            if _response.status_code == 500:
                raise InternalServerError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            _response_json = _response.json()
        except JSONDecodeError:
            raise core_api_error_ApiError(status_code=_response.status_code, body=_response.text)
        raise core_api_error_ApiError(status_code=_response.status_code, body=_response_json)

    async def set_entity_types_internal(
        self,
        *,
        entity_types: typing.Optional[typing.Sequence[EntityType]] = OMIT,
        request_options: typing.Optional[RequestOptions] = None
    ) -> SuccessResponse:
        """
        Sets the entity types for a project, replacing any existing ones.

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

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

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

        Examples
        --------
        import asyncio

        from zep_cloud.client 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._client_wrapper.httpx_client.request(
            "entity-types",
            method="PUT",
            json={"entity_types": entity_types},
            request_options=request_options,
            omit=OMIT,
        )
        try:
            if 200 <= _response.status_code < 300:
                return pydantic_v1.parse_obj_as(SuccessResponse, _response.json())  # type: ignore
            if _response.status_code == 400:
                raise BadRequestError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            if _response.status_code == 500:
                raise InternalServerError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            _response_json = _response.json()
        except JSONDecodeError:
            raise core_api_error_ApiError(status_code=_response.status_code, body=_response.text)
        raise core_api_error_ApiError(status_code=_response.status_code, body=_response_json)

    async def add(
        self,
        *,
        data: str,
        type: GraphDataType,
        group_id: typing.Optional[str] = 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. Note: each subscription tier has different limits on the amount of data that can be added to the graph please refer to the pricing page for more information.

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

        type : GraphDataType

        group_id : typing.Optional[str]

        source_description : typing.Optional[str]

        user_id : typing.Optional[str]

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

        Returns
        -------
        Episode
            Added episode

        Examples
        --------
        import asyncio

        from zep_cloud.client 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._client_wrapper.httpx_client.request(
            "graph",
            method="POST",
            json={
                "data": data,
                "group_id": group_id,
                "source_description": source_description,
                "type": type,
                "user_id": user_id,
            },
            request_options=request_options,
            omit=OMIT,
        )
        try:
            if 200 <= _response.status_code < 300:
                return pydantic_v1.parse_obj_as(Episode, _response.json())  # type: ignore
            if _response.status_code == 400:
                raise BadRequestError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            if _response.status_code == 500:
                raise InternalServerError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            _response_json = _response.json()
        except JSONDecodeError:
            raise core_api_error_ApiError(status_code=_response.status_code, body=_response.text)
        raise core_api_error_ApiError(status_code=_response.status_code, body=_response_json)

    async def add_batch(
        self,
        *,
        episodes: typing.Sequence[EpisodeData],
        group_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 (each episode processed concurrently). Note: each subscription tier has different limits on the amount of data that can be added to the graph please refer to the pricing page for more information.

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

        group_id : typing.Optional[str]

        user_id : typing.Optional[str]

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

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

        Examples
        --------
        import asyncio

        from zep_cloud import EpisodeData
        from zep_cloud.client import AsyncZep

        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._client_wrapper.httpx_client.request(
            "graph-batch",
            method="POST",
            json={"episodes": episodes, "group_id": group_id, "user_id": user_id},
            request_options=request_options,
            omit=OMIT,
        )
        try:
            if 200 <= _response.status_code < 300:
                return pydantic_v1.parse_obj_as(typing.List[Episode], _response.json())  # type: ignore
            if _response.status_code == 400:
                raise BadRequestError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            if _response.status_code == 500:
                raise InternalServerError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            _response_json = _response.json()
        except JSONDecodeError:
            raise core_api_error_ApiError(status_code=_response.status_code, body=_response.text)
        raise core_api_error_ApiError(status_code=_response.status_code, body=_response_json)

    async def add_fact_triple(
        self,
        *,
        fact: str,
        fact_name: str,
        target_node_name: str,
        created_at: typing.Optional[str] = OMIT,
        expired_at: typing.Optional[str] = OMIT,
        fact_uuid: typing.Optional[str] = OMIT,
        group_id: typing.Optional[str] = OMIT,
        invalid_at: typing.Optional[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_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)

        target_node_name : str
            The name of the target node to add

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

        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

        group_id : typing.Optional[str]

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

        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_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.client 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",
                target_node_name="target_node_name",
            )


        asyncio.run(main())
        """
        _response = await self._client_wrapper.httpx_client.request(
            "graph/add-fact-triple",
            method="POST",
            json={
                "created_at": created_at,
                "expired_at": expired_at,
                "fact": fact,
                "fact_name": fact_name,
                "fact_uuid": fact_uuid,
                "group_id": group_id,
                "invalid_at": invalid_at,
                "source_node_name": source_node_name,
                "source_node_summary": source_node_summary,
                "source_node_uuid": source_node_uuid,
                "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,
            omit=OMIT,
        )
        try:
            if 200 <= _response.status_code < 300:
                return pydantic_v1.parse_obj_as(AddTripleResponse, _response.json())  # type: ignore
            if _response.status_code == 400:
                raise BadRequestError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            if _response.status_code == 500:
                raise InternalServerError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            _response_json = _response.json()
        except JSONDecodeError:
            raise core_api_error_ApiError(status_code=_response.status_code, body=_response.text)
        raise core_api_error_ApiError(status_code=_response.status_code, body=_response_json)

    async def search(
        self,
        *,
        query: str,
        center_node_uuid: typing.Optional[str] = OMIT,
        group_id: typing.Optional[str] = OMIT,
        limit: typing.Optional[int] = OMIT,
        min_fact_rating: typing.Optional[float] = OMIT,
        min_score: typing.Optional[float] = OMIT,
        mmr_lambda: typing.Optional[float] = OMIT,
        reranker: typing.Optional[Reranker] = 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)

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

        group_id : typing.Optional[str]
            one of user_id or group_id must be provided

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

        min_fact_rating : typing.Optional[float]
            The minimum rating by which to filter relevant facts

        min_score : typing.Optional[float]
            Deprecated

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

        reranker : typing.Optional[Reranker]
            Defaults to RRF

        scope : typing.Optional[GraphSearchScope]
            Defaults to Edges. Communities will be added in the future.

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

        user_id : typing.Optional[str]
            one of user_id or group_id must be provided

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

        Returns
        -------
        GraphSearchResults
            Graph search results

        Examples
        --------
        import asyncio

        from zep_cloud.client 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._client_wrapper.httpx_client.request(
            "graph/search",
            method="POST",
            json={
                "center_node_uuid": center_node_uuid,
                "group_id": group_id,
                "limit": limit,
                "min_fact_rating": min_fact_rating,
                "min_score": min_score,
                "mmr_lambda": mmr_lambda,
                "query": query,
                "reranker": reranker,
                "scope": scope,
                "search_filters": search_filters,
                "user_id": user_id,
            },
            request_options=request_options,
            omit=OMIT,
        )
        try:
            if 200 <= _response.status_code < 300:
                return pydantic_v1.parse_obj_as(GraphSearchResults, _response.json())  # type: ignore
            if _response.status_code == 400:
                raise BadRequestError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            if _response.status_code == 500:
                raise InternalServerError(
                    pydantic_v1.parse_obj_as(types_api_error_ApiError, _response.json())  # type: ignore
                )
            _response_json = _response.json()
        except JSONDecodeError:
            raise core_api_error_ApiError(status_code=_response.status_code, body=_response.text)
        raise core_api_error_ApiError(status_code=_response.status_code, body=_response_json)