Sending Traces — Python SDK

There are three ways to send traces to Infinium, from simplest to most detailed. Choose based on how much control and detail you need.

Comparison

Method	Best For	Automatic Timing	Step-Level Detail	Auto-Send
`send_task()`	Simple, single-step agents	No	No	Yes
`TaskData` / `create_task_data()`	Full control, structured data	No	Yes	No
`TraceBuilder`	Real agents with multiple steps	Yes	Yes	No

1. `send_task()` — Simple, Inline

The quickest way. Call your LLM, then report what happened in a single call:

import time
from openai import OpenAI
from infinium import InfiniumClient

client = InfiniumClient(agent_id="...", agent_secret="...")
openai = OpenAI()

start = time.time()
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "Classify this support ticket. Return JSON."},
        {"role": "user", "content": ticket_text},
    ],
    response_format={"type": "json_object"},
)
duration = time.time() - start
result = response.choices[0].message.content
usage = response.usage

response = client.send_task(
    name="Classify support ticket",
    description="Categorized an inbound support ticket by type and urgency.",
    duration=duration,
    input_summary=f"Customer ticket: {ticket_text[:200]}",
    output_summary=f"Classification: {result[:300]}",
    llm_usage={
        "model": "gpt-4o",
        "provider": "openai",
        "prompt_tokens": usage.prompt_tokens,
        "completion_tokens": usage.completion_tokens,
        "total_tokens": usage.total_tokens,
        "api_calls_count": 1,
    },
    customer={"customer_name": customer_name, "customer_email": customer_email},
)

Parameters

Parameter	Type	Required	Description
`name`	`str`	Yes	Short name for the task (max 500 chars)
`description`	`str`	Yes	What the agent did (max 10,000 chars)
`duration`	`float`	Yes	Wall-clock time in seconds (0 to 86,400)
`current_datetime`	`str`	No	ISO 8601 timestamp (auto-generated if omitted)
`input_summary`	`str`	No	Summary of the input the agent received
`output_summary`	`str`	No	Summary of what the agent produced
`llm_usage`	`dict` or `LlmUsage`	No	Token counts, model, cost
`steps`	`list`	No	Execution steps
`expected_outcome`	`dict` or `ExpectedOutcome`	No	Maestro’s grading rubric
`environment`	`dict` or `EnvironmentContext`	No	Runtime metadata
`errors`	`list`	No	Errors encountered

send_task() also accepts any domain section as a keyword argument (e.g., customer=, sales=, research=). Raw dicts are automatically coerced to the corresponding dataclass.

2. `TaskData` / `create_task_data()` — Structured

When you need full control over every field, build a TaskData object. The static method create_task_data() validates inputs and coerces raw dicts to dataclasses:

from infinium import (
    InfiniumClient, ExecutionStep, LlmCall, LlmUsage,
    ExpectedOutcome, EnvironmentContext, ErrorDetail,
)

steps = []
errors = []
total_prompt = 0
total_completion = 0

# Step 1: Search
step1_start = time.time()
results = search_knowledge_base(query)
step1_ms = int((time.time() - step1_start) * 1000)

steps.append(ExecutionStep(
    step_number=1,
    action="tool_use",
    description="Search knowledge base for relevant documents",
    duration_ms=step1_ms,
    output_preview=f"Found {len(results)} results",
))

# Step 2: LLM analysis
step2_start = time.time()
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "Analyze these findings..."},
        {"role": "user", "content": str(results)},
    ],
)
step2_ms = int((time.time() - step2_start) * 1000)
analysis = response.choices[0].message.content
usage = response.usage
total_prompt += usage.prompt_tokens
total_completion += usage.completion_tokens

steps.append(ExecutionStep(
    step_number=2,
    action="llm_inference",
    description="Analyze search results with GPT-4o",
    duration_ms=step2_ms,
    llm_call=LlmCall(
        model="gpt-4o",
        provider="openai",
        prompt_tokens=usage.prompt_tokens,
        completion_tokens=usage.completion_tokens,
        latency_ms=step2_ms,
    ),
    output_preview=analysis[:500],
))

