Monitoring et Observabilité IA : LangSmith, Phoenix, Helicone

# Même prompt, réponses différentes
response1 = llm("Résume ce texte")  # "Le texte parle de..."
response2 = llm("Résume ce texte")  # "Ce document traite de..."
# → Comment comparer la qualité ?

# Coût par requête varie selon input + output
query1 = "Bonjour"              # $0.0001
query2 = "Analyse ce livre..."  # $0.50 (avec long contexte RAG)
# → Besoin de tracking granulaire

Total: 3.2s
├─ Embedding: 0.1s
├─ Vector search: 0.3s
├─ LLM: 2.5s (où est le bottleneck ?)
└─ Post-processing: 0.3s

Comment mesurer si "Réponse A" est meilleure que "Réponse B" ?
→ Besoin de métriques proxy + human eval

import time
from functools import wraps

def track_latency(component_name):
    """Decorator pour tracker latence"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time.time()
            result = func(*args, **kwargs)
            duration = time.time() - start

            # Log vers monitoring tool
            logger.info({
                "component": component_name,
                "duration_ms": duration * 1000,
                "timestamp": time.time()
            })

            return result
        return wrapper
    return decorator

# Usage
@track_latency("embedding")
def create_embedding(text):
    return embeddings.embed_query(text)

@track_latency("vector_search")
def search_docs(query_vector):
    return vectorstore.similarity_search(query_vector, k=5)

@track_latency("llm_generation")
def generate_response(prompt):
    return llm.invoke(prompt)

from tiktoken import get_encoding

class CostTracker:
    """Track coûts OpenAI en temps réel"""

    PRICES = {
        "gpt-4o": {"input": 2.50, "output": 10.00},  # $/1M tokens
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "text-embedding-3-small": {"input": 0.02, "output": 0}
    }

    def __init__(self):
        self.enc = get_encoding("cl100k_base")
        self.daily_cost = 0

    def calculate_cost(self, model: str, input_text: str, output_text: str) -> float:
        """Calcule coût d'un appel"""

        input_tokens = len(self.enc.encode(input_text))
        output_tokens = len(self.enc.encode(output_text))

        prices = self.PRICES.get(model, {"input": 0, "output": 0})

        cost = (
            (input_tokens * prices["input"] / 1_000_000) +
            (output_tokens * prices["output"] / 1_000_000)
        )

        self.daily_cost += cost

        return cost

    def log_cost(self, model: str, input_text: str, output_text: str, user_id: str):
        """Log coût vers monitoring"""

        cost = self.calculate_cost(model, input_text, output_text)

        logger.info({
            "event": "llm_call",
            "model": model,
            "user_id": user_id,
            "cost_usd": cost,
            "daily_total": self.daily_cost,
            "timestamp": time.time()
        })

        # Alerter si dépassement
        if self.daily_cost > 100:  # $100/jour
            send_alert(f"⚠️ Daily cost: ${self.daily_cost:.2f}")

# Usage
tracker = CostTracker()
tracker.log_cost("gpt-4o", prompt, response, user_id="user_123")

from langchain.evaluation import load_evaluator

# Évaluateur de pertinence
relevance_evaluator = load_evaluator("qa")

def evaluate_response(question: str, answer: str, context: str):
    """Évalue qualité d'une réponse RAG"""

    # 1. Pertinence (LLM-as-judge)
    relevance_score = relevance_evaluator.evaluate_strings(
        prediction=answer,
        input=question,
        reference=context
    )

    # 2. Groundedness (réponse basée sur contexte)
    grounding_prompt = f"""
    Contexte: {context}
    Réponse: {answer}

    La réponse est-elle entièrement basée sur le contexte ?
    Réponds uniquement par "oui" ou "non".
    """

    grounding = llm.invoke(grounding_prompt).content.lower()
    grounded = "oui" in grounding

    # 3. Log métriques
    logger.info({
        "question": question,
        "relevance_score": relevance_score["score"],
        "grounded": grounded,
        "answer_length": len(answer),
        "timestamp": time.time()
    })

    return {
        "relevance": relevance_score["score"],
        "grounded": grounded
    }

