auditi 0.1.5

Trace, monitor, and evaluate AI agents and LLM applications with simple decorators

Auditi Python SDK

Open Source LLM Observability & Evaluation Platform

Documentation | GitHub

---

Trace, evaluate, and improve your AI agents with just 2 lines of code.

import auditi
auditi.init()
auditi.instrument()  # Auto-captures all LLM calls!

That's it. Every OpenAI, Anthropic, and Google Gemini call is now traced with costs, latency, and token usage.

---

Features

• Zero-Config Auto-Instrumentation - 2 lines to trace all LLM calls
• Simple Decorators - @trace_agent, @trace_tool, @trace_llm for custom control
• Multi-Provider - Auto-detects OpenAI, Anthropic, and Google Gemini
• Cost Tracking - Automatic token usage and cost calculation
• Async & Sync - Works with both synchronous and asynchronous functions
• Custom Evaluators - Implement your own evaluation logic
• Production Ready - FastAPI, LangChain, and framework integrations

---

Installation

pip install auditi

Quick Start

Option 1: Auto-Instrumentation (Recommended)

The fastest way to get started - just 2 lines of code:

import auditi
from openai import OpenAI

# Initialize and auto-instrument all LLM libraries
auditi.init(api_key="your-api-key", user_id="user123")
auditi.instrument()

# Your existing code works unchanged!
client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}]
)
# Trace automatically captured with cost, latency, tokens

Auto-instrumentation supports:

• OpenAI - client.chat.completions.create()
• Anthropic - client.messages.create()
• Google Gemini - model.generate_content()

Note: Auto-instrumentation captures LLM calls only as standalone traces. Tool calls, retrieval, and embeddings are not auto-instrumented. For full agent tracing with multiple steps (tools, retrieval, etc.), use decorators as shown below. You can also combine both: use auditi.instrument() alongside @trace_agent — auto-instrumented LLM calls will automatically become nested spans within the agent trace.

Option 2: Decorator-Based (Fine-Grained Control)

For complex agents with multiple steps:

import auditi
from auditi import trace_agent, trace_tool, trace_llm

auditi.init(api_key="your-api-key")

@trace_agent(name="customer_support")
def support_agent(message: str):
    context = get_context(message)
    response = generate_response(message, context)
    return response

@trace_tool(name="get_context")
def get_context(query: str):
    return db.search(query)

@trace_llm(model="gpt-4o")
def generate_response(message: str, context: dict):
    return openai.chat.completions.create(...)

Option 3: Self-Hosted (Development)

Run your own Auditi backend for development:

# Clone and run with Docker
git clone https://github.com/deduu/auditi
cd auditi
docker-compose up

import auditi

# Point to your local instance
auditi.init(base_url="http://localhost:8000")
auditi.instrument()

---

Usage Patterns

Pattern 1: Simple LLM Calls (Auto-Instrumented)

With auto-instrumentation enabled, all LLM calls are traced automatically:

import auditi
from openai import OpenAI

auditi.init()
auditi.instrument()

client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "What is Python?"}]
)
# Trace created automatically!

Pattern 2: Complex Agents with Decorators

For multi-step agentic workflows, use decorators:

from auditi import trace_agent, trace_tool, trace_llm
import openai

@trace_agent(name="customer_support")
def customer_support_agent(user_message: str, user_id: str = None):
    """Your existing agent - just add the decorator!"""

    # Fetch user context
    context = get_user_context(user_id)

    # Search knowledge base
    docs = search_knowledge_base(user_message)

    # Generate response
    response = call_openai(user_message, context, docs)

    return response


@trace_tool(name="search_kb")
def search_knowledge_base(query: str):
    """Tool calls are automatically captured as spans."""
    results = vector_db.similarity_search(query, k=5)
    return results


@trace_llm(model="gpt-4o")
def call_openai(message: str, context: dict, docs: list):
    """LLM calls capture usage metrics and costs."""
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"Context: {context}"},
            {"role": "user", "content": message}
        ]
    )
    return response.choices[0].message.content

3. View Traces in Auditi Dashboard

That's it! Every call to customer_support_agent() will:

• ✅ Capture user input and assistant output
• ✅ Track all tool calls and LLM calls as spans
• ✅ Calculate token usage and costs
• ✅ Send to Auditi for evaluation and monitoring

Usage Patterns

