llmops-observability 37.2.0
LLMOps Observability SDK: decorators + SQS dispatch with compression
Requires Python: >=3.9

# LLMOps Observability SDK

A production-grade Python SDK for LLM observability with SQS-based event streaming for decoupled, scalable observability pipelines. Automatically captures traces, spans, token usage, costs, and metadata from your LLM applications.

## 🎯 Key Features

- ⚡ **SQS Event Streaming**: Batch events to AWS SQS with automatic spillover recovery
- 🎨 **Simple Decorators**: `@track_function` and `@track_llm_call` for instant instrumentation
- 🔄 **Sync & Async Support**: Works with both synchronous and asynchronous functions
- 🤖 **Provider Agnostic**: Compatible with any LLM provider (AWS Bedrock, OpenAI, Anthropic, etc.)
- 🪆 **Hierarchical Tracing**: Automatic parent-child span relationships with proper nesting
- 💰 **Cost Tracking**: Built-in token usage and cost calculation for AWS Bedrock models
- 🔍 **Smart Capture**: Optionally capture function locals and self for detailed debugging
- 📊 **Size Management**: Automatic truncation and compression (200KB limits) to prevent data issues
- 🛡️ **Production-Ready**: Daemon workers, batch processing, clean shutdown handling
- 🌍 **Auto-Injection**: Environment and project_id automatically added to every span

## 📦 Installation

```bash
# From source (development)
cd llmops-observability_sdk
pip install -e .

# Or with uv
uv sync
```

## ⚙️ Configuration

### Environment Variables

Create a `.env` file in your application directory:

```bash
# Project Configuration (Required - Auto-injected into every span)
PROJECT_ID=my_project          # Your project identifier
ENV=uat                        # Environment: development, staging, uat, production

# AWS SQS Configuration (Required for trace streaming)
AWS_SQS_URL=https://sqs.us-east-1.amazonaws.com/123456789/my-queue
AWS_PROFILE=default            # AWS profile name
AWS_REGION=us-east-1           # AWS region

# Model Configuration (Optional)
MODEL_ID=anthropic.claude-3-5-sonnet-20241022-v2:0  # Default model for cost calculation
```

**Key Configuration Values:**
- **`PROJECT_ID`**: Unique identifier for your project. Auto-injected into every span's metadata.
- **`ENV`**: Environment name (development/staging/uat/production). Auto-injected into every span's metadata.
- **`AWS_SQS_URL`**: SQS queue URL for sending traces to Lambda processor.

> 💡 **Important**: `PROJECT_ID` and `ENV` are automatically injected into every span's metadata by the SDK. You don't need to manually add them to decorators.

## 🚀 Quick Start

### 1. Basic Usage with Auto-Configuration
```python
from llmops_observability import TraceManager, track_function, track_llm_call

# Start a trace - PROJECT_ID and ENV are auto-loaded from config.py
TraceManager.start_trace(
    name="rag_pipeline_example",
    user_id="user_123",
    session_id="session_456",
    metadata={"version": "1.0.0"},
    tags=["example", "rag"]
)

# Track regular functions
@track_function()
def process_input(user_query: str):
    return {"query": user_query, "processed": True}

# Track LLM calls with automatic cost calculation
@track_llm_call(model="anthropic.claude-3-sonnet-20240229-v1:0")
def call_llm(prompt: str):
    response = bedrock_client.invoke_model(...)  # Your LLM call
    return response

# Execute your workflow
result = process_input("What is Python?")
answer = call_llm("Context: ...\n\nQuestion: What is Python?")

# Finalize and send to SQS (optional parameters)
TraceManager.finalize_and_send(
    trace_name="rag_pipeline_example",
    trace_input={"user_msg": "What is Python?"},
    trace_output={"bot_response": answer}
)
```

### 2. Automatic Environment & Project Injection

**Every span automatically gets `environment` and `project_id` in metadata:**

```python
# Set environment variables
os.environ["PROJECT_ID"] = "new_test"
os.environ["ENV"] = "uat"

# Start trace
TraceManager.start_trace("my_operation")

# Every @track_function and @track_llm_call span will automatically have:
# span.metadata = {
#     "environment": "uat",
#     "project_id": "new_test",
#     # ... other metadata ...
# }
```

✅ **No manual injection needed!** The SDK automatically adds these to every span.

### 3. Nested Spans (Parent-Child Relationships)

```python
@track_function()
def parent_function():
    # This creates a parent span
    child_result = child_function()
    return child_result

@track_function()
def child_function():
    # This automatically becomes a child of parent_function
    grandchild_result = grandchild_function()
    return grandchild_result

@track_function()
def grandchild_function():
    # This becomes a child of child_function
    return "result"

# Proper hierarchy maintained in Langfuse/NewRelic/S3
```

## 📊 Data Flow Architecture

