​
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:

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

​
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,
                  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.
​
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,
                  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.
​
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,
                  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.
​
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: 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:

​
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:

​
literalai.observability.step.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,
                root_run_id: Optional[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.
​
root_run_id
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:

​
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.

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:

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

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(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:

​
name
str
The name of the experiment.
​
dataset_id
Optional[str]
The unique identifier of the dataset.
​
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.

​
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:

​
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.
​
tools
Optional[List[Dict]]
Optional tool options for the model

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] = 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:

​
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_prompt_ab_testing

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

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

Arguments:

​
name
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:

​
name
str
The name of the prompt lineage.
​
rollouts
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:

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

​
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,
                        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.
​
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,
                        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.
​
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,
                        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.
​
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[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:

​
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
Deprecated - use step_id
​
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: Optional[str] = None,
        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,
                      root_run_id: Optional[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.
​
root_run_id
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:

​
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.

Was this page helpful?

ChangelogClient