Pattern 1: Simple LLM Calls (Standalone)

For simple chatbots or single LLM calls, you don't need @trace_agent:

@trace_llm(standalone=True)
def simple_chat(prompt: str):
    """Creates its own trace automatically - no agent wrapper needed!"""
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# This creates a complete trace automatically
result = simple_chat("What is the capital of France?")

Pattern 2: Complex Agents with Tools

For multi-step agentic workflows:

@trace_agent(name="research_assistant")
def research_assistant(query: str, user_id: str):
    """Main agent creates ONE trace that captures all spans."""

    # Step 1: Web search (creates a tool span)
    search_results = web_search(query)

    # Step 2: Generate initial response (creates an LLM span)
    initial_response = generate_response(query, search_results)

    # Step 3: Reflect on quality (creates another LLM span)
    quality_score = evaluate_response(initial_response)

    # Step 4: Refine if needed (creates another LLM span)
    if quality_score < 0.7:
        final_response = refine_response(initial_response, search_results)
    else:
        final_response = initial_response

    return final_response


@trace_tool("web_search")
def web_search(query: str):
    # Search implementation
    return results


@trace_llm(model="gpt-4o")
def generate_response(query: str, context: list):
    # LLM call
    return response


@trace_llm(model="gpt-4o")
def evaluate_response(text: str):
    # Another LLM call for reflection
    return score

Pattern 3: Embeddings and Retrieval

Embedding and retrieval operations are always standalone by default:

@trace_embedding()
def embed_text(text: str):
    """Creates a standalone trace for embedding."""
    response = openai.embeddings.create(
        input=text,
        model="text-embedding-3-small"
    )
    return response.data[0].embedding


@trace_retrieval("vector_search")
def search_docs(query: str):
    """Creates a standalone trace for retrieval."""
    embedding = embed_text(query)
    results = vector_db.similarity_search(embedding, k=5)
    return results

Pattern 4: RAG Pipeline

Combining all patterns in a full RAG workflow:

@trace_agent(name="rag_assistant")
def rag_query(question: str):
    """Full RAG pipeline - all steps captured as spans."""

    # Embedding step (creates span)
    query_embedding = embed_query(question)

    # Retrieval step (creates span)
    docs = retrieve_docs(query_embedding)

    # LLM step (creates span)
    answer = generate_answer(question, docs)

    return answer


@trace_embedding()
def embed_query(text: str):
    # Embedding logic
    return embedding


@trace_retrieval("doc_search")
def retrieve_docs(embedding: list):
    # Vector search logic
    return documents


@trace_llm(model="gpt-4o")
def generate_answer(question: str, context: list):
    # LLM generation
    return answer

Integration Examples

FastAPI Integration

from fastapi import FastAPI
from auditi import trace_agent, trace_tool, trace_llm

app = FastAPI()

@app.post("/chat")
async def chat_endpoint(message: str, user_id: str):
    response = await process_chat(message, user_id)
    return {"response": response}


@trace_agent(name="chat_agent")
async def process_chat(message: str, user_id: str):
    """Async agent - fully supported!"""
    context = await fetch_user_context(user_id)
    kb_results = await search_knowledge_base(message)
    response = await call_llm(message, context, kb_results)
    return response


@trace_tool("fetch_context")
async def fetch_user_context(user_id: str):
    # Async tool call
    return context


@trace_llm(model="gpt-4o")
async def call_llm(message: str, context: dict, docs: list):
    # Async LLM call
    return response

LangChain Integration

from langchain.agents import AgentExecutor, create_openai_functions_agent
from auditi import trace_agent, trace_tool

@trace_agent(name="langchain_agent")
def run_langchain_agent(query: str):
    """Wrap your LangChain execution."""
    agent_executor = create_agent()
    result = agent_executor.invoke({"input": query})
    return result["output"]


@trace_tool("vector_search")
def vector_search(query: str):
    """Individual tools can be traced too."""
    return vectorstore.similarity_search(query)

Custom Evaluators

Implement custom evaluation logic to assess trace quality:

from auditi import BaseEvaluator, EvaluationResult, TraceInput