# Créer compte sur smith.langchain.com
# Générer API key

# .env
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__...
LANGCHAIN_PROJECT=my-chatbot-prod

Run: user_query_abc123
├─ Retriever
│  ├─ Embedding (100ms, $0.00002)
│  └─ Vector Search (250ms)
│     → 5 documents retrieved
├─ Prompt Template (5ms)
│  → Rendered prompt (2,341 tokens)
└─ LLM (ChatOpenAI)
   ├─ Input: 2,341 tokens ($0.0059)
   ├─ Output: 287 tokens ($0.0029)
   ├─ Duration: 2,150ms
   └─ Response: "Voici la réponse..."

Total: 2.5s, $0.0088

from langsmith import Client

client = Client()

# Créer dataset d'évaluation
dataset = client.create_dataset("rag-eval-v1")

# Ajouter exemples
client.create_examples(
    dataset_id=dataset.id,
    inputs=[
        {"question": "Qu'est-ce que le RAG ?"},
        {"question": "Comment fonctionne l'embedding ?"}
    ],
    outputs=[
        {"answer": "Le RAG combine recherche et génération..."},
        {"answer": "L'embedding transforme texte en vecteurs..."}
    ]
)

from langsmith.evaluation import evaluate

def my_evaluator(run, example):
    """Évaluateur custom"""
    # run.outputs["answer"] = réponse du modèle
    # example.outputs["answer"] = réponse attendue

    # Calculer similarité
    from difflib import SequenceMatcher
    similarity = SequenceMatcher(
        None,
        run.outputs["answer"],
        example.outputs["answer"]
    ).ratio()

    return {"score": similarity}

# Lancer évaluation
results = evaluate(
    lambda inputs: my_chain.invoke(inputs),
    data=dataset.name,
    evaluators=[my_evaluator]
)

# Résultats disponibles dans dashboard

# Via UI LangSmith :
# 1. Settings → Automations
# 2. Create Alert
#    - Condition: error_rate > 5%
#    - Notification: Email, Slack, Webhook

# Ou via code
client.create_rule(
    project="my-chatbot-prod",
    rule_type="threshold",
    conditions={
        "metric": "error_rate",
        "operator": ">",
        "value": 0.05
    },
    actions=[
        {"type": "email", "recipients": ["[email protected]"]},
        {"type": "slack", "webhook": "https://hooks.slack.com/..."}
    ]
)

docker run -p 6006:6006 arizephoenix/phoenix:latest

pip install arize-phoenix

# Lancer serveur
python -m phoenix.server.main serve

from phoenix.trace.langchain import LangChainInstrumentor

# Instrumenter LangChain
LangChainInstrumentor().instrument()

# Vos chains fonctionnent normalement, traces automatiques !
from langchain.chains import RetrievalQA

qa_chain = RetrievalQA.from_chain_type(...)
response = qa_chain.invoke({"query": "Qu'est-ce que le RAG ?"})
# → Tracé automatiquement dans Phoenix

from openinference.instrumentation.openai import OpenAIInstrumentor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor

# Setup OpenTelemetry
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(
    SimpleSpanProcessor(OTLPSpanExporter("http://localhost:6006/v1/traces"))
)

# Instrumenter OpenAI
OpenAIInstrumentor().instrument(tracer_provider=tracer_provider)

# Utiliser OpenAI normalement
from openai import OpenAI
client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}]
)
# → Tracé dans Phoenix

import phoenix as px

# Analyser embeddings
df = px.load_dataset("your_embeddings.parquet")

# Visualiser clusters
px.launch_app(
    primary=px.Dataset(df, "embeddings_column"),
    schema=px.Schema(
        embedding_feature_column_names=["embedding"],
        prediction_label_column_name="category"
    )
)
# → Interface interactive pour explorer embeddings

from phoenix.experimental.evals import (
    HallucinationEvaluator,
    RelevanceEvaluator,
    run_evals
)

