> ## Documentation Index > Fetch the complete documentation index at: https://docs.trainlyai.com/llms.txt > Use this file to discover all available pages before exploring further. # Python SDK > Instrument your AI pipelines with Trainly's Python SDK for tracing, scoring, prompt management, and analytics. ## Installation ```bash theme={null} pip install trainly ``` ## Quick Start ```python theme={null} from trainly import TrainlyClient client = TrainlyClient( api_key="tk_...", project_id="proj_..." ) ``` You can also set credentials via environment variables: ```bash theme={null} export TRAINLY_API_KEY="tk_..." export TRAINLY_PROJECT_ID="proj_..." ``` ```python theme={null} 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. ```python theme={null} @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 ```python theme={null} @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. ```python theme={null} 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. ```python theme={null} 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. ```python theme={null} @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. ```python Manual Score theme={null} client.score( trace_id="trace_abc123", name="accuracy", value=0.95, comment="Output matched expected format" ) ``` ```python AI Judge Score theme={null} 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. ```python theme={null} 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. ```python theme={null} 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. ```python theme={null} 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. ```python theme={null} 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. ```python theme={null} 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}") ```