Api
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:
filtersOptional[users_filters] - Filters to apply to the user query.
Returns:
get_user
def get_user(id: Optional[str] = None, identifier: Optional[str] = None)
Retrieves a user based on the provided ID or identifier.
Arguments:
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:
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:
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:
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:
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:
filtersOptional[threads_filters] - Filters to apply on the threads query.order_byOptional[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:
filtersOptional[threads_filters] - Filters to apply on the threads listing.order_byOptional[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:
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:
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:
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:
Returns:
The updated thread.
delete_thread
def delete_thread(id: str)
Deletes a thread identified by its ID.
Arguments:
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:
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:
Returns:
The result of the update operation.
delete_score
def delete_score(id: str)
Deletes a score identified by its ID.
Arguments:
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:
Returns:
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:
Returns:
update_attachment
def update_attachment(id: str, update_params: AttachmentUpload)
Updates an existing attachment with new parameters.
Arguments:
Returns:
The result of the update operation.
get_attachment
def get_attachment(id: str)
Retrieves an attachment by its ID.
Arguments:
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:
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:
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:
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:
filtersOptional[steps_filters] - Filters to apply on the steps query.order_byOptional[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:
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:
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:
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:
filtersOptional[generations_filters] - Filters to apply to the generations query.order_byOptional[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:
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:
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:
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:
Returns:
The result of the dataset update operation.
delete_dataset
def delete_dataset(id: str)
Deletes a dataset identified by its ID.
Arguments:
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:
Returns:
create_experiment_item
def create_experiment_item(
experiment_item: DatasetExperimentItem) -> DatasetExperimentItem
Creates an experiment item within an existing experiment.
Arguments:
Returns:
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:
Returns:
get_dataset_item
def get_dataset_item(id: str)
Retrieves a dataset item by its unique identifier.
Arguments:
Returns:
delete_dataset_item
def delete_dataset_item(id: str)
Deletes a dataset item by its unique identifier.
Arguments:
Returns:
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:
Returns:
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:
Returns:
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:
Returns:
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:
Returns:
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
nameand (optional)version
Either the id or the name must be provided.
If both are provided, the id is used.
Arguments:
Returns:
get_prompt_ab_testing
def get_prompt_ab_testing(name: str) -> List[PromptRollout]
Get the A/B testing configuration for a prompt lineage.
Arguments:
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:
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:
filtersOptional[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:
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:
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:
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:
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:
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:
filtersOptional[threads_filters] - Filters to apply to the thread query.order_byOptional[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:
filtersOptional[threads_filters] - Filters to apply to the thread query.order_byOptional[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:
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:
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:
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:
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:
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:
filtersOptional[scores_filters] - Filters to apply to the scores query.order_byOptional[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:
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:
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:
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:
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:
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:
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:
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:
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:
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:
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:
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:
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:
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:
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:
filtersOptional[generations_filters] - Filters to apply to the generations query.order_byOptional[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:
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:
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:
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:
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:
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:
Returns:
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:
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:
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:
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:
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:
Returns:
The result of the GraphQL helper function for adding a generation to a dataset.