​

literalai.api.__init__

​

LiteralAPI Objects

class LiteralAPI(BaseLiteralAPI)

​

make_gql_call

def make_gql_call(description: str, query: str, variables: Dict[str,
                                                                Any]) -> Dict

Executes a GraphQL call with the provided query and variables.

Arguments:

• description str - Description of the GraphQL operation for logging purposes.
• query str - The GraphQL query to be executed.
• variables Dict[str, Any] - Variables required for the GraphQL query.

Returns:

• Dict - The JSON response from the GraphQL endpoint.

Raises:

• Exception - If the GraphQL call fails or returns errors.

​

make_rest_call

def make_rest_call(subpath: str, body: Dict[str, Any]) -> Dict

Executes a REST API call to the specified subpath with the given body.

Arguments:

• subpath str - The subpath of the REST API endpoint.
• body Dict[str, Any] - The JSON body to send with the POST request.

Returns:

• Dict - The JSON response from the REST API endpoint.

​

gql_helper

def gql_helper(query: str, description: str, variables: Dict,
               process_response: Callable[..., R]) -> R

Helper function to make a GraphQL call and process the response.

Arguments:

• query str - The GraphQL query to execute.
• description str - Description of the GraphQL operation for logging purposes.
• variables Dict - Variables required for the GraphQL query.
• process_response Callable[…, R] - A function to process the response.

Returns:

• R - The result of processing the response.

​

get_users

def get_users(first: Optional[int] = None,
              after: Optional[str] = None,
              before: Optional[str] = None,
              filters: Optional[users_filters] = None)

Retrieves a list of users based on pagination and optional filters.

Arguments:

• first Optional[int] - The number of users to retrieve.
• after Optional[str] - A cursor for use in pagination, fetching records after this cursor.
• before Optional[str] - A cursor for use in pagination, fetching records before this cursor.
• filters Optional[users_filters] - Filters to apply to the user query.

Returns:

• Dict - A dictionary containing the queried user data.

​

get_user

def get_user(id: Optional[str] = None, identifier: Optional[str] = None)

Retrieves a user based on the provided ID or identifier.

Arguments:

• id Optional[str] - The unique ID of the user.
• identifier Optional[str] - A unique identifier for the user, such as a username or email.

Returns:

The user data as returned by the GraphQL helper function.

​

create_user

def create_user(identifier: str, metadata: Optional[Dict] = None)

Creates a new user with the specified identifier and optional metadata.

Arguments:

• identifier str - A unique identifier for the user, such as a username or email.
• metadata Optional[Dict] - Additional data associated with the user.

Returns:

The result of the GraphQL call to create a user.

​

update_user

def update_user(id: str,
                identifier: Optional[str] = None,
                metadata: Optional[Dict] = None)

Updates an existing user identified by the given ID, with optional new identifier and metadata.

Arguments:

• id str - The unique ID of the user to update.
• identifier Optional[str] - A new identifier for the user, such as a username or email.
• metadata Optional[Dict] - New or updated metadata for the user.

Returns:

The result of the GraphQL call to update the user.

​

delete_user

def delete_user(id: str)

Deletes a user identified by the given ID.

Arguments:

• id str - The unique ID of the user to delete.

Returns:

The result of the GraphQL call to delete the user.

​

get_or_create_user

def get_or_create_user(identifier: str, metadata: Optional[Dict] = None)

Retrieves a user by their identifier, or creates a new user if they do not exist.

Arguments:

• identifier str - The identifier of the user to retrieve or create.
• metadata Optional[Dict] - Metadata to associate with the user if they are created.

Returns:

The existing or newly created user data.

​

get_threads

def get_threads(first: Optional[int] = None,
                after: Optional[str] = None,
                before: Optional[str] = None,
                filters: Optional[threads_filters] = None,
                order_by: Optional[threads_order_by] = None,
                step_types_to_keep: Optional[List[StepType]] = None)

Fetches a list of threads based on pagination and optional filters.

Arguments:

• first Optional[int] - Number of threads to fetch.
• after Optional[str] - Cursor for pagination, fetch threads after this cursor.
• before Optional[str] - Cursor for pagination, fetch threads before this cursor.
• filters Optional[threads_filters] - Filters to apply on the threads query.
• order_by Optional[threads_order_by] - Order by clause for threads.

Returns:

A list of threads that match the criteria.

​

list_threads

def list_threads(first: Optional[int] = None,
                 after: Optional[str] = None,
                 before: Optional[str] = None,
                 filters: Optional[threads_filters] = None,
                 order_by: Optional[threads_order_by] = None)

Lists threads based on pagination and optional filters, similar to get_threads but may include additional processing.

from literalai.thread import DateTimeFilter, ThreadFilter

filters = ThreadFilter(
    createdAt=DateTimeFilter(operator="gt", value="2023-12-14"),
)

page = 1
result = literal_client.api.list_threads(filters=filters)

Arguments:

• first Optional[int] - Number of threads to list.
• after Optional[str] - Cursor for pagination, list threads after this cursor.
• before Optional[str] - Cursor for pagination, list threads before this cursor.
• filters Optional[threads_filters] - Filters to apply on the threads listing.
• order_by Optional[threads_order_by] - Order by clause for threads.

Returns:

A list of threads that match the criteria.

​

