Depuração e Rastreamento

Por que print() não é suficiente

agent.run("Pesquise sobre IA")
# Saída: uma string final
# Você não sabe:
# - Quantas chamadas ao LLM?
# - Qual tool foi chamada?
# - Qual foi o custo?
# - Por que parou?

with langsmith.trace("agent-run"):
    agent.run("Pesquise sobre IA")
# Você vê:
# - 3 chamadas ao GPT-4o
# - 2 tool calls (search + calc)
# - $0.042 de custo
# - 8.3s de latência total

Estrutura de um trace hierárquico

Dashboard de métricas com LangSmith SDK

from langsmith import Client
from datetime import datetime, timedelta
import statistics

client = Client()

# Coleta runs das últimas 24h
runs = list(client.list_runs(
    project_name="meu-agente",
    start_time=datetime.now() - timedelta(hours=24)
))

# Calcula métricas
latencias = [
    (r.end_time - r.start_time).seconds
    for r in runs if r.end_time
]
custos = [r.total_cost for r in runs if r.total_cost]
sucessos = sum(1 for r in runs if r.status == "success")

print(f"Total de runs: {len(runs)}")
print(f"Success rate: {sucessos/len(runs)*100:.1f}%")
print(f"Latência mediana: {statistics.median(latencias):.1f}s")
print(f"Latência P95: {sorted(latencias)[int(len(latencias)*0.95)]:.1f}s")
print(f"Custo médio: ${statistics.mean(custos):.4f}")
print(f"Custo total: ${sum(custos):.2f}")

Callback handler completo para produção

import time, structlog
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult

logger = structlog.get_logger()

class ProductionCallbackHandler(BaseCallbackHandler):
    def __init__(self, user_id: str, session_id: str):
        self.user_id = user_id
        self.session_id = session_id
        self._tool_start_times = {}
        self._llm_start_times = {}

    def on_llm_start(self, serialized, prompts, run_id, **kwargs):
        self._llm_start_times[str(run_id)] = time.time()
        logger.info("llm_start",
            model=serialized.get("name"),
            prompt_chars=sum(len(p) for p in prompts),
            user_id=self.user_id
        )

    def on_llm_end(self, response: LLMResult, run_id, **kwargs):
        duration = time.time() - self._llm_start_times.get(str(run_id), time.time())
        usage = response.llm_output.get("token_usage", {})
        logger.info("llm_end",
            duration_ms=int(duration * 1000),
            prompt_tokens=usage.get("prompt_tokens", 0),
            completion_tokens=usage.get("completion_tokens", 0),
            total_tokens=usage.get("total_tokens", 0),
            user_id=self.user_id,
            session_id=self.session_id
        )

    def on_tool_start(self, serialized, input_str, run_id, **kwargs):
        self._tool_start_times[str(run_id)] = time.time()
        logger.info("tool_start",
            tool_name=serialized.get("name"),
            input_preview=str(input_str)[:300],
            user_id=self.user_id
        )

    def on_tool_end(self, output, run_id, **kwargs):
        duration = time.time() - self._tool_start_times.get(str(run_id), time.time())
        logger.info("tool_end",
            duration_ms=int(duration * 1000),
            output_preview=str(output)[:300],
            user_id=self.user_id,
            session_id=self.session_id
        )

    def on_tool_error(self, error, run_id, **kwargs):
        logger.error("tool_error",
            error=str(error),
            user_id=self.user_id,
            session_id=self.session_id
        )

# Uso
handler = ProductionCallbackHandler(user_id="usr_123", session_id="sess_456")
result = executor.invoke(
    {"input": "Analise este contrato"},
    config={"callbacks": [handler]}
)