xtrace_sdk.integrations.xtrace
==============================

.. py:module:: xtrace_sdk.integrations.xtrace


Classes
-------

.. autoapisummary::

   xtrace_sdk.integrations.xtrace.XTraceIntegration


Module Contents
---------------

.. py:class:: XTraceIntegration(org_id, api_key = None, api_url = 'https://api.production.xtrace.ai')

   Bases: :py:obj:`xtrace_sdk.integrations.base.IntegrationBase`


   XTrace integration for retrievers.

   :param XtraceIntegrationBase: Interface for XTrace integrations
   :type XtraceIntegrationBase: class


   .. py:attribute:: org_id


   .. py:attribute:: api_url
      :value: 'https://api.production.xtrace.ai'



   .. py:attribute:: batch_size
      :value: 25



   .. py:attribute:: session
      :type:  Optional[aiohttp.ClientSession]
      :value: None



   .. py:attribute:: _loop
      :type:  Optional[asyncio.AbstractEventLoop]
      :value: None



   .. py:attribute:: _lock


   .. py:attribute:: api_key
      :value: None



   .. py:method:: init_session()
      :async:


      Ensure there is an open ClientSession usable on the current event loop.

      - Create once on first use.
      - If the session was closed or the event loop changed, (re)create it.



   .. py:method:: close()
      :async:



   .. py:method:: __aenter__()
      :async:



   .. py:method:: __aexit__(exc_type, exc, tb)
      :async:



   .. py:method:: get_chunk_by_id(chunk_ids, kb_id)
      :async:


      Get a chunk from the XTrace storage.

      :param chunk_id: chunk id of the chunk to be retrieved
      :type chunk_id: str
      :param kb_id: knowledge base id of the chunk to be retrieved
      :type kb_id: str
      :return: the chunk
      :rtype: Chunk



   .. py:method:: store_index(index)

      Store the index in the storage.




   .. py:method:: _normalize_meta(md)

      Ensure meta_data has all required tags, filling missing ones with None.



   .. py:method:: preprocess_chunk(db, index, update = False)


   .. py:method:: store_db(db, index, kb_id, context_id, concurrent = False)
      :async:


      Store the database in the XTrace storage. Assumes index and db are built together.

      :param db: the database to be stored
      :type db: EncryptedDB
      :param index: the index to be stored
      :type index: Index
      :param kb_id: knowledge base id
      :type kb_id: str
      :param org_id: organization id
      :type org_id: str
      :param concurrent: whether to insert chunks concurrently, defaults to False
      :type concurrent: bool, optional
      :param meta_data: metadata for each chunk, defaults to None
      :type meta_data: list[MetaData], optional
      :raises AssertionError: if the length of db and index are not the same, or if the length of db and meta_data are not the same
      :return: the response from the API
      :rtype: requests.Response



   .. py:method:: insert_chunks(kb_id, chunks, context_id)
      :async:


      Insert a chunk into the XTrace storage.

      :param kb_id: the id of the knowledge base
      :type kb_id: str
      :param chunk_id: the id of the chunk
      :type chunk_id: ChunkID
      :param chunk_content: the content of the chunk
      :type chunk_content: str
      :param meta_data: the metadata of the chunk
      :type meta_data: MetaData
      :param index: the index of the chunk
      :type index: int
      :return: the response from the API
      :rtype: requests.Response



   .. py:method:: load_db()

      Load the database from the storage.




   .. py:method:: load_index()

      Load the index from the storage.




   .. py:method:: compute_hamming_distances_bench(query, pk, **kwargs)
      :async:



   .. py:method:: compute_hamming_distances(query, context_id, **kwargs)
      :async:


      Compute the hamming distances between the query and the data in the XTrace database.

      :param query: encrypted query to be used for comparison
      :type query: int
      :param pk: the public key used for encryption
      :type pk: HomomorphicPublicKey
      :param kb_id: the id of the knowledge base to query
      :type kb_id: str
      :return: list of tuples with the id and the encrypted hamming distance
      :rtype: List[Tuple[ChunkID,int]]



   .. py:method:: get_serialized_exec_context(context_id)
      :async:


      Get the execution context from the XTrace storage.

      :param context_id: the id of the execution context to be retrieved
      :type context_id: str
      :return: the serialized execution context
      :rtype: str



   .. py:method:: store_exec_context(exec_context_dict, context_id)
      :async:


      Store the execution context in the XTrace storage.

      :param exec_context: the execution context to be stored
      :type exec_context: ExecutionContext



   .. py:method:: list_exec_contexts()
      :async:


      List all execution contexts for the organization.

      :return: list of context ids
      :rtype: list[str]



   .. py:method:: update_chunks(index_updates, chunk_updates, context_id, kb_id)
      :async:


      Update chunks in the XTrace storage.

      :param kb_id: the id of the knowledge base
      :type kb_id: str
      :param chunks: list of chunks to be updated
      :type chunks: list[dict]
      :param context_id: the id of the execution context
      :type context_id: str
      :raises AssertionError: if any chunk does not have a chunk_id



   .. py:method:: delete_chunks(chunk_ids, kb_id)
      :async:


      Delete chunks from the XTrace storage.

      :param kb_id: the id of the knowledge base
      :type kb_id: str
      :param chunk_ids: list of chunk ids to be deleted
      :type chunk_ids: list[int]



   .. py:method:: meta_search(kb_id, meta_filter, context_id)
      :async:


      Search chunks in the XTrace storage based on metadata. Only Equality operator is supported for now.

      :param kb_id: the id of the knowledge base
      :type kb_id: str
      :param meta_filter: the metadata filter as a JSON string
      :type meta_filter: str
      :return: list of chunk ids matching the metadata filter
      :rtype: list[int]