ateam-llm-tracer 0.4.0

Lightweight tracing SDK for LLM-powered agents with Phoenix integration

ateam-llm-tracer

Lightweight tracing SDK for LLM-powered agents. Instrument once, evaluate continuously.

Built on OpenInference and OpenTelemetry, ships to Phoenix out of the box.

Installation

pip install ateam-llm-tracer

Quick Start

from ateam_llm_tracer import init_tracing, Tracer

# Initialize once at application startup
init_tracing(
    project_name="my-agent-project",
    service_name="my-agent",
    phoenix_endpoint="https://phoenix.internal.a.team",
)

# Create a tracer scoped to your task
tracer = Tracer(task="nl-to-sql")

# Trace an LLM call
with tracer.start_llm_span("generate-query") as span:
    span.set_input(user_question)
    span.set_model("claude-sonnet-4-20250514")

    response = llm.complete(user_question)

    span.set_output(response.content)
    span.set_token_counts(
        prompt=response.usage.input_tokens,
        completion=response.usage.output_tokens
    )
    span.mark_success()

Features

• Zero-config start: Sensible defaults, single line to enable
• Minimal code changes: Decorators and context managers
• Phoenix-native: Built on OpenTelemetry + OpenInference
• Span kinds: LLM, Agent, Tool, Chain, Retriever, and more
• Status tracking: Success, failure, partial completion
• Signal mapping: Connect user feedback to traces
• Force flush: Ensure traces are sent in serverless/CLI environments
• Suppress tracing: Temporarily disable tracing for sensitive operations

Span Kinds

• LLM: Direct model inference
• AGENT: Agentic loops with iterations
• TOOL: Tool/function execution
• CHAIN: Multi-step workflows
• RETRIEVER: RAG retrieval
• EMBEDDING: Embedding generation
• RERANKER: Reranking operations
• GUARDRAIL: Safety/validation checks

Configuration

Configure via environment variables or code:

# Environment variables
export CONTROL_ROOM_TRACING_ENABLED=true
export CONTROL_ROOM_TRACING_ENDPOINT=https://phoenix.internal.a.team
export CONTROL_ROOM_TRACING_SERVICE=my-agent
export CONTROL_ROOM_TRACING_API_KEY=your-api-key-here  # Optional, for authentication
export CONTROL_ROOM_TRACING_SAMPLE_RATE=1.0

Or in code:

from ateam_llm_tracer import TracingConfig, init_tracing

config = TracingConfig(
    enabled=True,
    phoenix_endpoint="https://phoenix.internal.a.team",
    service_name="my-agent",
    phoenix_api_key="your-api-key-here",  # Optional
    sample_rate=1.0,
)

init_tracing(project_name="my-agent-project", config=config)

Authentication

For production Phoenix deployments that require authentication, provide an API key:

# Via environment variable (recommended)
export CONTROL_ROOM_TRACING_API_KEY=your-api-key-here

# Or via function parameter
init_tracing(
    project_name="my-agent-project",
    service_name="my-agent",
    phoenix_endpoint="https://phoenix.example.com",
    phoenix_api_key="your-api-key-here"
)

The API key is automatically sent as a Bearer token in the Authorization header.

Examples

Agentic Loop

tracer = Tracer(task="research-query")

with tracer.start_agent_span("research-loop") as agent_span:
    agent_span.set_input(query)

    for i in range(10):
        agent_span.set_iteration(i + 1)

        with tracer.start_llm_span("plan") as llm_span:
            llm_span.set_model("claude-sonnet-4")
            response = llm.complete(messages)
            llm_span.set_output(response)
            llm_span.mark_success()

        if response.stop_reason == "end_turn":
            agent_span.set_output(response.content)
            agent_span.mark_success()
            break

Tool Execution

with tracer.start_tool_span("database-query") as span:
    span.set_tool_name("execute_sql")
    span.set_tool_parameters({"query": sql_query})

    result = execute_sql(sql_query)

    span.set_tool_result(result)
    span.mark_success()

User Feedback

The library provides a flexible signal recording system that captures user interactions without imposing interpretation:

from ateam_llm_tracer import record_signal, SignalType

# Record user signals (default: no quality scores, just facts)
record_signal(
    span_id=artifact.span_id,
    signal=SignalType.THUMBS_UP,
    metadata={"user_id": user.id}
)

# Available signals: THUMBS_UP, THUMBS_DOWN, EDIT, COPY, SAVE,
# EXECUTE, REGENERATE, FLAG, ACCEPT, REJECT, and more

Custom Signal Mappings

For domain-specific signal interpretation, provide a custom mapping at initialization:

from ateam_llm_tracer import SignalType, init_tracing

# Define your domain-specific signal interpretation
custom_mapping = {
    SignalType.EXECUTE: ("executed", 1.0),  # High quality
    SignalType.EDIT: ("edited", 0.6),       # Partial quality
    SignalType.REGENERATE: ("regen", 0.2),  # Low quality
    # Map only the signals you care about
}

init_tracing(
    project_name="my-project",
    service_name="my-service",
    phoenix_endpoint="https://phoenix.internal.a.team",
    signal_mapping=custom_mapping  # Pass your custom mapping
)

See examples/custom_signal_handler.py for complete examples.

Advanced Features

Force Flush

In serverless environments or CLI tools, traces may not be sent before the application exits. Use force_flush() to ensure all pending traces are sent:

from ateam_llm_tracer import init_tracing, force_flush, Tracer

init_tracing(project_name="my-cli-tool", service_name="cli")

tracer = Tracer(task="process-file")
with tracer.start_agent_span("processing") as span:
    # ... do work ...
    span.mark_success()

# Ensure all traces are sent before exit
if force_flush(timeout_millis=5000):
    print("Traces sent successfully")

Suppress Tracing

Temporarily disable tracing for sensitive operations or performance-critical sections:

from ateam_llm_tracer import suppress_tracing, Tracer

tracer = Tracer(task="user-request")

with tracer.start_agent_span("handle-request") as span:
    span.set_input("Processing user data")

    # This section won't be traced
    with suppress_tracing():
        # Handle sensitive data without tracing
        process_credentials(user_credentials)
        internal_metrics.record()

    span.set_output("Request completed")
    span.mark_success()

See examples/force_flush_example.py for complete examples.

Requirements

• Python 3.11+
• Phoenix instance for trace collection

License

MIT