get_thread

def get_thread(id: str)

Retrieves a single thread by its ID.

Arguments:

• id str - The unique identifier of the thread.

Returns:

The thread corresponding to the provided ID.

​

create_thread

def create_thread(name: Optional[str] = None,
                  metadata: Optional[Dict] = None,
                  participant_id: Optional[str] = None,
                  environment: Optional[str] = None,
                  tags: Optional[List[str]] = None)

Creates a new thread with the specified details.

Arguments:

• name Optional[str] - Name of the thread.
• metadata Optional[Dict] - Metadata associated with the thread.
• participant_id Optional[str] - Identifier for the participant.
• environment Optional[str] - Environment in which the thread operates.
• tags Optional[List[str]] - List of tags associated with the thread.

Returns:

The newly created thread.

​

upsert_thread

def upsert_thread(id: str,
                  name: Optional[str] = None,
                  metadata: Optional[Dict] = None,
                  participant_id: Optional[str] = None,
                  environment: Optional[str] = None,
                  tags: Optional[List[str]] = None)

Updates an existing thread or creates a new one if it does not exist.

Arguments:

• id str - The unique identifier of the thread.
• name Optional[str] - Name of the thread.
• metadata Optional[Dict] - Metadata associated with the thread.
• participant_id Optional[str] - Identifier for the participant.
• environment Optional[str] - Environment in which the thread operates.
• tags Optional[List[str]] - List of tags associated with the thread.

Returns:

The updated or newly created thread.

​

update_thread

def update_thread(id: str,
                  name: Optional[str] = None,
                  metadata: Optional[Dict] = None,
                  participant_id: Optional[str] = None,
                  environment: Optional[str] = None,
                  tags: Optional[List[str]] = None)

Updates the specified details of an existing thread.

Arguments:

• id str - The unique identifier of the thread to update.
• name Optional[str] - New name of the thread.
• metadata Optional[Dict] - New metadata for the thread.
• participant_id Optional[str] - New identifier for the participant.
• environment Optional[str] - New environment for the thread.
• tags Optional[List[str]] - New list of tags for the thread.

Returns:

The updated thread.

​

delete_thread

def delete_thread(id: str)

Deletes a thread identified by its ID.

Arguments:

• id str - The unique identifier of the thread to delete.

Returns:

The result of the deletion operation.

​

create_score

def create_score(name: str,
                 value: float,
                 type: ScoreType,
                 step_id: Optional[str] = None,
                 generation_id: Optional[str] = None,
                 dataset_experiment_item_id: Optional[str] = None,
                 comment: Optional[str] = None,
                 tags: Optional[List[str]] = None)

Creates a single score in the database.

Arguments:

• name str - The name of the score.
• value float - The numerical value of the score.
• type ScoreType - The type of the score.
• step_id Optional[str] - The ID of the step associated with the score.
• generation_id Optional[str] - The ID of the generation associated with the score.
• dataset_experiment_item_id Optional[str] - The ID of the dataset experiment item associated with the score.
• comment Optional[str] - An optional comment about the score.
• tags Optional[List[str]] - Optional tags associated with the score.

Returns:

The created Score object.

​

update_score

def update_score(id: str, update_params: ScoreUpdate)

Updates a score identified by its ID with new parameters.

Arguments:

• id str - The unique identifier of the score to update.
• update_params ScoreUpdate - A dictionary of parameters to update in the score.

Returns:

The result of the update operation.

​

delete_score

def delete_score(id: str)

Deletes a score identified by its ID.

Arguments:

• id str - The unique identifier of the score to delete.

Returns:

The result of the deletion operation.

​

upload_file

def upload_file(content: Union[bytes, str],
                thread_id: Optional[str] = None,
                mime: Optional[str] = "application/octet-stream") -> Dict

Uploads a file to the server.

Arguments:

• content Union[bytes, str] - The content of the file to upload.
• thread_id Optional[str] - The ID of the thread associated with the file.
• mime Optional[str] - The MIME type of the file. Defaults to ‘application/octet-stream’.

Returns:

• Dict - A dictionary containing the object key and URL of the uploaded file, or None values if the upload fails.

​

create_attachment

def create_attachment(thread_id: str,
                      step_id: str,
                      id: Optional[str] = None,
                      metadata: Optional[Dict] = None,
                      mime: Optional[str] = None,
                      name: Optional[str] = None,
                      object_key: Optional[str] = None,
                      url: Optional[str] = None,
                      content: Optional[Union[bytes, str]] = None,
                      path: Optional[str] = None) -> "Attachment"

Creates an attachment associated with a thread and step, potentially uploading file content.

Arguments:

• thread_id str - The ID of the thread to which the attachment is linked.
• step_id str - The ID of the step to which the attachment is linked.
• id Optional[str] - The ID of the attachment, if updating an existing one.
• metadata Optional[Dict] - Metadata associated with the attachment.
• mime Optional[str] - MIME type of the file, if content is provided.
• name Optional[str] - Name of the attachment.
• object_key Optional[str] - Object key of the uploaded file, if already known.
• url Optional[str] - URL of the uploaded file, if already known.
• content Optional[Union[bytes, str]] - File content to upload.
• path Optional[str] - Path where the file should be stored.

Returns:

• Attachment - The created or updated attachment object.

​

update_attachment

