Api - Literal AI Documentation

LiteralAPI

class LiteralAPI(BaseLiteralAPI)

from literalai import LiteralClient
# Initialize the client
literalai_client = LiteralClient(api_key="your_api_key_here")
# Access the API's methods
print(literalai_client.api)

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:

Optional[int]

The number of users to retrieve.

Optional[str]

A cursor for use in pagination, fetching records after this cursor.

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:

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:

Optional[str]

The unique ID of the user.

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:

str

A unique identifier for the user, such as a username or email.

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:

str

The unique ID of the user to update.

Optional[str]

A new identifier for the user, such as a username or email.

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:

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:

str

The identifier of the user to retrieve or create.

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:

Optional[int]

Number of threads to fetch.

Optional[str]

Cursor for pagination, fetch threads after this cursor.

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. step_types_to_keep (Optional[List[StepType]]) : If set, only steps of the corresponding types will be returned

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. Arguments:

Optional[int]

Number of threads to list.

Optional[str]

Cursor for pagination, list threads after this cursor.

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:

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,
                  tags: Optional[List[str]] = None)

Creates a new thread with the specified details. Arguments:

Optional[str]

Name of the thread.

Optional[Dict]

Metadata associated with the thread.

Optional[str]

Identifier for the participant.

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,
                  tags: Optional[List[str]] = None)

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

str

The unique identifier of the thread.

Optional[str]

Name of the thread.

Optional[Dict]

Metadata associated with the thread.

Optional[str]

Identifier for the participant.

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,
                  tags: Optional[List[str]] = None)

Updates the specified details of an existing thread. Arguments:

str

The unique identifier of the thread to update.

Optional[str]

New name of the thread.

Optional[Dict]

New metadata for the thread.

Optional[str]

New identifier for the participant.

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:

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:

str

The name of the score.

float

The numerical value of the score.

ScoreType

The type of the score.

Optional[str]

The ID of the step associated with the score.

Optional[str]

The ID of the generation associated with the score.

Optional[str]

The ID of the dataset experiment item associated with the score.

Optional[str]

An optional comment about the score.

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:

str

The unique identifier of the score to update.

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:

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:

Union[bytes, str]

The content of the file to upload.

Optional[str]

The ID of the thread associated with the file.

Optional[str]

The MIME type of the file. Defaults to ‘application/octet-stream’.

Returns:

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: Optional[str] = None,
                      step_id: Optional[str] = None,
                      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:

str

The ID of the thread to which the attachment is linked.

str

The ID of the step to which the attachment is linked.

Optional[str]

The ID of the attachment, if updating an existing one.

Optional[Dict]

Metadata associated with the attachment.

Optional[str]

MIME type of the file, if content is provided.

Optional[str]

Name of the attachment.

Optional[str]

Object key of the uploaded file, if already known.

Optional[str]

URL of the uploaded file, if already known.

Optional[Union[bytes, str]]

File content to upload.

Optional[str]

Path where the file should be stored.

Returns:

The created or updated attachment object.

update_attachment

def update_attachment(id: str, update_params: AttachmentUpload)

Updates an existing attachment with new parameters. Arguments:

str

The unique identifier of the attachment to update.

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:

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:

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,
                root_run_id: Optional[str] = None)

Creates a new step with the specified parameters. Arguments:

Optional[str]

The ID of the thread this step is associated with.

Optional[StepType]

The type of the step, defaults to “undefined”.

Optional[str]

The start time of the step.

Optional[str]

The end time of the step.

Optional[Dict]

Input data for the step.

Optional[Dict]

Output data from the step.

Optional[Dict]

Metadata associated with the step.

Optional[str]

The ID of the parent step, if any.

Optional[str]

The name of the step.

Optional[List[str]]

Tags associated with the step.

Optional[str]

The ID of the root run, if any.

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:

str

The unique identifier of the step to update.

Optional[StepType]

The type of the step.

Optional[str]

Input data for the step.