class ResponseQualityEvaluator(BaseEvaluator):
    def evaluate(self, trace: TraceInput) -> EvaluationResult:
        """Evaluate response quality based on custom criteria."""

        # Access trace data
        user_input = trace.user_input
        assistant_output = trace.assistant_output
        spans = trace.spans

        # Your evaluation logic
        score = self._calculate_quality_score(assistant_output)

        # Return evaluation result
        if score >= 0.8:
            status = "pass"
            reason = "High quality response"
        elif score >= 0.6:
            status = "pass"
            reason = "Acceptable quality"
        else:
            status = "fail"
            reason = "Low quality response - needs improvement"

        return EvaluationResult(
            status=status,
            score=score,
            reason=reason
        )

    def _calculate_quality_score(self, text: str) -> float:
        # Your scoring logic here
        return 0.85


# Use with trace_agent
@trace_agent(name="assistant", evaluator=ResponseQualityEvaluator())
def my_agent(message: str):
    response = generate_response(message)
    return response

Multi-Provider Support

Auditi automatically detects and handles multiple LLM providers:

# OpenAI
@trace_llm(model="gpt-4o")
def call_openai(prompt: str):
    response = openai.chat.completions.create(...)
    return response  # Auto-extracts usage from response.usage


# Anthropic Claude
@trace_llm(model="claude-sonnet-4-5-20250929")
def call_anthropic(prompt: str):
    response = anthropic.messages.create(...)
    return response  # Auto-extracts from response.usage


# Google Gemini
@trace_llm(model="gemini-2.0-flash-exp")
def call_google(prompt: str):
    response = genai.generate_content(...)
    return response  # Auto-extracts from response.usage_metadata

Supported Providers:

• ✅ OpenAI (GPT-4, GPT-4o, GPT-3.5, etc.)
• ✅ Anthropic (Claude 3.5, Claude 3, Claude 2)
• ✅ Google (Gemini Pro, Gemini Flash)
• ✅ Auto-detection from model names and response structures
• ✅ Automatic cost calculation with up-to-date pricing

Configuration

Environment Variables

# Enable debug logging
export AUDITI_DEBUG=true

# Set API key
export AUDITI_API_KEY=your-api-key

# Set base URL
export AUDITI_BASE_URL=http://localhost:8000

Programmatic Configuration

import auditi

# Production setup
auditi.init(
    api_key="your-api-key",
    base_url="http://localhost:8000"
)

# With user tracking (applied to all traces including auto-instrumented)
auditi.init(
    api_key="your-api-key",
    base_url="http://localhost:8000",
    user_id="user123",
    session_id="conv456"
)

# Or set user context dynamically (useful when user_id changes per request)
auditi.init(api_key="your-api-key")
auditi.set_context(user_id="user123", session_id="conv456")

# Development setup (prints traces to console)
from auditi.transport import DebugTransport

auditi.init(
    transport=DebugTransport()  # Prints to console instead of sending
)

FastAPI Middleware

For production backends serving many users, use AuditiMiddleware to automatically extract user_id from each request — no need to call set_context() in every endpoint:

from fastapi import FastAPI
import auditi
from auditi import AuditiMiddleware

app = FastAPI()
auditi.init(api_key="your-api-key")
auditi.instrument()

# Default: reads X-User-ID and X-Session-ID headers
app.add_middleware(AuditiMiddleware)

Custom header names:

app.add_middleware(
    AuditiMiddleware,
    user_id_header="X-Tenant-ID",
    session_id_header="X-Conversation-ID",
)

Custom resolver (e.g. JWT decoding):

def resolve_user(scope):
    headers = dict(scope.get("headers", []))
    token = headers.get(b"authorization", b"").decode().removeprefix("Bearer ")
    claims = decode_jwt(token)
    return claims.get("sub"), claims.get("session_id")

app.add_middleware(AuditiMiddleware, user_id_resolver=resolve_user)

Model Pricing

Auditi automatically calculates costs for every LLM call. Pricing is resolved in priority order:

1. User overrides (highest priority)
2. Remote pricing from your Auditi backend API
3. Hardcoded defaults in the SDK (fallback/offline)

Auto-Updated Pricing

When you call auditi.init(), the SDK connects to your backend to fetch the latest model pricing. This means you never need to update the SDK just because a provider changed their prices — update the backend, and all SDK users get the new pricing automatically.

import auditi

# SDK fetches latest pricing from your backend automatically
auditi.init(api_key="your-key", base_url="https://your-auditi-instance.com")

Pricing is cached for 1 hour and falls back to hardcoded defaults if the backend is unreachable.

Custom Pricing Overrides