# Évaluer hallucinations
hallucination_eval = HallucinationEvaluator()
relevance_eval = RelevanceEvaluator()

results = run_evals(
    dataframe=traces_df,
    evaluators=[hallucination_eval, relevance_eval],
    provide_explanation=True
)

# Résultats : scores + explications
print(results[["hallucination_score", "hallucination_explanation"]])

Votre app → Helicone Proxy → OpenAI
              ↓
         Monitoring

from openai import OpenAI

client = OpenAI(
    api_key="sk-...",  # Votre clé OpenAI
    base_url="https://oai.helicone.ai/v1",  # Proxy Helicone
    default_headers={
        "Helicone-Auth": "Bearer <HELICONE_API_KEY>"
    }
)

# Utiliser normalement
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}]
)
# → Requête passe par Helicone → loggée → envoyée à OpenAI

from langchain.chat_models import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o",
    openai_api_key="sk-...",
    openai_api_base="https://oai.helicone.ai/v1",
    model_kwargs={
        "extra_headers": {
            "Helicone-Auth": "Bearer <HELICONE_API_KEY>"
        }
    }
)

# Activer cache (économiser coûts)
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
    extra_headers={
        "Helicone-Auth": "Bearer ...",
        "Helicone-Cache-Enabled": "true"
    }
)
# Si même prompt déjà vu → réponse depuis cache (gratuit, <50ms)

# Associer requêtes à utilisateurs
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
    extra_headers={
        "Helicone-Auth": "Bearer ...",
        "Helicone-User-Id": "user_12345"
    }
)
# → Dashboard : coûts par user, usage patterns

# Ajouter métadonnées custom
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
    extra_headers={
        "Helicone-Auth": "Bearer ...",
        "Helicone-Property-Environment": "production",
        "Helicone-Property-Feature": "chatbot",
        "Helicone-Property-Version": "v2.1"
    }
)
# → Filtrer dashboard par propriétés

# LangSmith pour LangChain chains
LANGCHAIN_TRACING_V2=true

# Helicone pour appels OpenAI directs
openai_client = OpenAI(base_url="https://oai.helicone.ai/v1")

# 1. Coûts excessifs
if daily_cost > budget_daily * 1.5:
    send_alert(f"🚨 Daily cost: ${daily_cost:.2f} (budget: ${budget_daily})")

# 2. Latence dégradée
if latency_p95 > 5000:  # 5s
    send_alert(f"⚠️ Latency p95: {latency_p95}ms (SLA: 3000ms)")

# 3. Taux d'erreur élevé
if error_rate > 0.05:  # 5%
    send_alert(f"❌ Error rate: {error_rate:.1%}")

# 4. Rate limits atteints
if rate_limit_errors > 10:
    send_alert(f"⏱️ Rate limit hit {rate_limit_errors} times in 1h")

# 5. Qualité dégradée
if avg_relevance_score < 0.7:
    send_alert(f"📉 Relevance score dropped to {avg_relevance_score:.2f}")

import requests

def send_slack_alert(message: str, channel: str = "#alerts"):
    """Envoie alerte Slack"""

    webhook_url = os.getenv("SLACK_WEBHOOK_URL")

    payload = {
        "channel": channel,
        "username": "AI Monitoring",
        "icon_emoji": ":robot_face:",
        "text": message
    }

    response = requests.post(webhook_url, json=payload)
    return response.status_code == 200

# Usage
send_slack_alert("🚨 Daily AI cost exceeded $100: $127.50")

import numpy as np
from collections import deque

class AnomalyDetector:
    """Détecte anomalies par écart à la moyenne"""

    def __init__(self, window_size=100, threshold=3):
        self.window = deque(maxlen=window_size)
        self.threshold = threshold  # Nombre d'écarts types

    def is_anomaly(self, value: float) -> bool:
        """Retourne True si anomalie détectée"""

        if len(self.window) < 10:
            self.window.append(value)
            return False

        mean = np.mean(self.window)
        std = np.std(self.window)

        # Z-score
        z_score = abs((value - mean) / std) if std > 0 else 0

        self.window.append(value)

        return z_score > self.threshold