```
┌─────────────────────┐
│   Your LLM App      │
│  (with decorators)  │
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│  TraceManager       │
│  (collects spans)   │
│  + Auto-injects:    │
│    - environment    │
│    - project_id     │
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│  SQS Batch Workers  │
│  (compress & send)  │
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│  AWS SQS Queue      │
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│  Lambda Processor   │
│  (parallel routing) │
└─────┬──────┬────┬───┘
      │      │    │
      ▼      ▼    ▼
  Langfuse  S3  NewRelic
```

## 🎨 Decorator Reference

### @track_function - Complete Guide

#### Basic Usage (No Parameters)

```python
@track_function()
def process_data(input_data):
    # Automatically captures:
    # - Function name as span name
    # - Function arguments (args, kwargs)
    # - Return value
    # - Execution time
    # - Environment and project_id (auto-injected)
    return {"processed": input_data}
```

#### Parameter: `name` (Custom Span Name)

```python
@track_function(name="custom_span_name")
def my_function():
    # Span will appear as "custom_span_name" instead of "my_function"
    return result

# Use case: Make span names more descriptive in traces
@track_function(name="fetch_user_profile_from_db")
def get_user(user_id):
    return db.query(user_id)
```

#### Parameter: `metadata` (Add Custom Metadata)

```python
@track_function(metadata={"service": "auth", "priority": "high"})
def authenticate_user(username, password):
    # Span metadata will include:
    # {
    #     "service": "auth",
    #     "priority": "high",
    #     "environment": "uat",      # auto-injected
    #     "project_id": "new_test"   # auto-injected
    # }
    return auth_result

# Use case: Tag spans with business context
@track_function(metadata={
    "database": "postgres",
    "table": "users",
    "operation": "read"
})
def query_users(filters):
    return db.execute(query)
```

#### Parameter: `capture_locals=True` (Capture All Local Variables)

```python
@track_function(capture_locals=True)
def process_payment(amount, currency):
    user_id = "user_123"
    transaction_id = generate_id()
    tax = amount * 0.1
    total = amount + tax
    
    # All local variables captured in span.input_data.locals:
    # {
    #     "user_id": "user_123",
    #     "transaction_id": "txn_abc",
    #     "tax": 10.0,
    #     "total": 110.0,
    #     "amount": 100.0,
    #     "currency": "USD"
    # }
    
    return {"total": total}

# ⚠️ Warning: Can capture large amounts of data. Use for debugging only.
```

#### Parameter: `capture_locals=["var1", "var2"]` (Capture Specific Variables)

```python
@track_function(capture_locals=["user_id", "total"])
def process_payment(amount, currency):
    user_id = "user_123"
    transaction_id = generate_id()
    tax = amount * 0.1
    total = amount + tax
    
    # Only specified variables captured in span.input_data.locals:
    # {
    #     "user_id": "user_123",
    #     "total": 110.0
    # }
    # Note: transaction_id, tax, amount, currency are NOT captured
    
    return {"total": total}

# ✅ Recommended: Capture only what you need for debugging
```

#### Parameter: `capture_self=True` (Capture `self` in Class Methods)

```python
class PaymentProcessor:
    def __init__(self, merchant_id):
        self.merchant_id = merchant_id
        self.fee_rate = 0.029
    
    @track_function(capture_self=True)
    def process(self, amount):
        # Captures self attributes in span.input_data.self:
        # {
        #     "merchant_id": "merch_123",
        #     "fee_rate": 0.029
        # }
        fee = amount * self.fee_rate
        return amount - fee

# Use case: Debug class state during execution
class DatabaseConnection:
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self.connected = False
    
    @track_function(capture_self=True, capture_locals=["query"])
    def execute(self, query):
        # Captures both self and specific locals
        result = self._run_query(query)
        return result
```

#### Combined Parameters Example

```python
@track_function(
    name="complex_data_pipeline",
    metadata={"stage": "preprocessing", "version": "2.0"},
    capture_locals=["processed_count", "errors"],
    capture_self=False
)
def pipeline_stage(data):
    processed_count = 0
    errors = []
    temp_cache = {}  # Not captured
    
    for item in data:
        try:
            process_item(item)
            processed_count += 1
        except Exception as e:
            errors.append(str(e))
    
    return {"count": processed_count, "errors": errors}
```

---

### @track_llm_call - Complete Guide

#### Basic Usage (No Parameters)

```python
@track_llm_call()
def call_llm(prompt):
    # Automatically captures:
    # - Function arguments (prompt)
    # - LLM response
    # - Execution time
    # - Span type = "generation"
    # - Environment and project_id (auto-injected)
    response = bedrock_client.invoke_model(...)
    return response
```

#### Parameter: `name` (Custom Span Name)

