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)
The number of users to retrieve.
A cursor for use in pagination, fetching records after this cursor.
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)
The unique ID of the user.
A unique identifier for the user, such as a username or email.
create_user
def create_user(identifier: str, metadata: Optional[Dict] = None)
A unique identifier for the user, such as a username or email.
Additional data associated with the user.
update_user
def update_user(id: str,
identifier: Optional[str] = None,
metadata: Optional[Dict] = None)
The unique ID of the user to update.
A new identifier for the user, such as a username or email.
New or updated metadata for the user.
delete_user
Deletes a user identified by the given ID.
Arguments:
The unique ID of the user to delete.
get_or_create_user
def get_or_create_user(identifier: str, metadata: Optional[Dict] = None)
The identifier of the user to retrieve or create.
Metadata to associate with the user if they are created.
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)
Number of threads to fetch.
Cursor for pagination, fetch threads after this cursor.
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)
Number of threads to list.
Cursor for pagination, list threads after this cursor.
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
Retrieves a single thread by its ID.
Arguments:
The unique identifier of the thread.
create_thread
def create_thread(name: Optional[str] = None,
metadata: Optional[Dict] = None,
participant_id: Optional[str] = None,
tags: Optional[List[str]] = None)
Metadata associated with the thread.
Identifier for the participant.
List of tags associated with the 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)
The unique identifier of the thread.
Metadata associated with the thread.
Identifier for the participant.
List of tags associated with the 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)
The unique identifier of the thread to update.
New metadata for the thread.
New identifier for the participant.
New list of tags for the thread.
delete_thread
def delete_thread(id: str)
The unique identifier of the thread to delete.
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)
The numerical value of the score.
The ID of the step associated with the score.
The ID of the generation associated with the score.
dataset_experiment_item_id
The ID of the dataset experiment item associated with the score.
An optional comment about the score.
Optional tags associated with the score.
update_score
def update_score(id: str, update_params: ScoreUpdate)
The unique identifier of the score to update.
A dictionary of parameters to update in the score.
delete_score
def delete_score(id: str)
The unique identifier of the score to delete.
upload_file
def upload_file(content: Union[bytes, str],
thread_id: Optional[str] = None,
mime: Optional[str] = "application/octet-stream") -> Dict
The content of the file to upload.
The ID of the thread associated with the file.
The MIME type of the file. Defaults to ‘application/octet-stream’.
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"
The ID of the thread to which the attachment is linked.
The ID of the step to which the attachment is linked.
The ID of the attachment, if updating an existing one.
Metadata associated with the attachment.
MIME type of the file, if content is provided.
Object key of the uploaded file, if already known.
URL of the uploaded file, if already known.
content
Optional[Union[bytes, str]]
File content to upload.
Path where the file should be stored.
literalai.observability.step.Attachment
The created or updated attachment object.
update_attachment
def update_attachment(id: str, update_params: AttachmentUpload)
The unique identifier of the attachment to update.
The parameters to update in the attachment.
get_attachment
def get_attachment(id: str)
The unique identifier of the attachment to retrieve.
delete_attachment
def delete_attachment(id: str)
The unique identifier of the attachment to delete.
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)
The ID of the thread this step is associated with.
The type of the step, defaults to “undefined”.
The start time of the step.
The end time of the step.
Output data from the step.
Metadata associated with the step.
The ID of the parent step, if any.
Tags associated with the step.
The ID of the root run, if any.
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)
The unique identifier of the step to update.
Output data from the step.
Metadata associated with the step.
Tags associated with the step.
The start time of the step.
The end time of the step.
The ID of the parent step, if any.
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]
Number of steps to fetch.
Cursor for pagination, fetch steps after this cursor.
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
Retrieves a step by its ID.
Arguments:
The unique identifier of the step to retrieve.
delete_step
Deletes a step identified by its ID.
Arguments:
The unique identifier of the step to delete.
send_steps
def send_steps(steps: List[Union[StepDict, "Step"]])
steps
List[Union[StepDict, Step]]
A list of steps or step dictionaries to send.
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)
The number of generations to retrieve.
A cursor for use in pagination, fetching records after this cursor.
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])
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)
generation
Union[ChatGeneration, CompletionGeneration]
The generation data to create.
create_dataset
def create_dataset(name: str,
description: Optional[str] = None,
metadata: Optional[Dict] = None,
type: DatasetType = "key_value")
A description of the dataset.
Additional metadata for the dataset.
The type of the dataset, defaults to “key_value”.
get_dataset
def get_dataset(id: Optional[str] = None,
name: Optional[str] = None) -> Optional[Dataset]
The unique identifier of the dataset.
update_dataset
def update_dataset(id: str,
name: Optional[str] = None,
description: Optional[str] = None,
metadata: Optional[Dict] = None)
The unique identifier of the dataset to update.
A new name for the dataset.
A new description for the dataset.
New or updated metadata for the dataset.
delete_dataset
def delete_dataset(id: str)
The unique identifier of the dataset to delete.
create_experiment
def create_experiment(name: str,
dataset_id: Optional[str] = None,
prompt_id: Optional[str] = None,
params: Optional[Dict] = None) -> "DatasetExperiment"
The name of the experiment.
The unique identifier of the dataset.
The identifier of the prompt associated with the experiment.
Additional parameters for the experiment.
The newly created experiment object.
create_experiment_item
def create_experiment_item(
experiment_item: DatasetExperimentItem) -> DatasetExperimentItem
The experiment item to be created, containing all necessary data.
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)
The unique identifier of the dataset.
The input data for the dataset item.
The expected output data for the dataset item.
Additional metadata for the dataset item.
The result of the dataset item creation operation.
get_dataset_item
def get_dataset_item(id: str)
The unique identifier of the dataset item to retrieve.
delete_dataset_item
def delete_dataset_item(id: str)
The unique identifier of the dataset item to delete.
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)
The unique identifier of the dataset.
The unique identifier of the step to add.
Additional metadata for the step being added.
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)
The unique identifier of the dataset.
The unique identifier of the generation to add.
Additional metadata for the generation being added.
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)
The name of the prompt lineage.
An optional description of the prompt lineage.
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
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:
The name of the prompt to retrieve or create.
A list of template messages for the prompt.
Optional settings for the prompt.
Optional tool options for the model
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
id- or
name and (optional) version
Either the id or the name must be provided.
If both are provided, the id is used.
Arguments:
The unique identifier of the prompt to retrieve.
The name of the prompt to retrieve.
The version number of the prompt to retrieve.
The prompt with the given identifier or name.
get_prompt_ab_testing
def get_prompt_ab_testing(name: str) -> List[PromptRollout]
The name of the prompt lineage.
update_prompt_ab_testing
def update_prompt_ab_testing(name: str, rollouts: List[PromptRollout]) -> Dict
The name of the prompt lineage.
The percentage rollout for each prompt version.
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)
The number of users to retrieve.
A cursor for use in pagination, fetching records after this cursor.
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)
The unique identifier of the user to retrieve.
An alternative identifier for the user.
create_user
async def create_user(identifier: str, metadata: Optional[Dict] = None)
The identifier for the new user.
Additional metadata for the user.
update_user
async def update_user(id: str,
identifier: Optional[str] = None,
metadata: Optional[Dict] = None)
The unique identifier of the user to update.
New identifier for the user.
New metadata for the user.
delete_user
async def delete_user(id: str)
The unique identifier of the user to delete.
get_or_create_user
async def get_or_create_user(identifier: str, metadata: Optional[Dict] = None)
The identifier of the user to retrieve or create.
Metadata for the user if creation is necessary.
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)
The number of threads to retrieve.
A cursor for use in pagination, fetching records after this cursor.
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)
get_threads.
Arguments:
The number of threads to list.
A cursor for use in pagination, fetching records after this cursor.
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)
The unique identifier of the thread to retrieve.
create_thread
async def create_thread(name: Optional[str] = None,
metadata: Optional[Dict] = None,
participant_id: Optional[str] = None,
tags: Optional[List[str]] = None)
Metadata associated with the thread.
Identifier for the participant associated with the thread.
Tags associated with the 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)
The unique identifier of the thread to upsert.
Metadata associated with the thread.
Identifier for the participant associated with the thread.
Tags associated with the 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)
The unique identifier of the thread to update.
New metadata for the thread.
New identifier for the participant.
New list of tags for the thread.
delete_thread
async def delete_thread(id: str)
The unique identifier of the thread to delete.
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)
The number of scores to retrieve.
A cursor for use in pagination, fetching records after this cursor.
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])
scores
List[literalai.observability.step.ScoreDict]
A list of dictionaries representing the scores to be created.
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)
The numerical value of the score.
The ID of the step associated with the score.
dataset_experiment_item_id
The ID of the dataset experiment item associated with the score.
A comment associated with the score.
A list of tags associated with the score.
update_score
async def update_score(id: str, update_params: ScoreUpdate)
The unique identifier of the score to update.
A dictionary of parameters to update.
delete_score
async def delete_score(id: str)
The unique identifier of the score to delete.
upload_file
async def upload_file(
content: Union[bytes, str],
thread_id: Optional[str] = None,
mime: Optional[str] = "application/octet-stream") -> Dict
The content of the file to upload.
The ID of the thread associated with the file.
The MIME type of the 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"
The ID of the thread associated with the attachment.
The ID of the step associated with the attachment.
An optional unique identifier for the attachment.
Optional metadata for the attachment.
The MIME type of the attachment.
The name of the attachment.
The object key for the attachment if already uploaded.
The URL of the attachment if already uploaded.
content
Optional[Union[bytes, str]]
The content of the attachment to upload.
The file path of the attachment if it is to be uploaded from a local file.
update_attachment
async def update_attachment(id: str, update_params: AttachmentUpload)
The unique identifier of the attachment to update.
A dictionary of parameters to update the attachment.
get_attachment
async def get_attachment(id: str)
The unique identifier of the attachment to retrieve.
delete_attachment
async def delete_attachment(id: str)
The unique identifier of the attachment to delete.
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)
The ID of the thread associated with the step.
The type of the step, defaults to “undefined”.
The start time of the step.
The end time of the step.
Output data from the step.
Metadata associated with the step.
The ID of the parent step, if any.
Tags associated with the step.
The ID of the root run, if any.
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)
The unique identifier of the step to update.
Output data from the step.
Metadata associated with the step.
Tags associated with the step.
The start time of the step.
The end time of the step.
The ID of the parent step, if any.
get_step
async def get_step(id: str)
The unique identifier of the step to retrieve.
delete_step
async def delete_step(id: str)
The unique identifier of the step to delete.
send_steps
async def send_steps(steps: List[Union[StepDict, "Step"]])
steps
List[Union[StepDict, Step]]
A list of steps or step dictionaries to send.
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)
The number of generations to retrieve.
A cursor for use in pagination, fetching records after this cursor.
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])
generation
Union[ChatGeneration, CompletionGeneration]
The generation data to create.
create_dataset
async def create_dataset(name: str,
description: Optional[str] = None,
metadata: Optional[Dict] = None,
type: DatasetType = "key_value")
A description of the dataset.
Metadata associated with the dataset.
The type of the dataset, defaults to “key_value”.
get_dataset
async def get_dataset(id: Optional[str] = None, name: Optional[str] = None)
The unique identifier of the dataset to retrieve.
The name of the dataset to retrieve.
update_dataset
async def update_dataset(id: str,
name: Optional[str] = None,
description: Optional[str] = None,
metadata: Optional[Dict] = None)
The unique identifier of the dataset to update.
The new name of the dataset.
A new description for the dataset.
New metadata for the dataset.
delete_dataset
async def delete_dataset(id: str)
The unique identifier of the dataset to delete.
create_experiment_item
async def create_experiment_item(
experiment_item: DatasetExperimentItem) -> DatasetExperimentItem
The experiment item to be created.
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)
The unique identifier of the dataset.
The input data for the dataset item.
The expected output data for the dataset item.
Additional metadata for the dataset item.
get_dataset_item
async def get_dataset_item(id: str)
The unique identifier of the dataset item.
delete_dataset_item
async def delete_dataset_item(id: str)
The unique identifier of the dataset item to delete.
add_step_to_dataset
async def add_step_to_dataset(dataset_id: str,
step_id: str,
metadata: Optional[Dict] = None)
The unique identifier of the dataset.
The unique identifier of the step to add.
Additional metadata for the step being added.
add_generation_to_dataset
async def add_generation_to_dataset(dataset_id: str,
generation_id: str,
metadata: Optional[Dict] = None)
The unique identifier of the dataset.
The unique identifier of the generation to add.
Additional metadata for the generation being added.