Optional[str]

Output data from the step.

Optional[Dict]

Metadata associated with the step.

Optional[str]

The name of the step.

Optional[List[str]]

Tags associated with the step.

Optional[str]

The start time of the step.

Optional[str]

The end time of the step.

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:

Optional[int]

Number of steps to fetch.

Optional[str]

Cursor for pagination, fetch steps after this cursor.

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:

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:

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:

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:

Optional[int]

The number of generations to retrieve.

Optional[str]

A cursor for use in pagination, fetching records after this cursor.

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.

from literalai.observability.generation import ChatGeneration
from literalai import LiteralClient

literalai_client = LiteralClient(api_key=)

example_generation = ChatGeneration(
    messages=[
        {
            "role": "user",
            "content": "Hello, how can I help you today?"
        },
    ],
    message_completion={
        "role": "assistant",
        "content": "Sure, I can help with that. What do you need to know?"
    },
    model="gpt-4o-mini",
    provider="OpenAI"
)

literalai_client.api.create_generation(example_generation)

Arguments:

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:

str

The name of the dataset.

Optional[str]

A description of the dataset.

Optional[Dict]

Additional metadata for the dataset.

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) -> Optional[Dataset]

Retrieves a dataset by its ID or name. Arguments:

Optional[str]

The unique identifier of the dataset.

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:

str

The unique identifier of the dataset to update.

Optional[str]

A new name for the dataset.

Optional[str]

A new description for the dataset.

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:

str

The unique identifier of the dataset to delete.

Returns: The result of the deletion operation.

create_experiment

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

Creates a new experiment associated with a specific dataset. Arguments:

str

The name of the experiment.

Optional[str]

The unique identifier of the dataset.

Optional[str]

The identifier of the prompt associated with the experiment.

Optional[Dict]

Additional parameters for the experiment.

Returns:

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:

DatasetExperimentItem

The experiment item to be created, containing all necessary data.

Returns:

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:

str

The unique identifier of the dataset.

Dict

The input data for the dataset item.

Optional[Dict]

The expected output data for the dataset item.

Optional[Dict]

Additional metadata for the dataset item.

Returns:

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:

str

The unique identifier of the dataset item to retrieve.

Returns:

The dataset item data.

delete_dataset_item

def delete_dataset_item(id: str)

Deletes a dataset item by its unique identifier. Arguments:

str

The unique identifier of the dataset item to delete.

Returns:

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:

str

The unique identifier of the dataset.

str

The unique identifier of the step to add.

Optional[Dict]

Additional metadata for the step being added.

Returns:

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:

str

The unique identifier of the dataset.

str

The unique identifier of the generation to add.

Optional[Dict]

Additional metadata for the generation being added.

Returns:

The result of adding the generation to the dataset.

get_or_create_prompt_lineage

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

Creates a prompt lineage with the specified name and optional description. If the prompt lineage with that name already exists, it is returned. Arguments:

str

The name of the prompt lineage.

Optional[str]

An optional description of the prompt lineage.

Returns:

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:

str

The name of the prompt to retrieve or create.

List[GenerationMessage]

A list of template messages for the prompt.

Optional[Dict]

Optional settings for the prompt.

Optional[List[Dict]]

Optional tool options for the model

Returns:

The prompt that was retrieved or created.

get_prompt

def get_prompt(id: Optional[str] = None,
               name: Optional[str] = None,
               version: Optional[int] = None) -> 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:

str

The unique identifier of the prompt to retrieve.

str

The name of the prompt to retrieve.

Optional[int]

The version number of the prompt to retrieve.

Returns:

The prompt with the given identifier or name.

get_prompt_ab_testing

def get_prompt_ab_testing(name: str) -> List[PromptRollout]

Get the A/B testing configuration for a prompt lineage. Arguments:

str

The name of the prompt lineage.

Returns: List[PromptRollout]

update_prompt_ab_testing