```python
@track_llm_call(name="bedrock_claude_sonnet")
def call_llm(prompt):
    # Span appears as "bedrock_claude_sonnet" instead of "call_llm"
    response = bedrock_client.invoke_model(...)
    return response

# Use case: Distinguish between different LLM providers/models
@track_llm_call(name="openai_gpt4_turbo")
def call_openai(prompt):
    return openai.chat.completions.create(...)

@track_llm_call(name="anthropic_claude_opus")
def call_anthropic(prompt):
    return anthropic.messages.create(...)
```

#### Parameter: `model` (For Cost Calculation)

```python
@track_llm_call(model="anthropic.claude-3-sonnet-20240229-v1:0")
def call_bedrock(prompt):
    # SDK automatically calculates cost based on:
    # - Token usage from response
    # - Model pricing for Claude 3 Sonnet
    # 
    # Captured in span:
    # - usage.input_tokens
    # - usage.output_tokens
    # - cost.input_cost
    # - cost.output_cost
    # - cost.total_cost
    
    response = bedrock_client.invoke_model(
        modelId="anthropic.claude-3-sonnet-20240229-v1:0",
        body=json.dumps({"prompt": prompt})
    )
    return response

# Supported AWS Bedrock models (see pricing.py):
# - anthropic.claude-3-5-sonnet-20241022-v2:0
# - anthropic.claude-3-sonnet-20240229-v1:0
# - anthropic.claude-3-haiku-20240307-v1:0
# - anthropic.claude-3-opus-20240229-v1:0
# - And more...
```

#### Parameter: `metadata` (Add Custom Metadata)

```python
@track_llm_call(
    model="anthropic.claude-3-sonnet-20240229-v1:0",
    metadata={
        "temperature": 0.7,
        "max_tokens": 1000,
        "use_case": "code_generation"
    }
)
def generate_code(prompt):
    # Span metadata includes:
    # {
    #     "temperature": 0.7,
    #     "max_tokens": 1000,
    #     "use_case": "code_generation",
    #     "environment": "uat",      # auto-injected
    #     "project_id": "new_test"   # auto-injected
    # }
    return llm_response
```

#### Parameter: `capture_locals=True` (Capture All Locals)

```python
@track_llm_call(
    model="anthropic.claude-3-sonnet-20240229-v1:0",
    capture_locals=True
)
def enhanced_llm_call(user_query, context_docs):
    # Build prompt with context
    formatted_context = format_documents(context_docs)
    system_prompt = "You are a helpful assistant."
    final_prompt = f"{system_prompt}\n\nContext: {formatted_context}\n\nQuestion: {user_query}"
    
    # All locals captured:
    # {
    #     "user_query": "What is Python?",
    #     "context_docs": [...],
    #     "formatted_context": "...",
    #     "system_prompt": "You are a helpful assistant.",
    #     "final_prompt": "..."
    # }
    
    response = bedrock_client.invoke_model(...)
    return response
```

#### Parameter: `capture_locals=["prompt", "temperature"]` (Specific Variables)

```python
@track_llm_call(
    model="anthropic.claude-3-sonnet-20240229-v1:0",
    capture_locals=["final_prompt", "model_config"]
)
def call_with_config(user_input):
    model_config = {"temperature": 0.7, "max_tokens": 1000}
    system_message = "You are an AI assistant."  # NOT captured
    final_prompt = f"{system_message}\n\n{user_input}"
    temp_cache = {}  # NOT captured
    
    # Only captures:
    # {
    #     "final_prompt": "...",
    #     "model_config": {"temperature": 0.7, "max_tokens": 1000}
    # }
    
    response = call_llm(final_prompt, **model_config)
    return response
```

#### Parameter: `capture_self=True` (For Class Methods)

```python
class LLMOrchestrator:
    def __init__(self, model_id, api_key):
        self.model_id = model_id
        self.api_key = api_key
        self.request_count = 0
        self.total_cost = 0.0
    
    @track_llm_call(
        model="anthropic.claude-3-sonnet-20240229-v1:0",
        capture_self=True
    )
    def call_llm(self, prompt):
        # Captures self attributes:
        # {
        #     "model_id": "anthropic.claude-3-sonnet-20240229-v1:0",
        #     "request_count": 5,
        #     "total_cost": 0.042
        # }
        # Note: api_key might be captured - be careful with secrets!
        
        self.request_count += 1
        response = self._invoke_model(prompt)
        return response
```

#### Combined Parameters Example (Production Pattern)

```python
class ChatbotService:
    def __init__(self, model_id):
        self.model_id = model_id
        self.system_prompt = "You are a helpful chatbot."
    
    @track_llm_call(
        name="chatbot_generation",
        model="anthropic.claude-3-sonnet-20240229-v1:0",
        metadata={
            "service": "customer_support",
            "model_version": "v2.0",
            "priority": "high"
        },
        capture_locals=["full_prompt", "temperature"],
        capture_self=False  # Don't capture self to avoid secrets
    )
    def generate_response(self, user_message, conversation_history):
        temperature = 0.7
        full_prompt = self._build_prompt(user_message, conversation_history)
        cache_key = hash(full_prompt)  # Not captured
        
        response = bedrock_client.invoke_model(
            modelId=self.model_id,
            body=json.dumps({
                "prompt": full_prompt,
                "temperature": temperature,
                "max_tokens": 1000
            })
        )
        return response
```