# Usage
latency_detector = AnomalyDetector()
cost_detector = AnomalyDetector()

# Pour chaque requête
if latency_detector.is_anomaly(current_latency):
    send_alert(f"⚠️ Latency spike: {current_latency}ms")

if cost_detector.is_anomaly(current_cost):
    send_alert(f"💸 Cost anomaly: ${current_cost:.4f}")

# Version A
prompt_a = "Résume ce texte en 2 phrases."

# Version B
prompt_b = "Tu es un expert en résumés. Résume ce texte en 2 phrases concises."

# Laquelle performe mieux ?
# → A/B test pour le savoir

import random

class PromptABTest:
    """A/B testing de prompts"""

    def __init__(self):
        self.variants = {}
        self.results = {}

    def add_variant(self, name: str, prompt_template: str, weight: float = 0.5):
        """Ajoute variant"""
        self.variants[name] = {
            "template": prompt_template,
            "weight": weight
        }
        self.results[name] = {
            "count": 0,
            "total_latency": 0,
            "total_cost": 0,
            "thumbs_up": 0,
            "thumbs_down": 0
        }

    def get_variant(self):
        """Sélectionne variant (weighted random)"""
        variants = list(self.variants.keys())
        weights = [self.variants[v]["weight"] for v in variants]
        return random.choices(variants, weights=weights)[0]

    def log_result(self, variant: str, latency: float, cost: float, user_feedback: int):
        """Log résultat (user_feedback: 1 = 👍, -1 = 👎, 0 = none)"""

        self.results[variant]["count"] += 1
        self.results[variant]["total_latency"] += latency
        self.results[variant]["total_cost"] += cost

        if user_feedback == 1:
            self.results[variant]["thumbs_up"] += 1
        elif user_feedback == -1:
            self.results[variant]["thumbs_down"] += 1

    def get_stats(self):
        """Retourne statistiques"""
        stats = {}

        for variant, data in self.results.items():
            if data["count"] == 0:
                continue

            stats[variant] = {
                "requests": data["count"],
                "avg_latency": data["total_latency"] / data["count"],
                "avg_cost": data["total_cost"] / data["count"],
                "satisfaction": (
                    data["thumbs_up"] /
                    (data["thumbs_up"] + data["thumbs_down"])
                    if (data["thumbs_up"] + data["thumbs_down"]) > 0
                    else 0
                )
            }

        return stats

# Usage
ab_test = PromptABTest()

ab_test.add_variant(
    "control",
    "Résume ce texte en 2 phrases: {text}",
    weight=0.5
)

ab_test.add_variant(
    "expert",
    "Tu es un expert en résumés. Résume ce texte en 2 phrases concises: {text}",
    weight=0.5
)

# Pour chaque requête
variant = ab_test.get_variant()
prompt = ab_test.variants[variant]["template"].format(text=user_text)

# Appeler LLM, mesurer latence/coût
response, latency, cost = call_llm(prompt)

# Logger (user_feedback ajouté plus tard via thumbs up/down)
ab_test.log_result(variant, latency, cost, user_feedback=0)

# Après 1000 requêtes, analyser
print(ab_test.get_stats())
# {
#   "control": {"requests": 497, "avg_latency": 2.3s, "satisfaction": 0.72},
#   "expert": {"requests": 503, "avg_latency": 2.5s, "satisfaction": 0.81}
# }
# → "expert" gagne en satisfaction, déployer à 100%

from langsmith import Client

client = Client()

# Tag chaque run avec variant
def run_with_variant(variant: str, input_data: dict):
    # Ajouter tag
    with client.trace(
        name="rag_chain",
        tags=[f"variant:{variant}"],
        metadata={"variant": variant}
    ):
        response = chain.invoke(input_data)

    return response

# Analyser dans LangSmith UI :
# Filter runs by tag "variant:A" vs "variant:B"
# Comparer latency, cost, user feedback