def update_attachment(id: str, update_params: AttachmentUpload)

Updates an existing attachment with new parameters.

Arguments:

• id str - The unique identifier of the attachment to update.
• update_params AttachmentUpload - The parameters to update in the attachment.

Returns:

The result of the update operation.

​

get_attachment

def get_attachment(id: str)

Retrieves an attachment by its ID.

Arguments:

• id str - The unique identifier of the attachment to retrieve.

Returns:

The attachment data as returned by the GraphQL helper function.

​

delete_attachment

def delete_attachment(id: str)

Deletes an attachment identified by its ID.

Arguments:

• id str - The unique identifier of the attachment to delete.

Returns:

The result of the deletion operation.

​

create_step

def create_step(thread_id: Optional[str] = None,
                type: Optional[StepType] = "undefined",
                start_time: Optional[str] = None,
                end_time: Optional[str] = None,
                input: Optional[Dict] = None,
                output: Optional[Dict] = None,
                metadata: Optional[Dict] = None,
                parent_id: Optional[str] = None,
                name: Optional[str] = None,
                tags: Optional[List[str]] = None)

Creates a new step with the specified parameters.

Arguments:

• thread_id Optional[str] - The ID of the thread this step is associated with.
• type Optional[StepType] - The type of the step, defaults to “undefined”.
• start_time Optional[str] - The start time of the step.
• end_time Optional[str] - The end time of the step.
• input Optional[Dict] - Input data for the step.
• output Optional[Dict] - Output data from the step.
• metadata Optional[Dict] - Metadata associated with the step.
• parent_id Optional[str] - The ID of the parent step, if any.
• name Optional[str] - The name of the step.
• tags Optional[List[str]] - Tags associated with the step.

Returns:

The result of the GraphQL helper function for creating a step.

​

update_step

def update_step(id: str,
                type: Optional[StepType] = None,
                input: Optional[str] = None,
                output: Optional[str] = None,
                metadata: Optional[Dict] = None,
                name: Optional[str] = None,
                tags: Optional[List[str]] = None,
                start_time: Optional[str] = None,
                end_time: Optional[str] = None,
                parent_id: Optional[str] = None)

Updates an existing step identified by its ID with new parameters.

Arguments:

• id str - The unique identifier of the step to update.
• type Optional[StepType] - The type of the step.
• input Optional[str] - Input data for the step.
• output Optional[str] - Output data from the step.
• metadata Optional[Dict] - Metadata associated with the step.
• name Optional[str] - The name of the step.
• tags Optional[List[str]] - Tags associated with the step.
• start_time Optional[str] - The start time of the step.
• end_time Optional[str] - The end time of the step.
• parent_id Optional[str] - The ID of the parent step, if any.

Returns:

The result of the GraphQL helper function for updating a step.

​

get_steps

def get_steps(
        first: Optional[int] = None,
        after: Optional[str] = None,
        before: Optional[str] = None,
        filters: Optional[steps_filters] = None,
        order_by: Optional[steps_order_by] = None) -> PaginatedResponse[Step]

Fetches a list of steps based on pagination and optional filters.

Arguments:

• first Optional[int] - Number of steps to fetch.
• after Optional[str] - Cursor for pagination, fetch steps after this cursor.
• before Optional[str] - Cursor for pagination, fetch steps before this cursor.
• filters Optional[steps_filters] - Filters to apply on the steps query.
• order_by Optional[steps_order_by] - Order by clause for steps.

Returns:

A list of steps that match the criteria.

​

get_step

def get_step(id: str)

Retrieves a step by its ID.

Arguments:

• id str - The unique identifier of the step to retrieve.

Returns:

The step data as returned by the GraphQL helper function.

​

delete_step

def delete_step(id: str)

Deletes a step identified by its ID.

Arguments:

• id str - The unique identifier of the step to delete.

Returns:

The result of the deletion operation.

​

send_steps

def send_steps(steps: List[Union[StepDict, "Step"]])

Sends a list of steps to be processed.

Arguments:

• steps List[Union[StepDict, “Step”]] - A list of steps or step dictionaries to send.

Returns:

The result of the GraphQL helper function for sending steps.

​

get_generations

def get_generations(first: Optional[int] = None,
                    after: Optional[str] = None,
                    before: Optional[str] = None,
                    filters: Optional[generations_filters] = None,
                    order_by: Optional[generations_order_by] = None)

Fetches a list of generations based on pagination and optional filters.

Arguments:

• first Optional[int] - The number of generations to retrieve.
• after Optional[str] - A cursor for use in pagination, fetching records after this cursor.
• before Optional[str] - A cursor for use in pagination, fetching records before this cursor.
• filters Optional[generations_filters] - Filters to apply to the generations query.
• order_by Optional[generations_order_by] - Order by clause for generations.

Returns:

A list of generations that match the criteria.

​

create_generation

def create_generation(generation: Union[ChatGeneration, CompletionGeneration])

Creates a new generation, either a chat or completion type.

Arguments:

• generation Union[ChatGeneration, CompletionGeneration] - The generation data to create.

Returns:

The result of the creation operation.

​

create_dataset

def create_dataset(name: str,
                   description: Optional[str] = None,
                   metadata: Optional[Dict] = None,
                   type: DatasetType = "key_value")