---

### Quick Reference Table

| Parameter | @track_function | @track_llm_call | Type | Description |
|-----------|----------------|-----------------|------|-------------|
| `name` | ✅ | ✅ | `str` | Custom span name |
| `metadata` | ✅ | ✅ | `Dict[str, Any]` | Additional metadata |
| `capture_locals` | ✅ | ✅ | `bool` or `List[str]` | Capture local variables |
| `capture_self` | ✅ | ✅ | `bool` | Capture `self` in methods |
| `model` | ❌ | ✅ | `str` | Model ID for cost calculation |

---

### Best Practices

#### ✅ DO

```python
# Capture specific variables for debugging
@track_function(capture_locals=["error_code", "retry_count"])

# Use metadata for business context
@track_function(metadata={"team": "payments", "priority": "critical"})

# Specify model for accurate cost tracking
@track_llm_call(model="anthropic.claude-3-sonnet-20240229-v1:0")

# Use descriptive names
@track_llm_call(name="rag_retrieval_claude")
```

#### ❌ DON'T

```python
# Don't capture all locals in production (too much data)
@track_function(capture_locals=True)  # Only for debugging!

# Don't capture self if it contains secrets
@track_function(capture_self=True)  # Check for API keys first!

# Don't use generic names
@track_function(name="function_1")  # Not helpful

# Don't forget model for LLM calls
@track_llm_call()  # Missing model = no cost calculation
```

---

## 📈 What Gets Captured

### Trace-Level Data
- `trace_id`, `trace_name`, `project_id`, `environment`
- `user_id`, `session_id`
- `start_time`, `end_time`, `duration_ms`
- `trace_input`, `trace_output`
- `metadata`, `tags`
- `total_spans`, `total_generations`
- `sdk_name`, `sdk_version`

### Span-Level Data (Auto-captured for every span)
- **Core**: `span_id`, `span_name`, `span_type`, `parent_span_id`
- **Timing**: `start_time`, `end_time`, `duration_ms`
- **I/O**: `input_data`, `output_data`
- **Status**: `status`, `status_message`, `error`
- **LLM**: `model_id`, `prompt`, `response`
- **Usage**: `usage.input_tokens`, `usage.output_tokens`, `usage.total_tokens`
- **Cost**: Calculated from model pricing
- **Metadata**: `environment`, `project_id` (auto-injected), custom metadata
- **Context**: `tags`, `level`

## 🔧 Configuration Reference

### Size Limits (in config.py)

```python
MAX_OUTPUT_SIZE = 200 * 1024      # 200 KB - max individual field
MAX_SPAN_IO_SIZE = 20_000          # 20 KB - span input/output
MAX_TRACE_IO_SIZE = 50_000         # 50 KB - trace input/output
MAX_SQS_SIZE = 200_000             # 200 KB - SQS message
PROMPT_RESPONSE_MAX_SIZE = 10_000  # 10 KB - prompt/response fields
```

### SQS Configuration

```python
SQS_WORKER_COUNT = 4           # Background worker threads
SQS_BATCH_SIZE = 10            # Batch size before flush
SQS_BATCH_TIMEOUT = 0.2        # Timeout in seconds
SQS_FLUSH_TIME_THRESHOLD = 0.15
SQS_SHUTDOWN_TIMEOUT = 1.0
```

## 🏭 Production Best Practices

### 1. Proper Trace Lifecycle

```python
try:
    # Start trace
    TraceManager.start_trace("operation_name")
    
    # Your application logic with decorators
    result = my_tracked_function()
    
    # Finalize with trace data
    TraceManager.finalize_and_send(
        trace_input={"request": "data"},
        trace_output={"response": result}
    )
except Exception as e:
    # Trace will still be sent with error information
    logger.error(f"Error: {e}")
```

### 2. Environment-Specific Configuration

```python
# production.env
PROJECT_ID=my_app
ENV=production
AWS_SQS_URL=https://sqs.us-east-1.amazonaws.com/123/prod-queue

# staging.env
PROJECT_ID=my_app
ENV=staging
AWS_SQS_URL=https://sqs.us-east-1.amazonaws.com/123/staging-queue
```

### 3. Async Support

```python
@track_function()
async def async_function():
    result = await some_async_operation()
    return result

@track_llm_call(model="...")
async def async_llm_call():
    response = await async_bedrock_call()
    return response
```

## 📝 Example: Complete RAG Pipeline

