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:
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)
Retrieves a user based on the provided ID or identifier.
Arguments:
The unique ID of the user.
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:
A unique identifier for the user, such as a username or email.
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:
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.
Returns:
The result of the GraphQL call to update the user.
delete_user
Deletes a user identified by the given ID.
Arguments:
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:
The identifier of the user to retrieve or create.
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:
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)
Lists threads based on pagination and optional filters, similar to get_threads but may include additional processing.
Arguments:
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.
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:
Metadata associated with the thread.
Identifier for the participant.
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:
The unique identifier of the thread.
Metadata associated with the thread.
Identifier for the participant.
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:
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.
Returns:
The updated thread.
delete_thread
def delete_thread(id: str)
Deletes a thread identified by its ID.
Arguments:
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:
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.
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:
The unique identifier of the score to update.
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:
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:
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’.
Returns:
A dictionary containing the object key and URL of the uploaded file, or None values if the upload fails.
create_attachment
def create_attachment(thread_id: Optional[str] = None,
step_id: Optional[str] = None,
id: Optional[str] = None,
metadata: Optional[Dict] = None,
mime: Optional[str] = None,
name: Optional[str] = None,
object_key: Optional[str] = None,
url: Optional[str] = None,
content: Optional[Union[bytes, str]] = None,
path: Optional[str] = None) -> "Attachment"
Creates an attachment associated with a thread and step, potentially uploading file content.
Arguments:
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.
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:
The unique identifier of the attachment to update.
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:
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:
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:
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.
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:
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.
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:
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.
Returns:
The step data as returned by the GraphQL helper function.
delete_step
Deletes a step identified by its ID.
Arguments:
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:
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])
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:
A description of the dataset.
Additional metadata for the dataset.
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:
The unique identifier 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:
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.
Returns:
The result of the dataset update operation.
delete_dataset
def delete_dataset(id: str)
Deletes a dataset identified by its ID.
Arguments:
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:
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.
Returns:
The newly created experiment object.
create_experiment_item
def create_experiment_item(
experiment_item: DatasetExperimentItem) -> DatasetExperimentItem
Creates an experiment item within an existing experiment.
Arguments:
The experiment item to be created, containing all necessary data.
Returns:
The newly created experiment item with scores attached.
create_dataset_item
def create_dataset_item(dataset_id: str,
input: Dict,
expected_output: Optional[Dict] = None,
metadata: Optional[Dict] = None)
Creates a new dataset item with the specified properties.
Arguments:
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.
Returns:
The result of the dataset item creation operation.
get_dataset_item
def get_dataset_item(id: str)
Retrieves a dataset item by its unique identifier.
Arguments:
The unique identifier of the dataset item to retrieve.
Returns:
delete_dataset_item
def delete_dataset_item(id: str)
Deletes a dataset item by its unique identifier.
Arguments:
The unique identifier of the dataset item to delete.
Returns:
The result of the dataset item deletion operation.
add_step_to_dataset
def add_step_to_dataset(dataset_id: str,
step_id: str,
metadata: Optional[Dict] = None)
Adds a step to a dataset.
Arguments:
The unique identifier of the dataset.
The unique identifier of the step to add.
Additional metadata for the step being added.
Returns:
The result of adding the step to the dataset.
add_generation_to_dataset
def add_generation_to_dataset(dataset_id: str,
generation_id: str,
metadata: Optional[Dict] = None)
Adds a generation to a dataset.
Arguments:
The unique identifier of the dataset.
The unique identifier of the generation to add.
Additional metadata for the generation being added.
Returns:
The result of adding the generation to the dataset.
get_or_create_prompt_lineage
def get_or_create_prompt_lineage(name: str, description: Optional[str] = None)
Creates a prompt lineage with the specified name and optional description.
If the prompt lineage with that name already exists, it is returned.
Arguments:
The name of the prompt lineage.
An optional description of the prompt lineage.
Returns:
The result of the prompt lineage creation operation.
get_or_create_prompt
def get_or_create_prompt(name: str,
template_messages: List[GenerationMessage],
settings: Optional[ProviderSettings] = None,
tools: Optional[List[Dict]] = None) -> Prompt
A Prompt is fully defined by its name, template_messages, settings and tools.
If a prompt already exists for the given arguments, it is returned.
Otherwise, a new prompt is created.
Arguments:
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
Returns:
The prompt that was retrieved or created.
get_prompt
def get_prompt(id: Optional[str] = None,
name: Optional[str] = None,
version: Optional[int] = None) -> Prompt
Gets a prompt either by:
id- or
name and (optional) version
Either the id or the name must be provided.
If both are provided, the id is used.
Arguments:
The unique identifier of the prompt to retrieve.
The name of the prompt to retrieve.
The version number of the prompt to retrieve.
Returns:
The prompt with the given identifier or name.
get_prompt_ab_testing
def get_prompt_ab_testing(name: str) -> List[PromptRollout]
Get the A/B testing configuration for a prompt lineage.
Arguments:
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:
The name of the prompt lineage.
The percentage rollout for each prompt version.
Returns:
Dict
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:
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)
Asynchronously retrieves a user by ID or identifier.
Arguments:
The unique identifier of the user to retrieve.
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:
The identifier for the new user.
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:
The unique identifier of the user to update.
New identifier for the user.
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:
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:
The identifier of the user to retrieve or create.
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:
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)
Asynchronously lists threads based on pagination and optional filters and ordering, similar to 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)
Asynchronously retrieves a thread by its ID.
Arguments:
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:
Metadata associated with the thread.
Identifier for the participant associated with the thread.
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:
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.
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:
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.
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:
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:
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])
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:
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.
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:
The unique identifier of the score to update.
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:
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:
The content of the file to upload.
The ID of the thread associated with the file.
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:
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.
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:
The unique identifier of the attachment to update.
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:
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:
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:
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.
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:
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.
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:
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:
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:
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])
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:
A description of the dataset.
Metadata associated with the dataset.
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:
The unique identifier of the dataset to retrieve.
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:
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.
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:
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:
The experiment item to be created.
Returns:
The created experiment item with updated scores.
create_dataset_item
async def create_dataset_item(dataset_id: str,
input: Dict,
expected_output: Optional[Dict] = None,
metadata: Optional[Dict] = None)
Asynchronously creates a dataset item.
Arguments:
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.
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:
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:
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:
The unique identifier of the dataset.
The unique identifier of the step to add.
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:
The unique identifier of the dataset.
The unique identifier of the generation to add.
Additional metadata for the generation being added.
Returns:
The result of the GraphQL helper function for adding a generation to a dataset.
