📊 Observabilidade

# Configuração mínima (3 variáveis de ambiente)
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"] = "ls__..."

# Opcionalmente, defina o projeto
os.environ["LANGCHAIN_PROJECT"] = "meu-agente-producao"

# Após isso, TODA execução LangGraph é trackeada automaticamente
resultado = app.invoke(input, config=config)
# → Trace visível em smith.langchain.com imediatamente

# Tags para filtrar runs no dashboard
resultado = app.invoke(input, config={
    **config,
    "tags": ["producao", "usuario-premium", "analise-mercado"],
    "metadata": {"user_id": "u123", "version": "2.1.0"}
})

# Criando datasets para regression testing
from langsmith import Client
client = Client()

# Cria dataset a partir de runs anteriores bem-sucedidas
dataset = client.create_dataset("golden-set-analise-mercado")
runs = client.list_runs(
    project_name="meu-agente-producao",
    filter='and(eq(feedback_key, "qualidade"), gte(feedback_score, 4))'
)
for run in runs:
    client.create_example(
        inputs=run.inputs,
        outputs=run.outputs,
        dataset_id=dataset.id
    )

# Avaliação em batch no dataset
from langsmith.evaluation import evaluate, LangChainStringEvaluator

evaluator = LangChainStringEvaluator("criteria", config={
    "criteria": "helpfulness"
})
results = evaluate(
    lambda inputs: app.invoke(inputs),
    data=dataset.name,
    evaluators=[evaluator],
    experiment_prefix="v2.1-vs-v2.0"
)

# Instalação
# pip install arize-phoenix openinference-instrumentation-langchain

import phoenix as px
from openinference.instrumentation.langchain import LangChainInstrumentor
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider

# 1. Inicia o Phoenix localmente
session = px.launch_app()
# → UI disponível em http://localhost:6006

# 2. Configura OpenTelemetry para exportar para Phoenix
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import SimpleSpanProcessor

tracer_provider = TracerProvider()
tracer_provider.add_span_processor(
    SimpleSpanProcessor(
        OTLPSpanExporter(endpoint=session.url + "/v1/traces")
    )
)
trace.set_tracer_provider(tracer_provider)

# 3. Instrumenta automaticamente o LangChain/LangGraph
LangChainInstrumentor().instrument()

# 4. A partir daqui, toda execução aparece no Phoenix!
resultado = app.invoke(input)
# → Trace visível em http://localhost:6006

# Exportando para CSV para análise offline
df = px.Client().get_spans_dataframe()
df.to_csv("traces_export.csv", index=False)

# Avaliação de qualidade no Phoenix
from phoenix.evals import (
    HallucinationEvaluator, QAEvaluator, run_evals
)
from phoenix.evals.models import OpenAIModel

eval_model = OpenAIModel(model="gpt-4o-mini")
hallucination_evaluator = HallucinationEvaluator(eval_model)
qa_evaluator = QAEvaluator(eval_model)

results = run_evals(
    dataframe=df,
    evaluators=[hallucination_evaluator, qa_evaluator],
    provide_explanation=True
)

import time
from dataclasses import dataclass
from langsmith import Client

client = Client()

# Rastreando métricas de negócio via feedback
async def executar_com_metricas(input_data: dict, user_id: str):
    start_time = time.time()

    # Executa o agente
    resultado = await app.ainvoke(input_data, config={
        "configurable": {"thread_id": f"{user_id}_{int(start_time)}"},
        "metadata": {"user_id": user_id}
    })

    latencia = time.time() - start_time
    objetivo_atingido = verificar_objetivo(resultado, input_data)

    # Envia métricas para LangSmith como feedback
    run_id = resultado.get("run_id")  # ID do trace
    if run_id:
        client.create_feedback(
            run_id=run_id,
            key="objetivo_atingido",
            score=1.0 if objetivo_atingido else 0.0
        )
        client.create_feedback(
            run_id=run_id,
            key="latencia_segundos",
            value=latencia
        )
        client.create_feedback(
            run_id=run_id,
            key="custo_usd",
            value=calcular_custo(resultado)
        )

    return resultado

# Query de métricas agregadas
runs = client.list_runs(
    project_name="producao",
    start_time=datetime.now() - timedelta(days=7)
)
taxa_sucesso = sum(r.feedback_stats.get("objetivo_atingido", {}).get("avg", 0)
                   for r in runs) / len(list(runs))

import asyncio
from datetime import datetime, timedelta

# Thresholds de alerta
ALERTAS = {
    "taxa_erro_pct": {"warning": 5.0, "critical": 15.0},
    "latencia_p95_seg": {"warning": 30.0, "critical": 60.0},
    "custo_hora_usd": {"warning": 10.0, "critical": 25.0},
    "qualidade_score": {"warning": 3.5, "critical": 2.5},  # mínimo
}

