Python SDK - Trainly Documentation

Installation

pip install trainly

Quick Start

from trainly import TrainlyClient

client = TrainlyClient(
    api_key="tk_...",
    project_id="proj_..."
)

You can also set credentials via environment variables:

export TRAINLY_API_KEY="tk_..."
export TRAINLY_PROJECT_ID="proj_..."

client = TrainlyClient()  # reads from env

@observe Decorator

The @observe decorator is the primary way to instrument your AI calls. It automatically captures inputs, outputs, latency, and exceptions.

@client.observe(model="gpt-4o", tags=["production"])
def summarize(text: str) -> str:
    return openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": text}]
    ).choices[0].message.content

Parameters

Parameter	Type	Description
`model`	`str`	Model name (e.g. `"gpt-4o"`, `"claude-sonnet-4-20250514"`)
`tags`	`list[str]`	Filterable labels
`expected_output`	`str`	Ground truth for evaluation
`trace_id`	`str`	Custom trace identifier
`metadata`	`dict`	Arbitrary key-value pairs
`version`	`str`	Code or prompt version tag
`custom_attributes`	`dict`	Additional structured data
`span_name`	`str`	Override the default span name
`capture_exceptions`	`bool`	Log exceptions as trace errors (default `True`)
`session_id`	`str`	Group traces into a session

With All Options

@client.observe(
    model="gpt-4o",
    tags=["staging", "summarization"],
    version="v2.1",
    metadata={"team": "ml-ops"},
    capture_exceptions=True,
    session_id="session_abc123"
)
def generate_summary(article: str) -> str:
    return call_llm(article)

Manual Logging

Use client.observe.log() when you need full control over what gets recorded.

import time

start = time.time()
result = call_my_model(prompt)
latency = (time.time() - start) * 1000

client.observe.log(
    input=prompt,
    output=result,
    model="gpt-4o",
    latency_ms=latency,
    tags=["batch-job"],
    token_usage={"prompt_tokens": 120, "completion_tokens": 45},
    status="success",
    cost=0.0023,
    metadata={"run_id": "abc"},
)

Parameters

Parameter	Type	Description
`input`	`str`	The input sent to the model
`output`	`str`	The model’s response
`model`	`str`	Model identifier
`latency_ms`	`float`	Execution time in milliseconds
`tags`	`list[str]`	Filterable labels
`expected_output`	`str`	Ground truth for evaluation
`trace_id`	`str`	Custom trace identifier
`token_usage`	`dict`	Token counts (`prompt_tokens`, `completion_tokens`)
`metadata`	`dict`	Arbitrary key-value pairs
`status`	`str`	`"success"` or `"error"`
`error`	`str`	Error message if applicable
`version`	`str`	Version tag
`custom_attributes`	`dict`	Additional structured data
`tool_calls`	`list[dict]`	Tool/function calls made during execution
`input_structured`	`dict`	Structured input (e.g. message arrays)
`output_structured`	`dict`	Structured output (e.g. parsed JSON)
`spans`	`list[dict]`	Sub-step span data
`cost`	`float`	Estimated cost in USD
`session_id`	`str`	Session grouping identifier

Sessions

Group related traces into a session using the agent_session() context manager. This is useful for multi-step agent workflows.

with client.observe.agent_session() as session:
    plan = planner(task)       # trace 1
    result = executor(plan)    # trace 2
    review = reviewer(result)  # trace 3
    # All three traces share the same session_id

Spans

Break a single trace into sub-steps with the span() context manager.

@client.observe(model="gpt-4o")
def agent_pipeline(query: str) -> str:
    with client.observe.span("retrieval", kind="retriever") as span:
        span.set_input(query)
        docs = search_index(query)
        span.set_output(docs)
        span.set_attribute("num_results", len(docs))

    with client.observe.span("generation", kind="llm") as span:
        span.set_input(docs)
        span.set_model("gpt-4o")
        answer = generate(docs, query)
        span.set_output(answer)
        span.set_token_usage({"prompt_tokens": 200, "completion_tokens": 80})
        span.set_cost(0.004)

    return answer

Span Methods

Method	Description
`set_input(value)`	Record the span’s input
`set_output(value)`	Record the span’s output
`set_attribute(key, value)`	Attach a custom attribute
`set_model(name)`	Set the model used in this span
`set_token_usage(usage)`	Record token counts
`set_cost(amount)`	Record cost in USD

Scoring

Score traces for evaluation, either manually or with an AI judge.

client.score(
    trace_id="trace_abc123",
    name="accuracy",
    value=0.95,
    comment="Output matched expected format"
)

result = client.score_with_judge(
    scorer_slug="faithfulness",
    trace_id="trace_abc123",
    input="What is the capital of France?",
    output="The capital of France is Paris.",
    expected_output="Paris"
)

Prompt Management

Manage versioned prompts and build them with template variables.

prompt = client.get_prompt(slug="summarize-v2", version="latest")

rendered = prompt.build(
    topic="quarterly earnings",
    tone="professional",
    max_length="200 words"
)

Analytics

Access trace analytics and cost data programmatically.

analytics = client.analytics

# Get traces with filters
traces = analytics.get_query_traces(
    tags=["production"],
    model="gpt-4o",
    limit=50
)

# Get details for a specific trace
detail = analytics.get_trace_details(trace_id="trace_abc123")

# Aggregate metrics
summary = analytics.get_metrics_summary(
    start_date="2026-04-01",
    end_date="2026-04-07"
)

# Cost and performance breakdowns
costs = analytics.get_cost_breakdown(group_by="model")
perf = analytics.get_performance_stats(tags=["production"])

# Export logs as CSV/JSON
analytics.export_query_logs(format="csv", output_path="traces.csv")

Testing

Create test suites, add cases, and run evaluations against your AI functions.

testing = client.testing

suite = testing.create_suite(name="Summarization Tests")

testing.add_test_case(
    suite_id=suite.id,
    input="Explain gravity in one sentence.",
    expected_output="Gravity is the force that attracts objects toward each other."
)

run = testing.run_suite(suite_id=suite.id)

results = testing.get_run_results(run_id=run.id)
for case in results.cases:
    print(f"{case.status}: {case.score}")

Versions

Publish, track, and roll back versioned deployments of your AI pipelines.

versions = client.versions

versions.publish(
    version="v2.1.0",
    metadata={"changelog": "Improved summarization prompt"}
)

all_versions = versions.list_versions()
active = versions.get_active_version()

diff = versions.compare_versions("v2.0.0", "v2.1.0")

versions.rollback(version="v2.0.0")

Error Handling

All SDK errors raise TrainlyError with structured context.

from trainly import TrainlyClient, TrainlyError

client = TrainlyClient()

try:
    client.observe.log(input="test", output="result", model="gpt-4o")
except TrainlyError as e:
    print(f"Status: {e.status_code}")
    print(f"Details: {e.details}")

​Installation

​Quick Start

​@observe Decorator

​Parameters

​With All Options

​Manual Logging

​Parameters

​Sessions

​Spans

​Span Methods

​Scoring

​Prompt Management

​Analytics

​Testing

​Versions

​Error Handling

Installation

Quick Start

@observe Decorator

Parameters

With All Options

Manual Logging

Parameters

Sessions

Spans

Span Methods

Scoring

Prompt Management

Analytics

Testing

Versions

Error Handling