Creates a new dataset with the specified properties.

Arguments:

• name str - The name of the dataset.
• description Optional[str] - A description of the dataset.
• metadata Optional[Dict] - Additional metadata for the dataset.
• type DatasetType - The type of the dataset, defaults to “key_value”.

Returns:

The result of the dataset creation operation.

​

get_dataset

def get_dataset(id: Optional[str] = None, name: Optional[str] = None)

Retrieves a dataset by its ID or name.

Arguments:

• id Optional[str] - The unique identifier of the dataset.
• name Optional[str] - The name of the dataset.

Returns:

The dataset data as returned by the REST helper function.

​

update_dataset

def update_dataset(id: str,
                   name: Optional[str] = None,
                   description: Optional[str] = None,
                   metadata: Optional[Dict] = None)

Updates an existing dataset identified by its ID with new properties.

Arguments:

• id str - The unique identifier of the dataset to update.
• name Optional[str] - A new name for the dataset.
• description Optional[str] - A new description for the dataset.
• metadata Optional[Dict] - New or updated metadata for the dataset.

Returns:

The result of the dataset update operation.

​

delete_dataset

def delete_dataset(id: str)

Deletes a dataset identified by its ID.

Arguments:

• id str - The unique identifier of the dataset to delete.

Returns:

The result of the deletion operation.

​

create_experiment

def create_experiment(dataset_id: str,
                      name: str,
                      prompt_id: Optional[str] = None,
                      params: Optional[Dict] = None) -> "DatasetExperiment"

Creates a new experiment associated with a specific dataset.

Arguments:

• dataset_id str - The unique identifier of the dataset.
• name str - The name of the experiment.
• prompt_id Optional[str] - The identifier of the prompt associated with the experiment.
• params Optional[Dict] - Additional parameters for the experiment.

Returns:

• DatasetExperiment - The newly created experiment object.

​

create_experiment_item

def create_experiment_item(
        experiment_item: DatasetExperimentItem) -> DatasetExperimentItem

Creates an experiment item within an existing experiment.

Arguments:

• experiment_item DatasetExperimentItem - The experiment item to be created, containing all necessary data.

Returns:

• DatasetExperimentItem - The newly created experiment item with scores attached.

​

create_dataset_item

def create_dataset_item(dataset_id: str,
                        input: Dict,
                        expected_output: Optional[Dict] = None,
                        metadata: Optional[Dict] = None)

Creates a new dataset item with the specified properties.

Arguments:

• dataset_id str - The unique identifier of the dataset.
• input Dict - The input data for the dataset item.
• expected_output Optional[Dict] - The expected output data for the dataset item.
• metadata Optional[Dict] - Additional metadata for the dataset item.

Returns:

• Dict - The result of the dataset item creation operation.

​

get_dataset_item

def get_dataset_item(id: str)

Retrieves a dataset item by its unique identifier.

Arguments:

• id str - The unique identifier of the dataset item to retrieve.

Returns:

• Dict - The dataset item data.

​

delete_dataset_item

def delete_dataset_item(id: str)

Deletes a dataset item by its unique identifier.

Arguments:

• id str - The unique identifier of the dataset item to delete.

Returns:

• Dict - The result of the dataset item deletion operation.

​

add_step_to_dataset

def add_step_to_dataset(dataset_id: str,
                        step_id: str,
                        metadata: Optional[Dict] = None)

Adds a step to a dataset.

Arguments:

• dataset_id str - The unique identifier of the dataset.
• step_id str - The unique identifier of the step to add.
• metadata Optional[Dict] - Additional metadata for the step being added.

Returns:

• Dict - The result of adding the step to the dataset.

​

add_generation_to_dataset

def add_generation_to_dataset(dataset_id: str,
                              generation_id: str,
                              metadata: Optional[Dict] = None)

Adds a generation to a dataset.

Arguments:

• dataset_id str - The unique identifier of the dataset.
• generation_id str - The unique identifier of the generation to add.
• metadata Optional[Dict] - Additional metadata for the generation being added.

Returns:

• Dict - The result of adding the generation to the dataset.

​

create_prompt_lineage

def create_prompt_lineage(name: str, description: Optional[str] = None)

Creates a prompt lineage with the specified name and optional description.

Arguments:

• name str - The name of the prompt lineage.
• description Optional[str] - An optional description of the prompt lineage.

Returns:

• Dict - The result of the prompt lineage creation operation.

​

get_or_create_prompt

def get_or_create_prompt(name: str,
                         template_messages: List[GenerationMessage],
                         settings: Optional[ProviderSettings] = None,
                         tools: Optional[List[Dict]] = None) -> Prompt

A Prompt is fully defined by its name, template_messages, settings and tools. If a prompt already exists for the given arguments, it is returned. Otherwise, a new prompt is created.

Arguments:

• name str - The name of the prompt to retrieve or create.
• template_messages List[GenerationMessage] - A list of template messages for the prompt.
• settings Optional[Dict] - Optional settings for the prompt.

Returns:

• Prompt - The prompt that was retrieved or created.

​

get_prompt

def get_prompt(id: Optional[str] = None,
               name: Optional[str] = None,
               version: Optional[int] = 0) -> Prompt

Gets a prompt either by:

• id
• or name and (optional) version

Either the id or the name must be provided. If both are provided, the id is used.