# Track LLM calls
@track_llm_call()
def call_bedrock(prompt):
    # Call your LLM
    response = bedrock_client.converse(
        modelId="anthropic.claude-3-sonnet",
        messages=[{"role": "user", "content": prompt}]
    )
    return response

# Use the functions
result = process_data("some data")
llm_response = call_bedrock("Hello, world!")

# End the trace (flushes to Langfuse)
TraceManager.end_trace()
```

**Method 2: Explicit Project and Environment Override**
```python
# Override PROJECT_ID and ENV from .env
TraceManager.start_trace(
    name="chat_message",  # Operation name
    project_id="custom_project",  # Override PROJECT_ID
    environment="staging",  # Override ENV
    metadata={"user_id": "123"},
)

# Your code...

TraceManager.end_trace()
```

**Method 3: Using `finalize_and_send()` (llmops-observability)**
```python
# Start trace
TraceManager.start_trace(name="chat_session")

# Your code
user_input = "What is machine learning?"
response = await llm.generate(user_input)

# Finalize with input/output in one call
TraceManager.finalize_and_send(
    user_id="user_123",
    session_id="session_456",
    trace_name="chat_message",
    trace_input={"user_msg": user_input},
    trace_output={"bot_response": str(response)}
)
```

### 3. Capture Local Variables (Debugging)

```python
@track_function(capture_locals=True)
def complex_calculation(x, y, z):
    intermediate = x + y
    result = intermediate * z
    final = result ** 2
    # All local variables are captured in Langfuse
    return final

# Capture specific variables only
@track_function(capture_locals=["important_var", "result"])
def selective_capture(data):
    important_var = process(data)
    temp_var = "not captured"
    result = transform(important_var)
    return result
```

### 4. Nested Spans (Parent-Child Tracking)

```python
@track_function(name="parent_task")
def parent_function():
    data = fetch_data()
    # Child spans are automatically nested
    processed = child_function(data)
    return processed

@track_function(name="child_task")
def child_function(data):
    return data.upper()

# Langfuse will show: parent_task → child_task
```

### 5. ASGI Middleware (FastAPI Auto-Tracing)

```python
from fastapi import FastAPI
from llmops_observability import LLMOpsASGIMiddleware

app = FastAPI()
app.add_middleware(LLMOpsASGIMiddleware, service_name="my_api")

@app.get("/")
async def root():
    # Request is automatically traced
    return {"message": "Hello World"}

@app.post("/generate")
async def generate(prompt: str):
    # All decorated functions within request are nested
    result = await generate_text(prompt)
    return result
```

### 6. SQS Event Streaming (Event-Driven Architecture)

For event-driven, scalable deployments, the SDK supports optional event streaming to AWS SQS. Trace events are published to SQS queues where Lambda functions (or other consumers) can process them asynchronously:

```
Application (sends trace events)
    ↓
SQS Queue (decoupled message broker)
    ↓
Lambda Consumers (process & forward)
    ↓ ↓ ↓
  S3  New Relic  Datadog  (etc.)
```

**Setup:**

```bash
# Enable SQS streaming by setting AWS_SQS_URL
export AWS_SQS_URL=https://sqs.us-east-1.amazonaws.com/123456789/my-queue
export AWS_PROFILE=default
export AWS_REGION=us-east-1
```

```python
from llmops_observability import TraceManager, track_function

# When AWS_SQS_URL is set, events are automatically streamed to SQS
TraceManager.start_trace(
    name="chat_message",
    metadata={"channel": "web"}
)

@track_function()
def process_message(msg):
    return process(msg)

# All trace events are batched and sent to SQS (non-blocking)
TraceManager.end_trace()
```

**Lambda Consumer Example:**

```python
import json
import boto3

s3_client = boto3.client('s3')
newrelic = boto3.client('cloudwatch')  # Or use New Relic SDK

def lambda_handler(event, context):
    """Process trace events from SQS"""
    for record in event['Records']:
        # Parse trace event from SQS message
        trace_event = json.loads(record['body'])
        
        # Store to S3
        s3_client.put_object(
            Bucket='trace-events',
            Key=f"{trace_event['trace_id']}.json",
            Body=json.dumps(trace_event)
        )
        
        # Send metrics to New Relic
        if trace_event['event_type'] == 'llm_call':
            newrelic.put_metric_data(
                Namespace='LLMOps',
                MetricData=[{
                    'MetricName': 'TokenUsage',
                    'Value': trace_event['tokens_used'],
                    'Unit': 'Count'
                }]
            )