# Build TaskData
td = InfiniumClient.create_task_data(
    name="Research: quarterly earnings impact",
    description="Multi-step research with source gathering and LLM analysis",
    duration=time.time() - total_start,
    input_summary=f"Research query: {query}",
    output_summary=analysis[:500],
    steps=steps,
    errors=errors if errors else None,
    expected_outcome=ExpectedOutcome(
        task_objective="Produce sourced analysis of quarterly earnings impact",
        required_deliverables=["Source list", "Analysis", "Executive summary"],
        acceptance_criteria=["At least 3 sources", "Specific data points cited"],
    ),
    llm_usage=LlmUsage(
        model="gpt-4o",
        provider="openai",
        prompt_tokens=total_prompt,
        completion_tokens=total_completion,
        total_tokens=total_prompt + total_completion,
        api_calls_count=1,
    ),
    environment=EnvironmentContext(framework="custom", python_version="3.12"),
)

# Send
client.send_task_data(td)

Dict Coercion

You can pass raw dicts anywhere a dataclass is expected. create_task_data() auto-coerces them:

td = InfiniumClient.create_task_data(
    name="Quick task",
    description="Demo of dict coercion",
    duration=1.0,
    llm_usage={"model": "gpt-4o", "prompt_tokens": 100},  # Coerced to LlmUsage
    customer={"customer_name": "Acme"},                     # Coerced to Customer
)

3. `TraceBuilder` — Recommended for Real Agents

TraceBuilder wraps your agent’s execution. Steps are context managers that automatically capture timing and catch exceptions:

from infinium import TraceBuilder, LlmUsage

trace = TraceBuilder(
    "Blog Post Generator",
    "Generate a publish-ready blog post on a given topic.",
)

trace.set_expected_outcome(
    task_objective="Produce a 600-word blog post with SEO metadata",
    required_deliverables=["Blog post", "SEO title", "Meta description"],
)
trace.set_input_summary(f"Topic: {topic}. Audience: {audience}.")

total_prompt = 0
total_completion = 0

# Step 1 -- auto-timed, exceptions auto-captured
with trace.step("llm_inference", "Create outline") as step:
    resp = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Create a blog post outline."},
            {"role": "user", "content": topic},
        ],
    )
    outline = resp.choices[0].message.content
    total_prompt += resp.usage.prompt_tokens
    total_completion += resp.usage.completion_tokens
    step.set_output(outline[:500])
    step.record_llm_call("gpt-4o", provider="openai",
                         prompt_tokens=resp.usage.prompt_tokens,
                         completion_tokens=resp.usage.completion_tokens)

# Step 2
with trace.step("llm_inference", "Write full blog post") as step:
    resp = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Write a complete blog post from this outline."},
            {"role": "user", "content": outline},
        ],
    )
    blog_post = resp.choices[0].message.content
    total_prompt += resp.usage.prompt_tokens
    total_completion += resp.usage.completion_tokens
    step.set_output(blog_post[:500])
    step.record_llm_call("gpt-4o", provider="openai",
                         prompt_tokens=resp.usage.prompt_tokens,
                         completion_tokens=resp.usage.completion_tokens)

trace.set_output_summary(f"Generated {len(blog_post.split())}-word blog post.")
trace.set_llm_usage(LlmUsage(
    model="gpt-4o", provider="openai",
    prompt_tokens=total_prompt, completion_tokens=total_completion,
    total_tokens=total_prompt + total_completion,
    api_calls_count=2,
))

task_data = trace.build()  # Auto-computes total duration from wall clock
client.send_task_data(task_data)

TraceBuilder Methods

Method	Description
`step(action, description)`	Create an auto-timed step (context manager)
`set_input_summary(summary)`	What the agent received
`set_output_summary(summary)`	What the agent produced
`set_expected_outcome(...)`	Maestro’s grading rubric
`set_environment(...)`	Runtime metadata
`set_llm_usage(llm_usage)`	Aggregate token stats
`set_section(name, value)`	Set a domain section
`add_error(error)`	Record an error
`build()`	Build the final `TaskData`

All setters return self for fluent chaining:

trace.set_input_summary("...").set_expected_outcome(...).set_environment(...)

StepContext Methods

Inside a with trace.step(...) as step: block:

Method	Description
`set_input(preview)`	Input preview (truncated to 500 chars)
`set_output(preview)`	Output preview (truncated to 500 chars)
`record_tool_call(tool_name, ...)`	Record a tool invocation
`record_llm_call(model, ...)`	Record an LLM call
`set_metadata(metadata)`	Attach arbitrary metadata

If an unhandled exception occurs inside a step, it is automatically captured as an ErrorDetail on the step.