def update_prompt_ab_testing(name: str, rollouts: List[PromptRollout]) -> Dict

Update the A/B testing configuration for a prompt lineage. Arguments:

str

The name of the prompt lineage.

List[PromptRollout]

The percentage rollout for each prompt version.

Returns: Dict

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

class AsyncLiteralAPI(BaseLiteralAPI)

from literalai import AsyncLiteralClient
# Initialize the client
async_literalai_client = AsyncLiteralClient(api_key="your_api_key_here")
# Access the API's methods
print(async_literalai_client.api)

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:

Optional[int]

The number of users to retrieve.

Optional[str]

A cursor for use in pagination, fetching records after this cursor.

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:

Optional[str]

The unique identifier of the user to retrieve.

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:

str

The identifier for the new user.

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:

str

The unique identifier of the user to update.

Optional[str]

New identifier for the user.

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:

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:

str

The identifier of the user to retrieve or create.

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:

Optional[int]

The number of threads to retrieve.

Optional[str]

A cursor for use in pagination, fetching records after this cursor.

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. step_types_to_keep (Optional[List[StepType]]) : If set, only steps of the corresponding types will be returned

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:

Optional[int]

The number of threads to list.

Optional[str]

A cursor for use in pagination, fetching records after this cursor.

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:

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,
                        tags: Optional[List[str]] = None)

Asynchronously creates a new thread with specified details. Arguments:

Optional[str]

The name of the thread.

Optional[Dict]

Metadata associated with the thread.

Optional[str]

Identifier for the participant associated with the thread.

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,
                        tags: Optional[List[str]] = None)

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

str

The unique identifier of the thread to upsert.

Optional[str]

The name of the thread.

Optional[Dict]

Metadata associated with the thread.

Optional[str]

Identifier for the participant associated with the thread.

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,
                        tags: Optional[List[str]] = None)

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

str

The unique identifier of the thread to update.

Optional[str]

New name of the thread.

Optional[Dict]

New metadata for the thread.

Optional[str]

New identifier for the participant.

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:

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:

Optional[int]

The number of scores to retrieve.

Optional[str]

A cursor for use in pagination, fetching records after this cursor.

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:

List[literalai.observability.step.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:

str

The name of the score.

float

The numerical value of the score.

ScoreType

The type of the score.

Optional[str]

The ID of the step associated with the score.

Deprecated - use step_id

Optional[str]

The ID of the dataset experiment item associated with the score.

Optional[str]

A comment associated with the score.

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:

str

The unique identifier of the score to update.

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:

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: Optional[str] = None,
        mime: Optional[str] = "application/octet-stream") -> Dict

Asynchronously uploads a file to the server. Arguments:

Union[bytes, str]

The content of the file to upload.

str

The ID of the thread associated with the file.

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:

str

The ID of the thread associated with the attachment.

str

The ID of the step associated with the attachment.

Optional[str]

An optional unique identifier for the attachment.

Optional[Dict]

Optional metadata for the attachment.

Optional[str]

The MIME type of the attachment.

Optional[str]

The name of the attachment.

Optional[str]

The object key for the attachment if already uploaded.

Optional[str]

The URL of the attachment if already uploaded.

Optional[Union[bytes, str]]

The content of the attachment to upload.

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:

str

The unique identifier of the attachment to update.

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:

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:

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,
                      root_run_id: Optional[str] = None)

Asynchronously creates a new step with the specified parameters. Arguments:

Optional[str]

The ID of the thread associated with the step.

Optional[StepType]

The type of the step, defaults to “undefined”.

Optional[str]

The start time of the step.

Optional[str]

The end time of the step.

Optional[Dict]

Input data for the step.

Optional[Dict]

Output data from the step.

Optional[Dict]

Metadata associated with the step.

Optional[str]

The ID of the parent step, if any.

Optional[str]

The name of the step.

Optional[List[str]]

Tags associated with the step.

Optional[str]

The ID of the root run, if any.

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:

str