```

**SQS Features:**
- ✅ **Automatic Batching**: Groups events for efficient SQS sending (batch size 1-10)
- ✅ **Spillover Recovery**: Saves messages to disk if SQS is unavailable, retries on restart
- ✅ **Daemon Workers**: 4 background threads handle async SQS operations
- ✅ **Clean Shutdown**: Graceful shutdown flushes pending messages
- ✅ **Resilient**: Auto-restart failed workers, exponential backoff
- ✅ **No Blocking**: SQS operations never block main application thread

**Events Streamed to SQS:**
- `trace_start`: Trace initialization with metadata
- `span_created`: Function execution tracking
- `llm_call`: LLM API calls with token usage
- `trace_end`: Trace completion with duration

**Configuration:**
```bash
# Required: SQS queue URL
export AWS_SQS_URL=https://sqs.us-east-1.amazonaws.com/123456789/llm-traces

# Optional: AWS authentication (defaults to IAM role if in Lambda/EC2)
export AWS_PROFILE=custom-profile  # Default: "default"
export AWS_REGION=eu-west-1        # Default: "us-east-1"
```

**Use Cases:**
- 📊 Send trace events to New Relic, Datadog, CloudWatch
- 💾 Archive all LLM interactions to S3 for compliance/audit
- 🔄 Post-processing: cost calculation, quality analysis, retraining data
- 🚀 Scale: decouple application from storage/monitoring infrastructure

### 📥 Incoming SDK Message Schema

When SQS streaming is enabled, the SDK sends trace data in a compressed SQS message format that Lambda consumers can decompress and process. This section documents the message format and decompressed payload structure.

#### SQS Message Wrapper Format

```json
{
  "compressed": true,
  "data": "H4sIANPGn2YC/...",
  "trace_id": "87a41b12-cc61-4fdf-9bf2-a50a369b4d30",
  "type": "SDKTraceData"
}
```

**Wrapper Fields:**
- **`compressed`** (boolean): Indicates Base64 + Gzip compression is applied
- **`data`** (string): Base64-encoded, Gzip-compressed JSON payload
- **`trace_id`** (string): Unique trace identifier for deduplication
- **`type`** (string): Message type identifier ("SDKTraceData")

#### Decompressed SDK Trace Data Schema

```json
{
  "trace_id": "87a41b12-cc61-4fdf-9bf2-a50a369b4d30",
  "trace_name": "rag_pipeline_example",
  "project_id": "new_test",
  "environment": "uat",
  "user_id": "user_123",
  "session_id": "session_456",
  
  "start_time": 1769446311.0,
  "end_time": 1769446318.021,
  "duration_ms": 7021,
  
  "trace_input": {
    "user_msg": "What is Android ????"
  },
  "trace_output": {
    "bot_response": "Android is a mobile operating system..."
  },
  
  "token_usage": {
    "total_input_tokens": 145,
    "total_output_tokens": 87,
    "total_tokens": 232
  },
  
  "cost": {
    "total_cost": 0.000456
  },
  
  "spans": [
    {
      "span_id": "87a41b12-cc61-4fdf-9bf2-a50a369b4d31",
      "span_name": "retrieve_context",
      "span_type": "span",
      "parent_span_id": null,
      
      "start_time": 1769446311.0,
      "end_time": 1769446313.0,
      "duration_ms": 2000,
      
      "input_data": {"args": [], "kwargs": {"query": "Android"}, "locals": {}},
      "output_data": {"output": ["Doc 1", "Doc 2"]},
      
      "error": null,
      "model_id": null,
      "status": "success",
      
      "metadata": {
        "environment": "uat",
        "project_id": "new_test"
      },
      "tags": []
    },
    {
      "span_id": "87a41b12-cc61-4fdf-9bf2-a50a369b4d32",
      "span_name": "call_llm",
      "span_type": "generation",
      "parent_span_id": "87a41b12-cc61-4fdf-9bf2-a50a369b4d31",
      
      "start_time": 1769446313.0,
      "end_time": 1769446318.021,
      "duration_ms": 5021,
      
      "input_data": {"args": [], "kwargs": {"prompt": "Context: Doc 1, Doc 2\n\nQuestion: What is Android ????"}, "locals": {}},
      "output_data": {"output": {"message": {"content": "Android is a mobile operating system..."}}},
      
      "error": null,
      "model_id": "anthropic.claude-3-sonnet-20240229-v1:0",
      
      "usage": {
        "input_tokens": 145,
        "output_tokens": 87,
        "total_tokens": 232
      },
      
      "prompt": "Context: Doc 1, Doc 2\n\nQuestion: What is Android ????",
      "response": "Android is a mobile operating system developed by Google...",
      
      "metadata": {
        "environment": "uat",
        "project_id": "new_test"
      },
      
      "status": "success",
      "tags": []
    }
  ],
  
  "metadata": {
    "version": "1.0.0"
  },
  "tags": ["example", "rag"],
  
  "total_spans": 2,
  "total_generations": 1,
  
  "sdk_name": "llmops-observability",
  "sdk_version": "2.0.0"
}
```

#### Key Fields Reference

| Field | Type | Description | Auto-Injected |
|-------|------|-------------|---|
| `trace_id` | string | Unique trace identifier | - |
| `trace_name` | string | Trace/operation name | - |
| `project_id` | string | Project identifier from `PROJECT_ID` env var | ✅ |
| `environment` | string | Environment from `ENV` env var | ✅ |
| `user_id` | string | User identifier (optional) | - |
| `session_id` | string | Session identifier (optional) | - |
| `start_time` | float | Unix timestamp (seconds) | - |
| `end_time` | float | Unix timestamp (seconds) | - |
| `duration_ms` | int | Trace duration in milliseconds | - |
| `spans[].metadata.environment` | string | Environment (auto-injected to every span) | ✅ |
| `spans[].metadata.project_id` | string | Project ID (auto-injected to every span) | ✅ |
| `spans[].token_usage` | object | Input/output token counts | - |
| `spans[].cost` | object | Token cost calculation (Bedrock models) | - |

#### Lambda Decompression Example

```python
import json
import gzip
import base64