Override pricing for specific models if you have negotiated rates or want to track custom costs:

from auditi import configure_pricing

# Override specific model pricing (per 1M tokens)
configure_pricing(pricing={
    "openai": {
        "gpt-4o": (2.00, 8.00),           # Custom negotiated rate
        "gpt-4o-mini": (0.10, 0.40),
    },
    "anthropic": {
        "claude-3-5-sonnet-20241022": (2.50, 12.00),
    }
})

Disabling Remote Pricing

For offline or air-gapped environments, disable remote fetching entirely:

from auditi import configure_pricing

# Use only hardcoded defaults
configure_pricing(enable_remote=False)

Updating Pricing (Backend Admin)

If you self-host Auditi, default pricing is seeded on first startup. You can update pricing at any time via the REST API:

# View all current pricing
curl https://your-auditi-instance.com/api/v1/pricing

# Update a single model's pricing
curl -X POST https://your-auditi-instance.com/api/v1/pricing \
  -H "Content-Type: application/json" \
  -d '{"provider": "openai", "model": "gpt-4o", "input_price": 2.50, "output_price": 10.00}'

# Bulk update multiple models
curl -X POST https://your-auditi-instance.com/api/v1/pricing/bulk \
  -H "Content-Type: application/json" \
  -d '{
    "items": [
      {"provider": "openai", "model": "gpt-4o", "input_price": 2.50, "output_price": 10.00},
      {"provider": "openai", "model": "gpt-4o-mini", "input_price": 0.15, "output_price": 0.60}
    ]
  }'

---

API Reference

auditi.init(api_key=None, base_url="http://localhost:8000", transport=None, user_id=None, session_id=None)

Initialize the Auditi SDK.

Parameters:

• api_key (str, optional): API key for authentication
• base_url (str): Base URL of the Auditi API (default: http://localhost:8000)
• transport (BaseTransport, optional): Custom transport implementation
• user_id (str, optional): User identifier to set in global context (applies to all traces)
• session_id (str, optional): Session/conversation identifier to set in global context

auditi.set_context(user_id=None, session_id=None)

Set global context for the current execution context. Useful for dynamically changing user_id per request.

AuditiMiddleware(app, user_id_header="X-User-ID", session_id_header="X-Session-ID", user_id_resolver=None)

ASGI middleware for FastAPI/Starlette. Extracts user_id and session_id from each request and sets them via set_context().

Parameters:

• user_id_header (str): Header name to read user_id from (default: X-User-ID)
• session_id_header (str): Header name to read session_id from (default: X-Session-ID)
• user_id_resolver (callable, optional): Custom function (scope) -> (user_id, session_id). When provided, headers are ignored.

Decorators

@trace_agent(name=None, user_id=None, evaluator=None)

Trace a top-level agent function. Creates a complete trace with user input, assistant output, and all spans.

Parameters:

• name (str, optional): Custom name for the agent
• user_id (str, optional): User identifier
• evaluator (BaseEvaluator, optional): Custom evaluator instance

Returns: The decorated function's return value becomes assistant_output

@trace_tool(name=None, standalone=False)

Trace a tool/function call within an agent.

Parameters:

• name (str, optional): Custom name for the tool
• standalone (bool): If True, creates a standalone trace when not inside @trace_agent

@trace_llm(name=None, model=None, standalone=False)

Trace an LLM call within an agent.

Parameters:

• name (str, optional): Custom name for the LLM call
• model (str, optional): Model name (auto-detected from response if not provided)
• standalone (bool): If True, creates a standalone trace when not inside @trace_agent

@trace_embedding(name=None, model=None)

Trace an embedding operation. Always creates a standalone trace when not inside @trace_agent.

Parameters:

• name (str, optional): Custom name for the embedding operation
• model (str, optional): Model name (auto-detected if not provided)

@trace_retrieval(name=None)

Trace a retrieval/search operation. Always creates a standalone trace when not inside @trace_agent.

Parameters:

• name (str, optional): Custom name for the retrieval operation

Types

TraceInput

Complete trace data model.

Fields:

• trace_id (str): Unique trace identifier
• name (str): Agent name
• user_input (str): User's message
• assistant_output (str): Agent's response
• user_id (str, optional): User identifier
• conversation_id (str, optional): Conversation/session identifier
• spans (List[SpanInput]): List of spans (tools, LLM calls)
• input_tokens (int): Total input tokens
• output_tokens (int): Total output tokens
• total_tokens (int): Total tokens
• cost (float): Total cost in USD
• processing_time (float): Total processing time in seconds
• metadata (dict, optional): Additional metadata
• timestamp (datetime): Trace timestamp

SpanInput

Individual span within a trace.

Fields:

• span_id (str): Unique span identifier
• name (str): Span name
• span_type (str): Type: "tool", "llm", "embedding", "retrieval"
• inputs (dict): Input parameters
• outputs (Any): Output value
• input_tokens (int, optional): Input tokens (for LLM spans)
• output_tokens (int, optional): Output tokens (for LLM spans)
• total_tokens (int, optional): Total tokens
• cost (float, optional): Cost in USD
• model (str, optional): Model name
• processing_time (float, optional): Processing time in seconds
• timestamp (datetime): Span timestamp

EvaluationResult

Evaluation result data.

Fields:

• status (str): "pass" or "fail"
• score (float, optional): Evaluation score
• reason (str, optional): Explanation
• metadata (dict, optional): Additional evaluation data

Transport

SyncHttpTransport(api_key, base_url)

Default synchronous HTTP transport.

Parameters:

• api_key (str): API key for authentication
• base_url (str): Base URL of Auditi API

DebugTransport()

Debug transport that prints traces to console. Useful for local development.

Context Management

from auditi.context import (
    get_current_trace,
    set_current_trace,
    get_current_span,
    set_context,
    get_context
)

# Get current trace (if inside @trace_agent)
trace = get_current_trace()

# Get current span (if inside @trace_tool/@trace_llm)
span = get_current_span()

# Set global context (available across all traces)
set_context({"environment": "production", "version": "1.0"})

Development

Setup

# Clone the repository
git clone https://github.com/deduu/auditi
cd auditi

# Install dev dependencies
pip install -e ".[dev]"

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=auditi --cov-report=html

# Run specific test file
pytest tests/test_decorators.py -v

# Run async tests
pytest tests/test_decorators.py -k async

Code Quality

# Format code
black auditi/

# Lint code
ruff auditi/

# Type check
mypy auditi/

Project Structure

auditi/
├── __init__.py           # Package initialization
├── client.py             # SDK client and initialization
├── context.py            # Context management for traces/spans
├── decorators.py         # Core decorators (@trace_agent, etc.)
├── evaluator.py          # Base evaluator class
├── events.py             # Event types for streaming
├── transport.py          # Transport layer (HTTP, Debug)
├── providers/            # LLM provider abstractions
│   ├── __init__.py
│   ├── base.py          # Base provider interface
│   ├── openai.py        # OpenAI provider
│   ├── anthropic.py     # Anthropic provider
│   ├── google.py        # Google provider
│   └── registry.py      # Provider auto-detection
└── types/
    ├── __init__.py
    └── api_types.py     # Pydantic models for API types

Examples

The examples/ directory contains complete working examples:

• 01_basic_integration.py - Simple chatbot integration
• 02_fastapi_integration.py - Production FastAPI integration
• 03_langchain_integration.py - LangChain agent integration
• 04_simple_llm_traces.py - Standalone LLM call tracing
• 05_embedding_traces.py - Embedding and retrieval tracing

Run any example:

# Enable debug output
export AUDITI_DEBUG=true

# Run example
python examples/01_basic_integration.py

Troubleshooting

Traces Not Appearing

1. Check initialization:

   import auditi
   auditi.init(api_key="your-key", base_url="http://localhost:8000")
   

2. Enable debug logging:

   export AUDITI_DEBUG=true
   python your_script.py
   

3. Verify decorator order:

   • @trace_agent should be the outermost decorator
   • @trace_tool and @trace_llm should be inside functions called by the agent

Missing Usage Metrics

Make sure your LLM call returns the full response object:

@trace_llm(model="gpt-4o")
def call_openai(prompt: str):
    response = openai.chat.completions.create(...)
    return response  # ✅ Return full response, not just .choices[0].message.content

Async Functions Not Working

Both sync and async functions are supported. Make sure to use await:

@trace_agent(name="async_agent")
async def my_agent(message: str):
    result = await async_llm_call(message)
    return result

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

License

MIT License - see LICENSE file for details.

Links

• GitHub: https://github.com/deduu/auditi