async def monitorar_metricas():
    """Loop de monitoramento que roda a cada 5 minutos."""
    while True:
        metricas = await coletar_metricas_recentes(janela_minutos=5)
        alertas_disparados = []

        for metrica, valor in metricas.items():
            threshold = ALERTAS.get(metrica)
            if not threshold:
                continue

            if valor >= threshold.get("critical", float("inf")):
                alertas_disparados.append({
                    "metrica": metrica,
                    "valor": valor,
                    "nivel": "CRITICAL",
                    "threshold": threshold["critical"]
                })
            elif valor >= threshold.get("warning", float("inf")):
                alertas_disparados.append({
                    "metrica": metrica,
                    "valor": valor,
                    "nivel": "WARNING",
                    "threshold": threshold["warning"]
                })

        if alertas_disparados:
            await enviar_alerta(alertas_disparados)

        await asyncio.sleep(300)  # 5 minutos

async def enviar_alerta(alertas: list):
    """Envia alerta para Slack e/ou PagerDuty."""
    mensagem = f"🚨 Alerta em produção: {datetime.now()}\n"
    for alerta in alertas:
        emoji = "🔴" if alerta["nivel"] == "CRITICAL" else "🟡"
        mensagem += f"{emoji} {alerta['metrica']}: {alerta['valor']:.2f} "
        mensagem += f"(threshold: {alerta['threshold']})\n"

    # Slack
    await slack_client.post(channel="#alertas-prod", text=mensagem)

    # PagerDuty para críticos
    if any(a["nivel"] == "CRITICAL" for a in alertas):
        await pagerduty.trigger(summary=mensagem, severity="critical")

from langchain_openai import ChatOpenAI
from pydantic import BaseModel
import random

# Schema de avaliação
class AvaliacaoQualidade(BaseModel):
    relevancia: int        # 1-5: resposta atende ao objetivo?
    completude: int        # 1-5: cobre todos os aspectos?
    corretude: int         # 1-5: informações corretas?
    tom: int               # 1-5: tom apropriado para o contexto?
    score_geral: float     # média ponderada
    justificativa: str
    pontos_melhoria: list[str]

# Judge LLM (modelo separado, mais barato)
judge_llm = ChatOpenAI(model="gpt-4o-mini").with_structured_output(
    AvaliacaoQualidade
)

RUBRICA_AVALIACAO = """
Você é um avaliador de qualidade de assistentes de IA.
Avalie a resposta do assistente com a seguinte rubrica (1-5 cada):

- Relevância (1-5): A resposta atende ao objetivo específico da pergunta?
  1 = completamente irrelevante
  5 = perfeitamente alinhada ao objetivo

- Completude (1-5): Todos os aspectos importantes foram cobertos?
  1 = resposta superficial e incompleta
  5 = todos os aspectos abordados com profundidade

- Corretude (1-5): As informações fornecidas são corretas e verificáveis?
  1 = informações incorretas
  5 = todas as afirmações são corretas e verificáveis

- Tom (1-5): O tom é apropriado para o contexto e perfil do usuário?
  1 = tom completamente inadequado
  5 = tom perfeito para o contexto

Contexto da avaliação:
- Pergunta do usuário: {pergunta}
- Resposta do agente: {resposta}
- Contexto (se disponível): {contexto}
"""

async def avaliar_qualidade(
    pergunta: str,
    resposta: str,
    contexto: str = ""
) -> AvaliacaoQualidade:
    """Avalia a qualidade de uma resposta usando LLM-as-judge."""
    prompt = RUBRICA_AVALIACAO.format(
        pergunta=pergunta,
        resposta=resposta,
        contexto=contexto
    )
    avaliacao = await judge_llm.ainvoke(prompt)
    # Calcula score ponderado
    avaliacao.score_geral = (
        avaliacao.relevancia * 0.35 +
        avaliacao.completude * 0.25 +
        avaliacao.corretude * 0.30 +
        avaliacao.tom * 0.10
    )
    return avaliacao

# Pipeline de avaliação contínua
async def avaliar_amostra_producao(taxa_amostragem: float = 0.1):
    """
    Avalia 10% das execuções de produção automaticamente.
    Roda como background job a cada hora.
    """
    runs_recentes = await buscar_runs_sem_avaliacao(limite=100)

    for run in runs_recentes:
        # Amostragem aleatória para controle de custo
        if random.random() > taxa_amostragem:
            continue

        avaliacao = await avaliar_qualidade(
            pergunta=run["input"],
            resposta=run["output"]
        )

        # Registra no LangSmith para trending
        client.create_feedback(
            run_id=run["id"],
            key="llm_judge_score",
            score=avaliacao.score_geral / 5.0  # normaliza 0-1
        )

        # Alerta se qualidade abaixo do threshold
        if avaliacao.score_geral < 3.0:
            await notificar_qualidade_baixa(run, avaliacao)
            logger.warning(
                f"Qualidade baixa detectada: run_id={run['id']}, "
                f"score={avaliacao.score_geral:.2f}, "
                f"melhorias={avaliacao.pontos_melhoria}"
            )