Comprehensive comparison for Observability technology in AI applications

import os
from openai import OpenAI
from ddtrace import tracer, patch
from ddtrace.llmobs import LLMObs
from ddtrace.llmobs.decorators import embedding, llm, workflow, tool
from flask import Flask, request, jsonify
import logging

# Initialize Datadog LLM Observability
LLMObs.enable(
    ml_app="product-recommendation-service",
    api_key=os.getenv("DD_API_KEY"),
    site=os.getenv("DD_SITE", "datadoghq.com"),
    env=os.getenv("DD_ENV", "production"),
    service="recommendation-api"
)

# Patch OpenAI for automatic tracing
patch(openai=True)

app = Flask(__name__)
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
logger = logging.getLogger(__name__)

@tool(name="fetch_user_history")
def fetch_user_history(user_id: str) -> dict:
    """Simulates fetching user purchase history from database"""
    try:
        # In production, this would query your database
        history = {
            "recent_purchases": ["laptop", "wireless mouse", "USB-C cable"],
            "categories": ["electronics", "accessories"],
            "budget_range": "mid-high"
        }
        LLMObs.annotate(input_data={"user_id": user_id}, output_data=history)
        return history
    except Exception as e:
        logger.error(f"Error fetching user history: {str(e)}")
        LLMObs.annotate(metadata={"error": str(e)})
        raise

@llm(model_name="gpt-4", model_provider="openai")
def generate_recommendations(user_context: str, history: dict) -> str:
    """Generates personalized product recommendations using LLM"""
    try:
        prompt = f"""Based on the user's purchase history: {history['recent_purchases']}
        and their preferred categories: {history['categories']},
        recommend 3 complementary products. Budget: {history['budget_range']}.
        User query: {user_context}
        
        Provide concise recommendations with reasoning."""
        
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            max_tokens=500
        )
        
        recommendation = response.choices[0].message.content
        
        # Annotate with custom metadata
        LLMObs.annotate(
            input_data={"prompt": prompt},
            output_data={"recommendation": recommendation},
            metadata={
                "tokens_used": response.usage.total_tokens,
                "model": "gpt-4",
                "temperature": 0.7
            },
            tags={"feature": "recommendations", "budget": history['budget_range']}
        )
        
        return recommendation
    except Exception as e:
        logger.error(f"LLM generation failed: {str(e)}")
        LLMObs.annotate(metadata={"error": str(e), "error_type": type(e).__name__})
        raise

@workflow()
def recommendation_workflow(user_id: str, query: str) -> dict:
    """Main workflow orchestrating the recommendation process"""
    try:
        # Step 1: Fetch user data
        user_history = fetch_user_history(user_id)
        
        # Step 2: Generate recommendations
        recommendations = generate_recommendations(query, user_history)
        
        result = {
            "user_id": user_id,
            "recommendations": recommendations,
            "status": "success"
        }
        
        LLMObs.annotate(
            input_data={"user_id": user_id, "query": query},
            output_data=result,
            tags={"workflow_status": "completed"}
        )
        
        return result
    except Exception as e:
        logger.error(f"Workflow failed: {str(e)}")
        LLMObs.annotate(
            metadata={"error": str(e), "stage": "workflow"},
            tags={"workflow_status": "failed"}
        )
        return {"status": "error", "message": str(e)}

@app.route("/api/v1/recommendations", methods=["POST"])
def get_recommendations():
    """API endpoint for product recommendations"""
    try:
        data = request.get_json()
        user_id = data.get("user_id")
        query = data.get("query")
        
        if not user_id or not query:
            return jsonify({"error": "Missing user_id or query"}), 400
        
        # Execute workflow with Datadog tracing
        result = recommendation_workflow(user_id, query)
        
        if result.get("status") == "error":
            return jsonify(result), 500
        
        return jsonify(result), 200
    except Exception as e:
        logger.error(f"API error: {str(e)}")
        return jsonify({"error": "Internal server error"}), 500

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=False)