The unique identifier of the step to update.

Optional[StepType]

The type of the step.

Optional[str]

Input data for the step.

Optional[str]

Output data from the step.

Optional[Dict]

Metadata associated with the step.

Optional[str]

The name of the step.

Optional[List[str]]

Tags associated with the step.

Optional[str]

The start time of the step.

Optional[str]

The end time of the step.

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:

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:

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:

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:

Optional[int]

The number of generations to retrieve.

Optional[str]

A cursor for use in pagination, fetching records after this cursor.

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:

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:

str

The name of the dataset.

Optional[str]

A description of the dataset.

Optional[Dict]

Metadata associated with the dataset.

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:

Optional[str]

The unique identifier of the dataset to retrieve.

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:

str

The unique identifier of the dataset to update.

Optional[str]

The new name of the dataset.

Optional[str]

A new description for the dataset.

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:

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:

DatasetExperimentItem

The experiment item to be created.

Returns:

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:

str

The unique identifier of the dataset.

Dict

The input data for the dataset item.

Optional[Dict]

The expected output data for the dataset item.

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:

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:

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:

str

The unique identifier of the dataset.

str

The unique identifier of the step to add.

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:

str

The unique identifier of the dataset.

str

The unique identifier of the generation to add.

Optional[Dict]

Additional metadata for the generation being added.

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

​LiteralAPI

​get_users

​get_user

​create_user

​update_user

​delete_user

​get_or_create_user

​get_threads

​list_threads

​get_thread

​create_thread

​upsert_thread

​update_thread

​delete_thread

​create_score

​update_score

​delete_score

​upload_file

​create_attachment

​update_attachment

​get_attachment

​delete_attachment

​create_step

​update_step

​get_steps

​get_step

​delete_step

​send_steps

​get_generations

​create_generation

​create_dataset

​get_dataset

​update_dataset

​delete_dataset

​create_experiment

​create_experiment_item

​create_dataset_item

​get_dataset_item

​delete_dataset_item

​add_step_to_dataset

​add_generation_to_dataset

​get_or_create_prompt_lineage

​get_or_create_prompt

​get_prompt

​get_prompt_ab_testing

​update_prompt_ab_testing

​get_my_project_id

​AsyncLiteralAPI

​get_users

​get_user

​create_user

​update_user

​delete_user

​get_or_create_user

​get_threads

​list_threads

​get_thread

​create_thread

​upsert_thread

​update_thread

​delete_thread

​get_scores

​create_scores

​create_score

​update_score

​delete_score

​upload_file

​create_attachment

​update_attachment

​get_attachment

​delete_attachment

​create_step

​update_step

​get_step

​delete_step

​send_steps

​get_generations

​create_generation

​create_dataset

​get_dataset

LiteralAPI

get_users

get_user

create_user

update_user

delete_user

get_or_create_user

get_threads

list_threads

get_thread

create_thread

upsert_thread

update_thread

delete_thread

create_score

update_score

delete_score

upload_file

create_attachment

update_attachment

get_attachment

delete_attachment

create_step

update_step

get_steps

get_step

delete_step

send_steps

get_generations

create_generation

create_dataset

get_dataset

update_dataset

delete_dataset

create_experiment

create_experiment_item

create_dataset_item

get_dataset_item

delete_dataset_item

add_step_to_dataset

add_generation_to_dataset

get_or_create_prompt_lineage

get_or_create_prompt

get_prompt

get_prompt_ab_testing

update_prompt_ab_testing

get_my_project_id

AsyncLiteralAPI

get_users

get_user

create_user

update_user

delete_user

get_or_create_user

get_threads

list_threads

get_thread

create_thread

upsert_thread

update_thread

delete_thread

get_scores

create_scores

create_score

update_score

delete_score

upload_file

create_attachment

update_attachment

get_attachment

delete_attachment

create_step

update_step

get_step

delete_step

send_steps

get_generations

create_generation

create_dataset

get_dataset