Arguments:

• id str - The unique identifier of the prompt to retrieve.
• name str - The name of the prompt to retrieve.
• version Optional[int] - The version number of the prompt to retrieve.

Returns:

• Prompt - The prompt with the given identifier or name.

​

get_my_project_id

def get_my_project_id()

Retrieves the projectId associated to the API key.

Returns:

The projectId associated to the API key.

​

AsyncLiteralAPI Objects

class AsyncLiteralAPI(BaseLiteralAPI)

​

make_gql_call

async def make_gql_call(description: str, query: str,
                        variables: Dict[str, Any]) -> Dict

Asynchronously makes a GraphQL call using the provided query and variables.

Arguments:

• description str - Description of the GraphQL operation for logging purposes.
• query str - The GraphQL query to be executed.
• variables Dict[str, Any] - Variables required for the GraphQL query.

Returns:

• Dict - The JSON response from the GraphQL endpoint.

Raises:

• Exception - If the GraphQL call fails or returns errors.

​

make_rest_call

async def make_rest_call(subpath: str, body: Dict[str, Any]) -> Dict

Asynchronously makes a REST API call to a specified subpath with the provided body.

Arguments:

• subpath str - The endpoint subpath to which the POST request is made.
• body Dict[str, Any] - The JSON body of the POST request.

Returns:

• Dict - The JSON response from the REST API endpoint.

​

gql_helper

async def gql_helper(query: str, description: str, variables: Dict,
                     process_response: Callable[..., R]) -> R

Helper function to process a GraphQL query by making an asynchronous call and processing the response.

Arguments:

• query str - The GraphQL query to be executed.
• description str - Description of the GraphQL operation for logging purposes.
• variables Dict - Variables required for the GraphQL query.
• process_response Callable[…, R] - The function to process the response.

Returns:

• R - The result of processing the response.

​

get_users

async def get_users(first: Optional[int] = None,
                    after: Optional[str] = None,
                    before: Optional[str] = None,
                    filters: Optional[users_filters] = None)

Asynchronously fetches a list of users based on pagination and optional filters.

Arguments:

• first Optional[int] - The number of users to retrieve.
• after Optional[str] - A cursor for use in pagination, fetching records after this cursor.
• before Optional[str] - A cursor for use in pagination, fetching records before this cursor.
• filters Optional[users_filters] - Filters to apply to the user query.

Returns:

The result of the GraphQL helper function for fetching users.

​

get_user

async def get_user(id: Optional[str] = None, identifier: Optional[str] = None)

Asynchronously retrieves a user by ID or identifier.

Arguments:

• id Optional[str] - The unique identifier of the user to retrieve.
• identifier Optional[str] - An alternative identifier for the user.

Returns:

The result of the GraphQL helper function for fetching a user.

​

create_user

async def create_user(identifier: str, metadata: Optional[Dict] = None)

Asynchronously creates a new user with the specified identifier and optional metadata.

Arguments:

• identifier str - The identifier for the new user.
• metadata Optional[Dict] - Additional metadata for the user.

Returns:

The result of the GraphQL helper function for creating a user.

​

update_user

async def update_user(id: str,
                      identifier: Optional[str] = None,
                      metadata: Optional[Dict] = None)

Asynchronously updates an existing user identified by ID with new identifier and/or metadata.

Arguments:

• id str - The unique identifier of the user to update.
• identifier Optional[str] - New identifier for the user.
• metadata Optional[Dict] - New metadata for the user.

Returns:

The result of the GraphQL helper function for updating a user.

​

delete_user

async def delete_user(id: str)

Asynchronously deletes a user identified by ID.

Arguments:

• id str - The unique identifier of the user to delete.

Returns:

The result of the GraphQL helper function for deleting a user.

​

get_or_create_user

async def get_or_create_user(identifier: str, metadata: Optional[Dict] = None)

Asynchronously retrieves a user by identifier or creates a new one if it does not exist.

Arguments:

• identifier str - The identifier of the user to retrieve or create.
• metadata Optional[Dict] - Metadata for the user if creation is necessary.

Returns:

The existing or newly created user.

​

get_threads

async def get_threads(first: Optional[int] = None,
                      after: Optional[str] = None,
                      before: Optional[str] = None,
                      filters: Optional[threads_filters] = None,
                      order_by: Optional[threads_order_by] = None,
                      step_types_to_keep: Optional[List[StepType]] = None)

Asynchronously fetches a list of threads based on pagination and optional filters and ordering.

Arguments:

• first Optional[int] - The number of threads to retrieve.
• after Optional[str] - A cursor for use in pagination, fetching records after this cursor.
• before Optional[str] - A cursor for use in pagination, fetching records before this cursor.
• filters Optional[threads_filters] - Filters to apply to the thread query.
• order_by Optional[threads_order_by] - Ordering criteria for the threads.

Returns:

The result of the GraphQL helper function for fetching threads.

​

list_threads

async def list_threads(first: Optional[int] = None,
                       after: Optional[str] = None,
                       before: Optional[str] = None,
                       filters: Optional[threads_filters] = None,
                       order_by: Optional[threads_order_by] = None)

Asynchronously lists threads based on pagination and optional filters and ordering, similar to get_threads.

Arguments:

• first Optional[int] - The number of threads to list.
• after Optional[str] - A cursor for use in pagination, fetching records after this cursor.
• before Optional[str] - A cursor for use in pagination, fetching records before this cursor.
• filters Optional[threads_filters] - Filters to apply to the thread query.
• order_by Optional[threads_order_by] - Ordering criteria for the threads.

Returns:

The result of the GraphQL helper function for listing threads.

​

get_thread

async def get_thread(id: str)

Asynchronously retrieves a thread by its ID.

Arguments:

• id str - The unique identifier of the thread to retrieve.

Returns:

The result of the GraphQL helper function for fetching a thread.

​

create_thread

async def create_thread(name: Optional[str] = None,
                        metadata: Optional[Dict] = None,
                        participant_id: Optional[str] = None,
                        environment: Optional[str] = None,
                        tags: Optional[List[str]] = None)

Asynchronously creates a new thread with specified details.

Arguments:

• name Optional[str] - The name of the thread.
• metadata Optional[Dict] - Metadata associated with the thread.
• participant_id Optional[str] - Identifier for the participant associated with the thread.
• environment Optional[str] - The environment in which the thread operates.
• tags Optional[List[str]] - Tags associated with the thread.

Returns:

The result of the GraphQL helper function for creating a thread.

​

upsert_thread

async def upsert_thread(id: str,
                        name: Optional[str] = None,
                        metadata: Optional[Dict] = None,
                        participant_id: Optional[str] = None,
                        environment: Optional[str] = None,
                        tags: Optional[List[str]] = None)

Asynchronously updates or inserts a thread based on the provided ID.

Arguments:

• id str - The unique identifier of the thread to upsert.
• name Optional[str] - The name of the thread.
• metadata Optional[Dict] - Metadata associated with the thread.
• participant_id Optional[str] - Identifier for the participant associated with the thread.
• environment Optional[str] - The environment in which the thread operates.
• tags Optional[List[str]] - Tags associated with the thread.

Returns:

The result of the GraphQL helper function for upserting a thread.

​

update_thread

async def update_thread(id: str,
                        name: Optional[str] = None,
                        metadata: Optional[Dict] = None,
                        participant_id: Optional[str] = None,
                        environment: Optional[str] = None,
                        tags: Optional[List[str]] = None)

Asynchronously updates an existing thread identified by ID with new details.

Arguments:

• id str - The unique identifier of the thread to update.
• name Optional[str] - New name of the thread.
• metadata Optional[Dict] - New metadata for the thread.
• participant_id Optional[str] - New identifier for the participant.
• environment Optional[str] - New environment for the thread.
• tags Optional[List[str]] - New list of tags for the thread.

Returns:

The result of the GraphQL helper function for updating a thread.

​

delete_thread

async def delete_thread(id: str)

Asynchronously deletes a thread identified by its ID.

Arguments:

• id str - The unique identifier of the thread to delete.

Returns:

The result of the GraphQL helper function for deleting a thread.

​

get_scores

async def get_scores(first: Optional[int] = None,
                     after: Optional[str] = None,
                     before: Optional[str] = None,
                     filters: Optional[scores_filters] = None,
                     order_by: Optional[scores_order_by] = None)

Asynchronously fetches scores based on pagination and optional filters.

Arguments:

• first Optional[int] - The number of scores to retrieve.
• after Optional[str] - A cursor for use in pagination, fetching records after this cursor.
• before Optional[str] - A cursor for use in pagination, fetching records before this cursor.
• filters Optional[scores_filters] - Filters to apply to the scores query.
• order_by Optional[scores_order_by] - Ordering options for the scores.

Returns:

The result of the GraphQL helper function for fetching scores.

​

create_scores

async def create_scores(scores: List[ScoreDict])

Asynchronously creates multiple scores.

Arguments:

• scores List[ScoreDict] - A list of dictionaries representing the scores to be created.

Returns:

The result of the GraphQL helper function for creating scores.

​

create_score

async def create_score(name: str,
                       value: float,
                       type: ScoreType,
                       step_id: Optional[str] = None,
                       generation_id: Optional[str] = None,
                       dataset_experiment_item_id: Optional[str] = None,
                       comment: Optional[str] = None,
                       tags: Optional[List[str]] = None)

Asynchronously creates a single score.

Arguments:

• name str - The name of the score.
• value float - The numerical value of the score.
• type ScoreType - The type of the score.
• step_id Optional[str] - The ID of the step associated with the score.
• generation_id Optional[str] - The ID of the generation associated with the score.
• dataset_experiment_item_id Optional[str] - The ID of the dataset experiment item associated with the score.
• comment Optional[str] - A comment associated with the score.
• tags Optional[List[str]] - A list of tags associated with the score.

Returns:

The result of the GraphQL helper function for creating a score.

​

update_score

async def update_score(id: str, update_params: ScoreUpdate)

Asynchronously updates a score identified by its ID.

Arguments:

• id str - The unique identifier of the score to update.
• update_params ScoreUpdate - A dictionary of parameters to update.

Returns:

The result of the GraphQL helper function for updating a score.

​

delete_score

async def delete_score(id: str)

Asynchronously deletes a score identified by its ID.

Arguments:

• id str - The unique identifier of the score to delete.

Returns:

The result of the GraphQL helper function for deleting a score.

​

upload_file