def lambda_handler(event, context):
    """Decompress and process SDK trace messages from SQS"""
    for record in event['Records']:
        # Parse SQS message
        message = json.loads(record['body'])
        
        if message.get('compressed'):
            # Decode Base64
            compressed_data = base64.b64decode(message['data'])
            
            # Decompress Gzip
            decompressed_data = gzip.decompress(compressed_data)
            
            # Parse JSON
            trace_data = json.loads(decompressed_data)
        else:
            trace_data = message
        
        # Now trace_data contains the full SDK trace with spans
        print(f"Trace: {trace_data['trace_id']}")
        print(f"Project: {trace_data['project_id']}")
        print(f"Environment: {trace_data['environment']}")
        print(f"Spans: {len(trace_data['spans'])}")
        
        # Process further (send to Langfuse, S3, NewRelic, etc.)
        process_trace(trace_data)
```

#### Size Limits and Truncation

- **Max Message Size**: 256KB (SQS hard limit)
- **Auto-Truncation**: Fields > 200KB are automatically truncated
- **Fallback to Disk**: If SQS is unavailable, messages spill to disk and retry on restart
- **Compression**: Typical traces compress to 10-30% of original size

### 8. Token Pricing & Cost Calculation

The SDK includes built-in AWS Bedrock token pricing for cost analysis:

```python
from llmops_observability.pricing import calculate_cost

# Calculate cost for a single LLM call
cost = calculate_cost(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    input_tokens=1500,
    output_tokens=800
)

print(f"Cost: ${cost:.4f}")  # Output: Cost: $0.0075

# Analyze costs by model
models_used = {
    "anthropic.claude-3-sonnet-20240229-v1:0": {
        "input_tokens": 10000,
        "output_tokens": 5000
    },
    "anthropic.claude-3-haiku-20240307-v1:0": {
        "input_tokens": 20000,
        "output_tokens": 10000
    }
}

total_cost = sum(
    calculate_cost(model, data["input_tokens"], data["output_tokens"])
    for model, data in models_used.items()
)
print(f"Total cost: ${total_cost:.4f}")
```

**Supported Models:**
- Claude 3.5 Sonnet (all variants)
- Claude 3 Sonnet/Opus/Haiku
- Claude 2.1 & 2.0
- Amazon Titan Text (Express, Lite)
- Cohere Command
- AI21 Jurassic
- Meta Llama 2 & 3

**Pricing Reference:**
All prices are updated as of 2024 and reflect AWS Bedrock official pricing. Update the pricing table in [src/llmops_observability/pricing.py](src/llmops_observability/pricing.py) as needed.

### 9. Async Support

```python
@track_function()
async def async_process(data):
    return await some_async_operation(data)

@track_llm_call(name="summarize")
async def async_llm_call(text):
    return await chain.ainvoke({"text": text})

# Both sync and async work seamlessly
```

### Per-Application Configuration

Each Gen AI application using this SDK should have **its own Langfuse project and credentials**. This ensures proper isolation and organization.

#### Step 1: Create Langfuse Project
1. Go to your Langfuse instance
2. Create a new project for your application (e.g., "chatbot-api", "doc-analyzer")
3. Copy the project's public key, secret key, and base URL

#### Step 2: Configure in Your Application

**Method 1: Environment Variables** (Recommended for production)

```bash
# .env file in your application root
LANGFUSE_PUBLIC_KEY=pk-lf-abc123...
LANGFUSE_SECRET_KEY=sk-lf-xyz789...
LANGFUSE_BASE_URL=https://langfuse.company.com
LANGFUSE_VERIFY_SSL=false
```

```python
from llmops_observability import TraceManager
from dotenv import load_dotenv

load_dotenv()  # Loads .env from current directory
# SDK auto-configures from environment variables
```

**Method 2: Explicit Configuration** (Recommended for testing)

```python
from llmops_observability import configure
import os

