# coding: utf-8 """ Hindsight HTTP API HTTP API for Hindsight The version of the OpenAPI document: 0.6.1 Generated by OpenAPI Generator (https://openapi-generator.tech) Do not edit the class manually. """ # noqa: E501 import warnings from pydantic import validate_call, Field, StrictFloat, StrictStr, StrictInt from typing import Any, Dict, List, Optional, Tuple, Union from typing_extensions import Annotated from pydantic import Field, StrictInt, StrictStr, field_validator from typing import Any, List, Optional from typing_extensions import Annotated from hindsight_client_api.models.clear_memory_observations_response import ClearMemoryObservationsResponse from hindsight_client_api.models.delete_response import DeleteResponse from hindsight_client_api.models.graph_data_response import GraphDataResponse from hindsight_client_api.models.list_memory_units_response import ListMemoryUnitsResponse from hindsight_client_api.models.list_tags_response import ListTagsResponse from hindsight_client_api.models.recall_request import RecallRequest from hindsight_client_api.models.recall_response import RecallResponse from hindsight_client_api.models.reflect_request import ReflectRequest from hindsight_client_api.models.reflect_response import ReflectResponse from hindsight_client_api.models.retain_request import RetainRequest from hindsight_client_api.models.retain_response import RetainResponse from hindsight_client_api.api_client import ApiClient, RequestSerialized from hindsight_client_api.api_response import ApiResponse from hindsight_client_api.rest import RESTResponseType class MemoryApi: """NOTE: This class is auto generated by OpenAPI Generator Ref: https://openapi-generator.tech Do not edit the class manually. """ def __init__(self, api_client=None) -> None: if api_client is None: api_client = ApiClient.get_default() self.api_client = api_client @validate_call async def clear_bank_memories( self, bank_id: StrictStr, type: Annotated[Optional[StrictStr], Field(description="Optional fact type filter (world, experience, opinion)")] = None, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> DeleteResponse: """Clear memory bank memories Delete memory units for a memory bank. Optionally filter by type (world, experience, opinion) to delete only specific types. This is a destructive operation that cannot be undone. The bank profile (disposition and background) will be preserved. :param bank_id: (required) :type bank_id: str :param type: Optional fact type filter (world, experience, opinion) :type type: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._clear_bank_memories_serialize( bank_id=bank_id, type=type, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "DeleteResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ).data @validate_call async def clear_bank_memories_with_http_info( self, bank_id: StrictStr, type: Annotated[Optional[StrictStr], Field(description="Optional fact type filter (world, experience, opinion)")] = None, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[DeleteResponse]: """Clear memory bank memories Delete memory units for a memory bank. Optionally filter by type (world, experience, opinion) to delete only specific types. This is a destructive operation that cannot be undone. The bank profile (disposition and background) will be preserved. :param bank_id: (required) :type bank_id: str :param type: Optional fact type filter (world, experience, opinion) :type type: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._clear_bank_memories_serialize( bank_id=bank_id, type=type, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "DeleteResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ) @validate_call async def clear_bank_memories_without_preload_content( self, bank_id: StrictStr, type: Annotated[Optional[StrictStr], Field(description="Optional fact type filter (world, experience, opinion)")] = None, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """Clear memory bank memories Delete memory units for a memory bank. Optionally filter by type (world, experience, opinion) to delete only specific types. This is a destructive operation that cannot be undone. The bank profile (disposition and background) will be preserved. :param bank_id: (required) :type bank_id: str :param type: Optional fact type filter (world, experience, opinion) :type type: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._clear_bank_memories_serialize( bank_id=bank_id, type=type, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "DeleteResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _clear_bank_memories_serialize( self, bank_id, type, authorization, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[ str, Union[str, bytes, List[str], List[bytes], List[Tuple[str, bytes]]] ] = {} _body_params: Optional[bytes] = None # process the path parameters if bank_id is not None: _path_params['bank_id'] = bank_id # process the query parameters if type is not None: _query_params.append(('type', type)) # process the header parameters if authorization is not None: _header_params['authorization'] = authorization # process the form parameters # process the body parameter # set the HTTP header `Accept` if 'Accept' not in _header_params: _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/json' ] ) # authentication setting _auth_settings: List[str] = [ ] return self.api_client.param_serialize( method='DELETE', resource_path='/v1/default/banks/{bank_id}/memories', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth ) @validate_call async def clear_memory_observations( self, bank_id: StrictStr, memory_id: StrictStr, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ClearMemoryObservationsResponse: """Clear observations for a memory Delete all observations derived from a specific memory and reset it for re-consolidation. The memory itself is not deleted. A consolidation job is triggered automatically so the memory will produce fresh observations on the next consolidation run. :param bank_id: (required) :type bank_id: str :param memory_id: (required) :type memory_id: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._clear_memory_observations_serialize( bank_id=bank_id, memory_id=memory_id, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "ClearMemoryObservationsResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ).data @validate_call async def clear_memory_observations_with_http_info( self, bank_id: StrictStr, memory_id: StrictStr, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[ClearMemoryObservationsResponse]: """Clear observations for a memory Delete all observations derived from a specific memory and reset it for re-consolidation. The memory itself is not deleted. A consolidation job is triggered automatically so the memory will produce fresh observations on the next consolidation run. :param bank_id: (required) :type bank_id: str :param memory_id: (required) :type memory_id: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._clear_memory_observations_serialize( bank_id=bank_id, memory_id=memory_id, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "ClearMemoryObservationsResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ) @validate_call async def clear_memory_observations_without_preload_content( self, bank_id: StrictStr, memory_id: StrictStr, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """Clear observations for a memory Delete all observations derived from a specific memory and reset it for re-consolidation. The memory itself is not deleted. A consolidation job is triggered automatically so the memory will produce fresh observations on the next consolidation run. :param bank_id: (required) :type bank_id: str :param memory_id: (required) :type memory_id: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._clear_memory_observations_serialize( bank_id=bank_id, memory_id=memory_id, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "ClearMemoryObservationsResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _clear_memory_observations_serialize( self, bank_id, memory_id, authorization, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[ str, Union[str, bytes, List[str], List[bytes], List[Tuple[str, bytes]]] ] = {} _body_params: Optional[bytes] = None # process the path parameters if bank_id is not None: _path_params['bank_id'] = bank_id if memory_id is not None: _path_params['memory_id'] = memory_id # process the query parameters # process the header parameters if authorization is not None: _header_params['authorization'] = authorization # process the form parameters # process the body parameter # set the HTTP header `Accept` if 'Accept' not in _header_params: _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/json' ] ) # authentication setting _auth_settings: List[str] = [ ] return self.api_client.param_serialize( method='DELETE', resource_path='/v1/default/banks/{bank_id}/memories/{memory_id}/observations', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth ) @validate_call async def get_graph( self, bank_id: StrictStr, type: Optional[StrictStr] = None, limit: Optional[StrictInt] = None, q: Optional[StrictStr] = None, tags: Optional[List[Optional[StrictStr]]] = None, tags_match: Optional[StrictStr] = None, document_id: Optional[StrictStr] = None, chunk_id: Optional[StrictStr] = None, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> GraphDataResponse: """Get memory graph data Retrieve graph data for visualization, optionally filtered by type (world/experience/opinion). :param bank_id: (required) :type bank_id: str :param type: :type type: str :param limit: :type limit: int :param q: :type q: str :param tags: :type tags: List[Optional[str]] :param tags_match: :type tags_match: str :param document_id: :type document_id: str :param chunk_id: :type chunk_id: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_graph_serialize( bank_id=bank_id, type=type, limit=limit, q=q, tags=tags, tags_match=tags_match, document_id=document_id, chunk_id=chunk_id, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "GraphDataResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ).data @validate_call async def get_graph_with_http_info( self, bank_id: StrictStr, type: Optional[StrictStr] = None, limit: Optional[StrictInt] = None, q: Optional[StrictStr] = None, tags: Optional[List[Optional[StrictStr]]] = None, tags_match: Optional[StrictStr] = None, document_id: Optional[StrictStr] = None, chunk_id: Optional[StrictStr] = None, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[GraphDataResponse]: """Get memory graph data Retrieve graph data for visualization, optionally filtered by type (world/experience/opinion). :param bank_id: (required) :type bank_id: str :param type: :type type: str :param limit: :type limit: int :param q: :type q: str :param tags: :type tags: List[Optional[str]] :param tags_match: :type tags_match: str :param document_id: :type document_id: str :param chunk_id: :type chunk_id: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_graph_serialize( bank_id=bank_id, type=type, limit=limit, q=q, tags=tags, tags_match=tags_match, document_id=document_id, chunk_id=chunk_id, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "GraphDataResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ) @validate_call async def get_graph_without_preload_content( self, bank_id: StrictStr, type: Optional[StrictStr] = None, limit: Optional[StrictInt] = None, q: Optional[StrictStr] = None, tags: Optional[List[Optional[StrictStr]]] = None, tags_match: Optional[StrictStr] = None, document_id: Optional[StrictStr] = None, chunk_id: Optional[StrictStr] = None, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """Get memory graph data Retrieve graph data for visualization, optionally filtered by type (world/experience/opinion). :param bank_id: (required) :type bank_id: str :param type: :type type: str :param limit: :type limit: int :param q: :type q: str :param tags: :type tags: List[Optional[str]] :param tags_match: :type tags_match: str :param document_id: :type document_id: str :param chunk_id: :type chunk_id: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_graph_serialize( bank_id=bank_id, type=type, limit=limit, q=q, tags=tags, tags_match=tags_match, document_id=document_id, chunk_id=chunk_id, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "GraphDataResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _get_graph_serialize( self, bank_id, type, limit, q, tags, tags_match, document_id, chunk_id, authorization, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { 'tags': 'multi', } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[ str, Union[str, bytes, List[str], List[bytes], List[Tuple[str, bytes]]] ] = {} _body_params: Optional[bytes] = None # process the path parameters if bank_id is not None: _path_params['bank_id'] = bank_id # process the query parameters if type is not None: _query_params.append(('type', type)) if limit is not None: _query_params.append(('limit', limit)) if q is not None: _query_params.append(('q', q)) if tags is not None: _query_params.append(('tags', tags)) if tags_match is not None: _query_params.append(('tags_match', tags_match)) if document_id is not None: _query_params.append(('document_id', document_id)) if chunk_id is not None: _query_params.append(('chunk_id', chunk_id)) # process the header parameters if authorization is not None: _header_params['authorization'] = authorization # process the form parameters # process the body parameter # set the HTTP header `Accept` if 'Accept' not in _header_params: _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/json' ] ) # authentication setting _auth_settings: List[str] = [ ] return self.api_client.param_serialize( method='GET', resource_path='/v1/default/banks/{bank_id}/graph', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth ) @validate_call async def get_memory( self, bank_id: StrictStr, memory_id: StrictStr, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> object: """Get memory unit Get a single memory unit by ID with all its metadata including entities and tags. Note: the 'history' field is deprecated and always returns an empty list - use GET /memories/{memory_id}/history instead. :param bank_id: (required) :type bank_id: str :param memory_id: (required) :type memory_id: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_memory_serialize( bank_id=bank_id, memory_id=memory_id, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "object", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ).data @validate_call async def get_memory_with_http_info( self, bank_id: StrictStr, memory_id: StrictStr, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[object]: """Get memory unit Get a single memory unit by ID with all its metadata including entities and tags. Note: the 'history' field is deprecated and always returns an empty list - use GET /memories/{memory_id}/history instead. :param bank_id: (required) :type bank_id: str :param memory_id: (required) :type memory_id: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_memory_serialize( bank_id=bank_id, memory_id=memory_id, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "object", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ) @validate_call async def get_memory_without_preload_content( self, bank_id: StrictStr, memory_id: StrictStr, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """Get memory unit Get a single memory unit by ID with all its metadata including entities and tags. Note: the 'history' field is deprecated and always returns an empty list - use GET /memories/{memory_id}/history instead. :param bank_id: (required) :type bank_id: str :param memory_id: (required) :type memory_id: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_memory_serialize( bank_id=bank_id, memory_id=memory_id, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "object", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _get_memory_serialize( self, bank_id, memory_id, authorization, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[ str, Union[str, bytes, List[str], List[bytes], List[Tuple[str, bytes]]] ] = {} _body_params: Optional[bytes] = None # process the path parameters if bank_id is not None: _path_params['bank_id'] = bank_id if memory_id is not None: _path_params['memory_id'] = memory_id # process the query parameters # process the header parameters if authorization is not None: _header_params['authorization'] = authorization # process the form parameters # process the body parameter # set the HTTP header `Accept` if 'Accept' not in _header_params: _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/json' ] ) # authentication setting _auth_settings: List[str] = [ ] return self.api_client.param_serialize( method='GET', resource_path='/v1/default/banks/{bank_id}/memories/{memory_id}', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth ) @validate_call async def get_observation_history( self, bank_id: StrictStr, memory_id: StrictStr, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> object: """Get observation history Get the full history of an observation, with each change's source facts resolved to their text. :param bank_id: (required) :type bank_id: str :param memory_id: (required) :type memory_id: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_observation_history_serialize( bank_id=bank_id, memory_id=memory_id, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "object", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ).data @validate_call async def get_observation_history_with_http_info( self, bank_id: StrictStr, memory_id: StrictStr, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[object]: """Get observation history Get the full history of an observation, with each change's source facts resolved to their text. :param bank_id: (required) :type bank_id: str :param memory_id: (required) :type memory_id: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_observation_history_serialize( bank_id=bank_id, memory_id=memory_id, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "object", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ) @validate_call async def get_observation_history_without_preload_content( self, bank_id: StrictStr, memory_id: StrictStr, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """Get observation history Get the full history of an observation, with each change's source facts resolved to their text. :param bank_id: (required) :type bank_id: str :param memory_id: (required) :type memory_id: str :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_observation_history_serialize( bank_id=bank_id, memory_id=memory_id, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "object", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _get_observation_history_serialize( self, bank_id, memory_id, authorization, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[ str, Union[str, bytes, List[str], List[bytes], List[Tuple[str, bytes]]] ] = {} _body_params: Optional[bytes] = None # process the path parameters if bank_id is not None: _path_params['bank_id'] = bank_id if memory_id is not None: _path_params['memory_id'] = memory_id # process the query parameters # process the header parameters if authorization is not None: _header_params['authorization'] = authorization # process the form parameters # process the body parameter # set the HTTP header `Accept` if 'Accept' not in _header_params: _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/json' ] ) # authentication setting _auth_settings: List[str] = [ ] return self.api_client.param_serialize( method='GET', resource_path='/v1/default/banks/{bank_id}/memories/{memory_id}/history', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth ) @validate_call async def list_memories( self, bank_id: StrictStr, type: Optional[StrictStr] = None, q: Optional[StrictStr] = None, consolidation_state: Optional[StrictStr] = None, limit: Optional[StrictInt] = None, offset: Optional[StrictInt] = None, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ListMemoryUnitsResponse: """List memory units List memory units with pagination and optional full-text search. Supports filtering by type. Results are sorted by most recent first (mentioned_at DESC, then created_at DESC). :param bank_id: (required) :type bank_id: str :param type: :type type: str :param q: :type q: str :param consolidation_state: :type consolidation_state: str :param limit: :type limit: int :param offset: :type offset: int :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._list_memories_serialize( bank_id=bank_id, type=type, q=q, consolidation_state=consolidation_state, limit=limit, offset=offset, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "ListMemoryUnitsResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ).data @validate_call async def list_memories_with_http_info( self, bank_id: StrictStr, type: Optional[StrictStr] = None, q: Optional[StrictStr] = None, consolidation_state: Optional[StrictStr] = None, limit: Optional[StrictInt] = None, offset: Optional[StrictInt] = None, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[ListMemoryUnitsResponse]: """List memory units List memory units with pagination and optional full-text search. Supports filtering by type. Results are sorted by most recent first (mentioned_at DESC, then created_at DESC). :param bank_id: (required) :type bank_id: str :param type: :type type: str :param q: :type q: str :param consolidation_state: :type consolidation_state: str :param limit: :type limit: int :param offset: :type offset: int :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._list_memories_serialize( bank_id=bank_id, type=type, q=q, consolidation_state=consolidation_state, limit=limit, offset=offset, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "ListMemoryUnitsResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ) @validate_call async def list_memories_without_preload_content( self, bank_id: StrictStr, type: Optional[StrictStr] = None, q: Optional[StrictStr] = None, consolidation_state: Optional[StrictStr] = None, limit: Optional[StrictInt] = None, offset: Optional[StrictInt] = None, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """List memory units List memory units with pagination and optional full-text search. Supports filtering by type. Results are sorted by most recent first (mentioned_at DESC, then created_at DESC). :param bank_id: (required) :type bank_id: str :param type: :type type: str :param q: :type q: str :param consolidation_state: :type consolidation_state: str :param limit: :type limit: int :param offset: :type offset: int :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._list_memories_serialize( bank_id=bank_id, type=type, q=q, consolidation_state=consolidation_state, limit=limit, offset=offset, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "ListMemoryUnitsResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _list_memories_serialize( self, bank_id, type, q, consolidation_state, limit, offset, authorization, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[ str, Union[str, bytes, List[str], List[bytes], List[Tuple[str, bytes]]] ] = {} _body_params: Optional[bytes] = None # process the path parameters if bank_id is not None: _path_params['bank_id'] = bank_id # process the query parameters if type is not None: _query_params.append(('type', type)) if q is not None: _query_params.append(('q', q)) if consolidation_state is not None: _query_params.append(('consolidation_state', consolidation_state)) if limit is not None: _query_params.append(('limit', limit)) if offset is not None: _query_params.append(('offset', offset)) # process the header parameters if authorization is not None: _header_params['authorization'] = authorization # process the form parameters # process the body parameter # set the HTTP header `Accept` if 'Accept' not in _header_params: _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/json' ] ) # authentication setting _auth_settings: List[str] = [ ] return self.api_client.param_serialize( method='GET', resource_path='/v1/default/banks/{bank_id}/memories/list', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth ) @validate_call async def list_tags( self, bank_id: StrictStr, q: Annotated[Optional[StrictStr], Field(description="Wildcard pattern to filter tags (e.g., 'user:*' for user:alice, '*-admin' for role-admin). Use '*' as wildcard. Case-insensitive.")] = None, source: Annotated[Optional[StrictStr], Field(description="Where to read tags from: 'memories' (memory_units, default) or 'mental_models'.")] = None, limit: Annotated[Optional[StrictInt], Field(description="Maximum number of tags to return")] = None, offset: Annotated[Optional[StrictInt], Field(description="Offset for pagination")] = None, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ListTagsResponse: """List tags List all unique tags in a memory bank with usage counts. Supports wildcard search using '*' (e.g., 'user:*', '*-fred', 'tag*-2'). Case-insensitive. Use `source=mental_models` to list tags used on mental models instead of memories. :param bank_id: (required) :type bank_id: str :param q: Wildcard pattern to filter tags (e.g., 'user:*' for user:alice, '*-admin' for role-admin). Use '*' as wildcard. Case-insensitive. :type q: str :param source: Where to read tags from: 'memories' (memory_units, default) or 'mental_models'. :type source: str :param limit: Maximum number of tags to return :type limit: int :param offset: Offset for pagination :type offset: int :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._list_tags_serialize( bank_id=bank_id, q=q, source=source, limit=limit, offset=offset, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "ListTagsResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ).data @validate_call async def list_tags_with_http_info( self, bank_id: StrictStr, q: Annotated[Optional[StrictStr], Field(description="Wildcard pattern to filter tags (e.g., 'user:*' for user:alice, '*-admin' for role-admin). Use '*' as wildcard. Case-insensitive.")] = None, source: Annotated[Optional[StrictStr], Field(description="Where to read tags from: 'memories' (memory_units, default) or 'mental_models'.")] = None, limit: Annotated[Optional[StrictInt], Field(description="Maximum number of tags to return")] = None, offset: Annotated[Optional[StrictInt], Field(description="Offset for pagination")] = None, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[ListTagsResponse]: """List tags List all unique tags in a memory bank with usage counts. Supports wildcard search using '*' (e.g., 'user:*', '*-fred', 'tag*-2'). Case-insensitive. Use `source=mental_models` to list tags used on mental models instead of memories. :param bank_id: (required) :type bank_id: str :param q: Wildcard pattern to filter tags (e.g., 'user:*' for user:alice, '*-admin' for role-admin). Use '*' as wildcard. Case-insensitive. :type q: str :param source: Where to read tags from: 'memories' (memory_units, default) or 'mental_models'. :type source: str :param limit: Maximum number of tags to return :type limit: int :param offset: Offset for pagination :type offset: int :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._list_tags_serialize( bank_id=bank_id, q=q, source=source, limit=limit, offset=offset, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "ListTagsResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ) @validate_call async def list_tags_without_preload_content( self, bank_id: StrictStr, q: Annotated[Optional[StrictStr], Field(description="Wildcard pattern to filter tags (e.g., 'user:*' for user:alice, '*-admin' for role-admin). Use '*' as wildcard. Case-insensitive.")] = None, source: Annotated[Optional[StrictStr], Field(description="Where to read tags from: 'memories' (memory_units, default) or 'mental_models'.")] = None, limit: Annotated[Optional[StrictInt], Field(description="Maximum number of tags to return")] = None, offset: Annotated[Optional[StrictInt], Field(description="Offset for pagination")] = None, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """List tags List all unique tags in a memory bank with usage counts. Supports wildcard search using '*' (e.g., 'user:*', '*-fred', 'tag*-2'). Case-insensitive. Use `source=mental_models` to list tags used on mental models instead of memories. :param bank_id: (required) :type bank_id: str :param q: Wildcard pattern to filter tags (e.g., 'user:*' for user:alice, '*-admin' for role-admin). Use '*' as wildcard. Case-insensitive. :type q: str :param source: Where to read tags from: 'memories' (memory_units, default) or 'mental_models'. :type source: str :param limit: Maximum number of tags to return :type limit: int :param offset: Offset for pagination :type offset: int :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._list_tags_serialize( bank_id=bank_id, q=q, source=source, limit=limit, offset=offset, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "ListTagsResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _list_tags_serialize( self, bank_id, q, source, limit, offset, authorization, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[ str, Union[str, bytes, List[str], List[bytes], List[Tuple[str, bytes]]] ] = {} _body_params: Optional[bytes] = None # process the path parameters if bank_id is not None: _path_params['bank_id'] = bank_id # process the query parameters if q is not None: _query_params.append(('q', q)) if source is not None: _query_params.append(('source', source)) if limit is not None: _query_params.append(('limit', limit)) if offset is not None: _query_params.append(('offset', offset)) # process the header parameters if authorization is not None: _header_params['authorization'] = authorization # process the form parameters # process the body parameter # set the HTTP header `Accept` if 'Accept' not in _header_params: _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/json' ] ) # authentication setting _auth_settings: List[str] = [ ] return self.api_client.param_serialize( method='GET', resource_path='/v1/default/banks/{bank_id}/tags', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth ) @validate_call async def recall_memories( self, bank_id: StrictStr, recall_request: RecallRequest, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RecallResponse: """Recall memory Recall memory using semantic similarity and spreading activation. The type parameter is optional and must be one of: - `world`: General knowledge about people, places, events, and things that happen - `experience`: Memories about experience, conversations, actions taken, and tasks performed :param bank_id: (required) :type bank_id: str :param recall_request: (required) :type recall_request: RecallRequest :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._recall_memories_serialize( bank_id=bank_id, recall_request=recall_request, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "RecallResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ).data @validate_call async def recall_memories_with_http_info( self, bank_id: StrictStr, recall_request: RecallRequest, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[RecallResponse]: """Recall memory Recall memory using semantic similarity and spreading activation. The type parameter is optional and must be one of: - `world`: General knowledge about people, places, events, and things that happen - `experience`: Memories about experience, conversations, actions taken, and tasks performed :param bank_id: (required) :type bank_id: str :param recall_request: (required) :type recall_request: RecallRequest :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._recall_memories_serialize( bank_id=bank_id, recall_request=recall_request, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "RecallResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ) @validate_call async def recall_memories_without_preload_content( self, bank_id: StrictStr, recall_request: RecallRequest, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """Recall memory Recall memory using semantic similarity and spreading activation. The type parameter is optional and must be one of: - `world`: General knowledge about people, places, events, and things that happen - `experience`: Memories about experience, conversations, actions taken, and tasks performed :param bank_id: (required) :type bank_id: str :param recall_request: (required) :type recall_request: RecallRequest :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._recall_memories_serialize( bank_id=bank_id, recall_request=recall_request, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "RecallResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _recall_memories_serialize( self, bank_id, recall_request, authorization, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[ str, Union[str, bytes, List[str], List[bytes], List[Tuple[str, bytes]]] ] = {} _body_params: Optional[bytes] = None # process the path parameters if bank_id is not None: _path_params['bank_id'] = bank_id # process the query parameters # process the header parameters if authorization is not None: _header_params['authorization'] = authorization # process the form parameters # process the body parameter if recall_request is not None: _body_params = recall_request # set the HTTP header `Accept` if 'Accept' not in _header_params: _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/json' ] ) # set the HTTP header `Content-Type` if _content_type: _header_params['Content-Type'] = _content_type else: _default_content_type = ( self.api_client.select_header_content_type( [ 'application/json' ] ) ) if _default_content_type is not None: _header_params['Content-Type'] = _default_content_type # authentication setting _auth_settings: List[str] = [ ] return self.api_client.param_serialize( method='POST', resource_path='/v1/default/banks/{bank_id}/memories/recall', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth ) @validate_call async def reflect( self, bank_id: StrictStr, reflect_request: ReflectRequest, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ReflectResponse: """Reflect and generate answer Reflect and formulate an answer using bank identity, world facts, and opinions. This endpoint: 1. Retrieves experience (conversations and events) 2. Retrieves world facts relevant to the query 3. Retrieves existing opinions (bank's perspectives) 4. Uses LLM to formulate a contextual answer 5. Returns plain text answer and the facts used :param bank_id: (required) :type bank_id: str :param reflect_request: (required) :type reflect_request: ReflectRequest :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._reflect_serialize( bank_id=bank_id, reflect_request=reflect_request, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "ReflectResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ).data @validate_call async def reflect_with_http_info( self, bank_id: StrictStr, reflect_request: ReflectRequest, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[ReflectResponse]: """Reflect and generate answer Reflect and formulate an answer using bank identity, world facts, and opinions. This endpoint: 1. Retrieves experience (conversations and events) 2. Retrieves world facts relevant to the query 3. Retrieves existing opinions (bank's perspectives) 4. Uses LLM to formulate a contextual answer 5. Returns plain text answer and the facts used :param bank_id: (required) :type bank_id: str :param reflect_request: (required) :type reflect_request: ReflectRequest :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._reflect_serialize( bank_id=bank_id, reflect_request=reflect_request, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "ReflectResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ) @validate_call async def reflect_without_preload_content( self, bank_id: StrictStr, reflect_request: ReflectRequest, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """Reflect and generate answer Reflect and formulate an answer using bank identity, world facts, and opinions. This endpoint: 1. Retrieves experience (conversations and events) 2. Retrieves world facts relevant to the query 3. Retrieves existing opinions (bank's perspectives) 4. Uses LLM to formulate a contextual answer 5. Returns plain text answer and the facts used :param bank_id: (required) :type bank_id: str :param reflect_request: (required) :type reflect_request: ReflectRequest :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._reflect_serialize( bank_id=bank_id, reflect_request=reflect_request, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "ReflectResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _reflect_serialize( self, bank_id, reflect_request, authorization, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[ str, Union[str, bytes, List[str], List[bytes], List[Tuple[str, bytes]]] ] = {} _body_params: Optional[bytes] = None # process the path parameters if bank_id is not None: _path_params['bank_id'] = bank_id # process the query parameters # process the header parameters if authorization is not None: _header_params['authorization'] = authorization # process the form parameters # process the body parameter if reflect_request is not None: _body_params = reflect_request # set the HTTP header `Accept` if 'Accept' not in _header_params: _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/json' ] ) # set the HTTP header `Content-Type` if _content_type: _header_params['Content-Type'] = _content_type else: _default_content_type = ( self.api_client.select_header_content_type( [ 'application/json' ] ) ) if _default_content_type is not None: _header_params['Content-Type'] = _default_content_type # authentication setting _auth_settings: List[str] = [ ] return self.api_client.param_serialize( method='POST', resource_path='/v1/default/banks/{bank_id}/reflect', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth ) @validate_call async def retain_memories( self, bank_id: StrictStr, retain_request: RetainRequest, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RetainResponse: """Retain memories Retain memory items with automatic fact extraction. This is the main endpoint for storing memories. It supports both synchronous and asynchronous processing via the `async` parameter. **Features:** - Efficient batch processing - Automatic fact extraction from natural language - Entity recognition and linking - Document tracking with automatic upsert (when document_id is provided) - Temporal and semantic linking - Optional asynchronous processing **The system automatically:** 1. Extracts semantic facts from the content 2. Generates embeddings 3. Deduplicates similar facts 4. Creates temporal, semantic, and entity links 5. Tracks document metadata **When `async=true`:** Returns immediately after queuing. Use the operations endpoint to monitor progress. **When `async=false` (default):** Waits for processing to complete. **Note:** If a memory item has a `document_id` that already exists, the old document and its memory units will be deleted before creating new ones (upsert behavior). :param bank_id: (required) :type bank_id: str :param retain_request: (required) :type retain_request: RetainRequest :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._retain_memories_serialize( bank_id=bank_id, retain_request=retain_request, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "RetainResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ).data @validate_call async def retain_memories_with_http_info( self, bank_id: StrictStr, retain_request: RetainRequest, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[RetainResponse]: """Retain memories Retain memory items with automatic fact extraction. This is the main endpoint for storing memories. It supports both synchronous and asynchronous processing via the `async` parameter. **Features:** - Efficient batch processing - Automatic fact extraction from natural language - Entity recognition and linking - Document tracking with automatic upsert (when document_id is provided) - Temporal and semantic linking - Optional asynchronous processing **The system automatically:** 1. Extracts semantic facts from the content 2. Generates embeddings 3. Deduplicates similar facts 4. Creates temporal, semantic, and entity links 5. Tracks document metadata **When `async=true`:** Returns immediately after queuing. Use the operations endpoint to monitor progress. **When `async=false` (default):** Waits for processing to complete. **Note:** If a memory item has a `document_id` that already exists, the old document and its memory units will be deleted before creating new ones (upsert behavior). :param bank_id: (required) :type bank_id: str :param retain_request: (required) :type retain_request: RetainRequest :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._retain_memories_serialize( bank_id=bank_id, retain_request=retain_request, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "RetainResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) await response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, ) @validate_call async def retain_memories_without_preload_content( self, bank_id: StrictStr, retain_request: RetainRequest, authorization: Optional[StrictStr] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """Retain memories Retain memory items with automatic fact extraction. This is the main endpoint for storing memories. It supports both synchronous and asynchronous processing via the `async` parameter. **Features:** - Efficient batch processing - Automatic fact extraction from natural language - Entity recognition and linking - Document tracking with automatic upsert (when document_id is provided) - Temporal and semantic linking - Optional asynchronous processing **The system automatically:** 1. Extracts semantic facts from the content 2. Generates embeddings 3. Deduplicates similar facts 4. Creates temporal, semantic, and entity links 5. Tracks document metadata **When `async=true`:** Returns immediately after queuing. Use the operations endpoint to monitor progress. **When `async=false` (default):** Waits for processing to complete. **Note:** If a memory item has a `document_id` that already exists, the old document and its memory units will be deleted before creating new ones (upsert behavior). :param bank_id: (required) :type bank_id: str :param retain_request: (required) :type retain_request: RetainRequest :param authorization: :type authorization: str :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :type _request_timeout: int, tuple(int, int), optional :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._retain_memories_serialize( bank_id=bank_id, retain_request=retain_request, authorization=authorization, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "RetainResponse", '422': "HTTPValidationError", } response_data = await self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _retain_memories_serialize( self, bank_id, retain_request, authorization, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[ str, Union[str, bytes, List[str], List[bytes], List[Tuple[str, bytes]]] ] = {} _body_params: Optional[bytes] = None # process the path parameters if bank_id is not None: _path_params['bank_id'] = bank_id # process the query parameters # process the header parameters if authorization is not None: _header_params['authorization'] = authorization # process the form parameters # process the body parameter if retain_request is not None: _body_params = retain_request # set the HTTP header `Accept` if 'Accept' not in _header_params: _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/json' ] ) # set the HTTP header `Content-Type` if _content_type: _header_params['Content-Type'] = _content_type else: _default_content_type = ( self.api_client.select_header_content_type( [ 'application/json' ] ) ) if _default_content_type is not None: _header_params['Content-Type'] = _default_content_type # authentication setting _auth_settings: List[str] = [ ] return self.api_client.param_serialize( method='POST', resource_path='/v1/default/banks/{bank_id}/memories', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth )