ateam-llm-tracer 0.4.1
Lightweight tracing SDK for LLM-powered agents with Phoenix integration
License: MIT
Requires Python: >=3.11
Keywords: llm, tracing, observability, phoenix, openinference, agents

# 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

```bash
pip install ateam-llm-tracer
```

## Quick Start

```python
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:

```python
# 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:

```python
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:

```python
# 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

```python
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

```python
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:

```python
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:

```python
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:

```python
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:

```python
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