async def upload_file(
        content: Union[bytes, str],
        thread_id: str,
        mime: Optional[str] = "application/octet-stream") -> Dict

Asynchronously uploads a file to the server.

Arguments:

• content Union[bytes, str] - The content of the file to upload.
• thread_id str - The ID of the thread associated with the file.
• mime Optional[str] - The MIME type of the file.

Returns:

A dictionary containing the object key and URL of the uploaded file.

​

create_attachment

async def create_attachment(thread_id: str,
                            step_id: str,
                            id: Optional[str] = None,
                            metadata: Optional[Dict] = None,
                            mime: Optional[str] = None,
                            name: Optional[str] = None,
                            object_key: Optional[str] = None,
                            url: Optional[str] = None,
                            content: Optional[Union[bytes, str]] = None,
                            path: Optional[str] = None) -> "Attachment"

Asynchronously creates an attachment and uploads it if content is provided.

Arguments:

• thread_id str - The ID of the thread associated with the attachment.
• step_id str - The ID of the step associated with the attachment.
• id Optional[str] - An optional unique identifier for the attachment.
• metadata Optional[Dict] - Optional metadata for the attachment.
• mime Optional[str] - The MIME type of the attachment.
• name Optional[str] - The name of the attachment.
• object_key Optional[str] - The object key for the attachment if already uploaded.
• url Optional[str] - The URL of the attachment if already uploaded.
• content Optional[Union[bytes, str]] - The content of the attachment to upload.
• path Optional[str] - The file path of the attachment if it is to be uploaded from a local file.

Returns:

The attachment object created after the upload and creation process.

​

update_attachment

async def update_attachment(id: str, update_params: AttachmentUpload)

Asynchronously updates an attachment identified by its ID.

Arguments:

• id str - The unique identifier of the attachment to update.
• update_params AttachmentUpload - A dictionary of parameters to update the attachment.

Returns:

The result of the GraphQL helper function for updating an attachment.

​

get_attachment

async def get_attachment(id: str)

Asynchronously retrieves an attachment by its ID.

Arguments:

• id str - The unique identifier of the attachment to retrieve.

Returns:

The result of the GraphQL helper function for fetching an attachment.

​

delete_attachment

async def delete_attachment(id: str)

Asynchronously deletes an attachment identified by its ID.

Arguments:

• id str - The unique identifier of the attachment to delete.

Returns:

The result of the GraphQL helper function for deleting an attachment.

​

create_step

async def create_step(thread_id: Optional[str] = None,
                      type: Optional[StepType] = "undefined",
                      start_time: Optional[str] = None,
                      end_time: Optional[str] = None,
                      input: Optional[Dict] = None,
                      output: Optional[Dict] = None,
                      metadata: Optional[Dict] = None,
                      parent_id: Optional[str] = None,
                      name: Optional[str] = None,
                      tags: Optional[List[str]] = None)

Asynchronously creates a new step with the specified parameters.

Arguments:

• thread_id Optional[str] - The ID of the thread associated with the step.
• type Optional[StepType] - The type of the step, defaults to “undefined”.
• start_time Optional[str] - The start time of the step.
• end_time Optional[str] - The end time of the step.
• input Optional[Dict] - Input data for the step.
• output Optional[Dict] - Output data from the step.
• metadata Optional[Dict] - Metadata associated with the step.
• parent_id Optional[str] - The ID of the parent step, if any.
• name Optional[str] - The name of the step.
• tags Optional[List[str]] - Tags associated with the step.

Returns:

The result of the GraphQL helper function for creating a step.

​

update_step

async def update_step(id: str,
                      type: Optional[StepType] = None,
                      input: Optional[str] = None,
                      output: Optional[str] = None,
                      metadata: Optional[Dict] = None,
                      name: Optional[str] = None,
                      tags: Optional[List[str]] = None,
                      start_time: Optional[str] = None,
                      end_time: Optional[str] = None,
                      parent_id: Optional[str] = None)

Asynchronously updates an existing step identified by its ID with new parameters.

Arguments:

• id str - The unique identifier of the step to update.
• type Optional[StepType] - The type of the step.
• input Optional[str] - Input data for the step.
• output Optional[str] - Output data from the step.
• metadata Optional[Dict] - Metadata associated with the step.
• name Optional[str] - The name of the step.
• tags Optional[List[str]] - Tags associated with the step.
• start_time Optional[str] - The start time of the step.
• end_time Optional[str] - The end time of the step.
• parent_id Optional[str] - The ID of the parent step, if any.

Returns:

The result of the GraphQL helper function for updating a step.

​

get_step

async def get_step(id: str)

Asynchronously retrieves a step by its ID.

Arguments:

• id str - The unique identifier of the step to retrieve.

Returns:

The result of the GraphQL helper function for fetching a step.

​

delete_step

async def delete_step(id: str)

Asynchronously deletes a step identified by its ID.

Arguments:

• id str - The unique identifier of the step to delete.

Returns:

The result of the GraphQL helper function for deleting a step.

​

send_steps

async def send_steps(steps: List[Union[StepDict, "Step"]])

Asynchronously sends a list of steps to be processed.

Arguments:

• steps List[Union[StepDict, “Step”]] - A list of steps or step dictionaries to send.

Returns:

The result of the GraphQL helper function for sending steps.

