AI Observability Stack

Overview

Full observability for AI systems with OpenTelemetry. This guide provides practical, production-ready implementations.

Category: ai-infrastructure Primary Tool: opentelemetry Tags: infrastructure, devops, opentelemetry, production

Prerequisites

bash
pip install openai anthropic opentelemetry python-dotenv
export OPENAI_API_KEY="sk-..."

Core Implementation

python
import os
from openai import OpenAI
from typing import Optional, Any
import json
client = OpenAI()
class AI_Observability_Stack:
    """AI Observability Stack
    
    Full observability for AI systems with OpenTelemetry
    """
    
    def __init__(self, model: str = "gpt-4o", temperature: float = 0.3):
        self.client = OpenAI()
        self.model = model
        self.temperature = temperature
        self.system = """You are an AI expert in ai-infrastructure. 
        Provide accurate, practical, production-ready assistance.
        Be clear, concise, and well-structured."""
    
    def run(self, query: str, context: Optional[dict] = None) -> dict:
        """Execute the main workflow."""
        
        messages = [{"role": "system", "content": self.system}]
        
        if context:
            messages.append({
                "role": "user",
                "content": f"Context: {json.dumps(context, indent=2)}"
            })
        
        messages.append({"role": "user", "content": query})
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=self.temperature,
            max_tokens=2000
        )
        
        return {
            "output": response.choices[0].message.content,
            "model": self.model,
            "tokens": response.usage.total_tokens,
            "category": "ai-infrastructure"
        }
    
    def batch_run(self, queries: list[str]) -> list[dict]:
        """Process multiple queries."""
        return [self.run(q) for q in queries]
Usage
tool_instance = AI_Observability_Stack()
result = tool_instance.run("How do I implement ai observability stack?")
print(result["output"])

Advanced Usage

python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI(title="AI Observability Stack API")
tool_instance = AI_Observability_Stack()
class Request(BaseModel):
    query: str
    context: dict = {}
@app.post("/run")
async def run_endpoint(req: Request):
    try:
        result = tool_instance.run(req.query, req.context)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health():
    return {"status": "ok", "tool": "AI Observability Stack"}

Best Practices

Input validation — always validate and sanitize inputs

Error handling — handle API failures gracefully with retries

Rate limiting — respect API rate limits with backoff

Caching — cache responses to reduce costs

Monitoring — track usage, costs, and quality metrics

Testing

python
import pytest
@pytest.fixture
def tool():
    return AI_Observability_Stack(model="gpt-4o-mini")
def test_basic_functionality(tool):
    result = tool.run("Test query for AI Observability Stack")
    assert "output" in result
    assert len(result["output"]) > 10
    assert result["category"] == "ai-infrastructure"
def test_batch_processing(tool):
    queries = ["Query 1", "Query 2", "Query 3"]
    results = tool.batch_run(queries)
    assert len(results) == 3
    assert all("output" in r for r in results)

Resources

OpenAI API: https://platform.openai.com/docs

opentelemetry documentation

Related tutorials on infrastructure, devops, opentelemetry, production