# At application startup (e.g., main.py)
configure(
    public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
    secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
    base_url=os.getenv("LANGFUSE_BASE_URL"),
    verify_ssl=False
)
```

### Environment Variables Reference

| Variable | Required | Default | Description |
|----------|----------|---------|-------------|
| `LANGFUSE_PUBLIC_KEY` | Yes | None | Langfuse public key from your project |
| `LANGFUSE_SECRET_KEY` | Yes | None | Langfuse secret key from your project |
| `LANGFUSE_BASE_URL` | Yes | None | Langfuse instance URL |
| `LANGFUSE_VERIFY_SSL` | No | `false` | Whether to verify SSL certificates |
| `PROJECT_ID` | No | `unknown_project` | Project identifier (used as trace name in Langfuse) |
| `ENV` | No | `development` | Environment name (production, staging, development, etc.) - automatically mapped to `LANGFUSE_TRACING_ENVIRONMENT` |
| `MODEL_ID` | No | `anthropic.claude-3-5-sonnet-20241022-v2:0` | Default model ID for cost calculation when not explicitly provided |
| `AWS_SQS_URL` | No | None | AWS SQS queue URL (when provided, enables SQS event streaming) |
| `AWS_PROFILE` | No | `default` | AWS profile name for SQS authentication |
| `AWS_REGION` | No | `us-east-1` | AWS region for SQS |
| `LANGFUSE_DEBUG` | No | `false` | Enable debug logging for Langfuse client |

**Environment Tracking:**
- The `ENV` variable is automatically mapped to Langfuse's `LANGFUSE_TRACING_ENVIRONMENT`
- This applies the environment as a top-level attribute to all traces and observations
- Allows easy filtering by environment in Langfuse UI
- Must follow regex: `^(?!langfuse)[a-z0-9-_]+$` with max 40 characters
Track regular function execution with optional local variable capture.

```python
@track_function()
def my_function(x, y):
    return x + y

@track_function(name="custom_name", tags={"version": "1.0"})
def another_function():
    pass

# Capture all local variables for debugging
@track_function(capture_locals=True)
def debug_function(data):
    step1 = process(data)
    step2 = transform(step1)
    return step2  # All locals captured in Langfuse

# Capture specific variables only
@track_function(capture_locals=["result", "important_var"])
def selective_function(input):
    temp = input * 2  # Not captured
    result = temp + 10  # Captured
    important_var = compute(result)  # Captured
    return important_var
```

**Parameters:**
- `name`: Custom span name (default: function name)
- `tags`: Dictionary of tags/metadata
- `capture_locals`: Capture local variables - `True` (all), `False` (none), or list of variable names
- `capture_self`: Whether to capture `self` in methods (default: `True`)

## API Reference

### TraceManager

#### `start_trace(name, project_id=None, environment=None, metadata=None, user_id=None, session_id=None, tags=None)`
Start a new trace with project and environment tracking.

```python
TraceManager.start_trace(
    name="chat_message",  # Operation name (required)
    project_id="my_project",  # Optional: defaults to PROJECT_ID env var
    environment="production",  # Optional: defaults to ENV env var
    metadata={"custom": "data"},
    user_id="user_123",
    session_id="session_456",
    tags=["experiment"]
)
```

**Parameters:**
- `name` (required): Operation/trace name (e.g., "chat_message", "document_analysis")
- `project_id` (optional): Project identifier. Defaults to `PROJECT_ID` from `.env`. Used as trace name in Langfuse.
- `environment` (optional): Environment name (e.g., "production", "staging"). Defaults to `ENV` from `.env`. Automatically mapped to `LANGFUSE_TRACING_ENVIRONMENT`.
- `metadata` (optional): Custom metadata dictionary
- `user_id` (optional): User identifier
- `session_id` (optional): Session identifier
- `tags` (optional): List of tags

**Returns:** Trace ID (string)

**Example with .env auto-loading:**
```bash
# .env file
PROJECT_ID=chatbot-api
ENV=production
```

```python
# Automatically uses PROJECT_ID and ENV from .env
TraceManager.start_trace(
    name="user_query",
    metadata={"version": "2.0"}
)
# Trace name in Langfuse: "chatbot-api"
# Environment in Langfuse: "production"
```

#### `end_trace()`
End the current trace and flush to Langfuse.

```python
TraceManager.end_trace()
```

#### `finalize_and_send(user_id, session_id, trace_name, trace_input, trace_output)`
Finalize and send the trace with input/output metadata.

This is a convenience method that combines setting trace metadata and ending the trace in one call.

```python
TraceManager.start_trace(name="chat_message")

# ... your code executes ...

# Finalize with input/output details
TraceManager.finalize_and_send(
    user_id="user_123",
    session_id="session_456",
    trace_name="be