​

get_generations

async def get_generations(first: Optional[int] = None,
                          after: Optional[str] = None,
                          before: Optional[str] = None,
                          filters: Optional[generations_filters] = None,
                          order_by: Optional[generations_order_by] = None)

Asynchronously fetches a list of generations based on pagination and optional filters.

Arguments:

• first Optional[int] - The number of generations to retrieve.
• after Optional[str] - A cursor for use in pagination, fetching records after this cursor.
• before Optional[str] - A cursor for use in pagination, fetching records before this cursor.
• filters Optional[generations_filters] - Filters to apply to the generations query.
• order_by Optional[generations_order_by] - Ordering options for the generations.

Returns:

The result of the GraphQL helper function for fetching generations.

​

create_generation

async def create_generation(generation: Union[ChatGeneration,
                                              CompletionGeneration])

Asynchronously creates a new generation with the specified details.

Arguments:

• generation Union[ChatGeneration, CompletionGeneration] - The generation data to create.

Returns:

The result of the GraphQL helper function for creating a generation.

​

create_dataset

async def create_dataset(name: str,
                         description: Optional[str] = None,
                         metadata: Optional[Dict] = None,
                         type: DatasetType = "key_value")

Asynchronously creates a new dataset with the specified details.

Arguments:

• name str - The name of the dataset.
• description Optional[str] - A description of the dataset.
• metadata Optional[Dict] - Metadata associated with the dataset.
• type DatasetType - The type of the dataset, defaults to “key_value”.

Returns:

The result of the GraphQL helper function for creating a dataset.

​

get_dataset

async def get_dataset(id: Optional[str] = None, name: Optional[str] = None)

Asynchronously retrieves a dataset by its ID or name.

Arguments:

• id Optional[str] - The unique identifier of the dataset to retrieve.
• name Optional[str] - The name of the dataset to retrieve.

Returns:

The processed response from the REST API call.

​

update_dataset

async def update_dataset(id: str,
                         name: Optional[str] = None,
                         description: Optional[str] = None,
                         metadata: Optional[Dict] = None)

Asynchronously updates an existing dataset identified by its ID with new details.

Arguments:

• id str - The unique identifier of the dataset to update.
• name Optional[str] - The new name of the dataset.
• description Optional[str] - A new description for the dataset.
• metadata Optional[Dict] - New metadata for the dataset.

Returns:

The result of the GraphQL helper function for updating a dataset.

​

delete_dataset

async def delete_dataset(id: str)

Asynchronously deletes a dataset identified by its ID.

Arguments:

• id str - The unique identifier of the dataset to delete.

Returns:

The result of the GraphQL helper function for deleting a dataset.

​

create_experiment_item

async def create_experiment_item(
        experiment_item: DatasetExperimentItem) -> DatasetExperimentItem

Asynchronously creates an item within an experiment.

Arguments:

• experiment_item DatasetExperimentItem - The experiment item to be created.

Returns:

• DatasetExperimentItem - The created experiment item with updated scores.

​

create_dataset_item

async def create_dataset_item(dataset_id: str,
                              input: Dict,
                              expected_output: Optional[Dict] = None,
                              metadata: Optional[Dict] = None)

Asynchronously creates a dataset item.

Arguments:

• dataset_id str - The unique identifier of the dataset.
• input Dict - The input data for the dataset item.
• expected_output Optional[Dict] - The expected output data for the dataset item.
• metadata Optional[Dict] - Additional metadata for the dataset item.

Returns:

The result of the GraphQL helper function for creating a dataset item.

​

get_dataset_item

async def get_dataset_item(id: str)

Asynchronously retrieves a dataset item by its ID.

Arguments:

• id str - The unique identifier of the dataset item.

Returns:

The result of the GraphQL helper function for fetching a dataset item.

​

delete_dataset_item

async def delete_dataset_item(id: str)

Asynchronously deletes a dataset item by its ID.

Arguments:

• id str - The unique identifier of the dataset item to delete.

Returns:

The result of the GraphQL helper function for deleting a dataset item.

​

add_step_to_dataset

async def add_step_to_dataset(dataset_id: str,
                              step_id: str,
                              metadata: Optional[Dict] = None)

Asynchronously adds a step to a dataset.

Arguments:

• dataset_id str - The unique identifier of the dataset.
• step_id str - The unique identifier of the step to add.
• metadata Optional[Dict] - Additional metadata for the step being added.

Returns:

The result of the GraphQL helper function for adding a step to a dataset.

​

add_generation_to_dataset

async def add_generation_to_dataset(dataset_id: str,
                                    generation_id: str,
                                    metadata: Optional[Dict] = None)

Asynchronously adds a generation to a dataset.

Arguments:

• dataset_id str - The unique identifier of the dataset.
• generation_id str - The unique identifier of the generation to add.
• metadata Optional[Dict] - Additional metadata for the generation being added.

Returns:

The result of the GraphQL helper function for adding a generation to a dataset.

​

create_prompt_lineage

async def create_prompt_lineage(name: str, description: Optional[str] = None)

Asynchronously creates a prompt lineage.

Arguments:

• name str - The name of the prompt lineage.
• description Optional[str] - An optional description of the prompt lineage.

Returns:

The result of the GraphQL helper function for creating a prompt lineage.