4.4 Avaluació i Observabilitat d'Agents

Per Què Avaluar i Observar?

Un agent que funciona en local pot fallar de maneres imprevistes en producció. L'avaluació ens diu si l'agent fa el que ha de fer; l'observabilitat ens explica per què falla quan falla. Sense aquestes dues pràctiques, un agent en producció és una caixa negra.

---

🎯 El Repte d'Avaluar Agents

Avaluar un agent és molt més difícil que avaluar una funció normal:

Funció clàssica:
  input → funció → output determinista
  Test: assert output == valor_esperat  ✅ Fàcil

Agent d'IA:
  input → [LLM + eines + memòria + múltiples passos] → output variable
  Test: ??? com avaluem "ha resolt correctament"?  🤔 Difícil

Els reptes principals:

• No-determinisme: el mateix input pot produir outputs lleugerament diferents
• Tasques obertes: no hi ha una única resposta "correcta"
• Cadenes llargues: un error inicial es propaga per tots els passos
• Eines externes: resultats de cerca web, APIs, bases de dades canvien

---

📏 Mètriques d'Avaluació

Mètriques Objectives

Quan hi ha una resposta correcta clara:

# Exemple: agent de Q&A sobre documents
def avaluar_exactitud(respostes_agent: list[str],
                      respostes_correctes: list[str]) -> float:
    """Percentatge de respostes exactament correctes."""
    correctes = sum(
        1 for agent, correcta in zip(respostes_agent, respostes_correctes)
        if agent.strip().lower() == correcta.strip().lower()
    )
    return correctes / len(respostes_correctes)

def avaluar_f1_token(prediccio: str, referencia: str) -> float:
    """F1 a nivell de tokens (útil per a respostes parcials correctes)."""
    tokens_pred = set(prediccio.lower().split())
    tokens_ref = set(referencia.lower().split())
    if not tokens_pred or not tokens_ref:
        return 0.0
    precisio = len(tokens_pred & tokens_ref) / len(tokens_pred)
    recall = len(tokens_pred & tokens_ref) / len(tokens_ref)
    if precisio + recall == 0:
        return 0.0
    return 2 * precisio * recall / (precisio + recall)

# Exemple d'ús
respostes = ["Barcelona", "Python 3.11", "1991"]
correctes = ["Barcelona", "Python 3.10", "1991"]

print(f"Exactitud: {avaluar_exactitud(respostes, correctes):.1%}")
print(f"F1 'Python 3.11' vs 'Python 3.10': "
      f"{avaluar_f1_token('Python 3.11', 'Python 3.10'):.2f}")

LLM-as-Judge

Per a respostes obertes, fem servir un LLM com a jutge:

import json
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from pydantic import BaseModel

llm_jutge = ChatOpenAI(model="gpt-4o", temperature=0)

class Puntuacio(BaseModel):
    puntuacio: int       # 1-5
    justificacio: str
    aspectes_positius: list[str]
    aspectes_negatius: list[str]

def llm_com_jutge(pregunta: str, resposta_agent: str,
                  resposta_referencia: str = "") -> Puntuacio:
    """Usa un LLM per avaluar la qualitat d'una resposta."""
    referencia_text = (f"\nResposta de referència:\n{resposta_referencia}"
                       if resposta_referencia else "")
    prompt = (
        f"Avalua la qualitat d'aquesta resposta d'un agent d'IA.\n\n"
        f"Pregunta: {pregunta}\n"
        f"Resposta de l'agent: {resposta_agent}{referencia_text}\n\n"
        "Puntua de l'1 al 5 on:\n"
        "1=Completament incorrecta | 2=Errors importants | 3=Acceptable\n"
        "4=Bona resposta | 5=Excel·lent i precisa\n\n"
        'Respon en JSON: {"puntuacio": 1-5, "justificacio": "...", '
        '"aspectes_positius": [...], "aspectes_negatius": [...]}'
    )
    resposta = llm_jutge.invoke([
        SystemMessage(content="Ets un avaluador expert i imparcial d'agents d'IA. "
                              "Respon SEMPRE en JSON vàlid."),
        HumanMessage(content=prompt)
    ])
    text = resposta.content.strip()
    if "```json" in text:
        text = text.split("```json")[1].split("```")[0].strip()
    elif "```" in text:
        text = text.split("```")[1].split("```")[0].strip()
    return Puntuacio(**json.loads(text))

# Exemple d'ús
resultat = llm_com_jutge(
    pregunta="Explica com funciona la memòria RAM",
    resposta_agent="La RAM és una memòria volàtil d'accés ràpid que "
                   "emmagatzema temporalment les dades que el processador "
                   "necessita en cada moment.",
    resposta_referencia="La RAM (Random Access Memory) és memòria volàtil "
                        "d'accés aleatori que el sistema usa per executar "
                        "programes actius."
)
print(f"Puntuació: {resultat.puntuacio}/5")
print(f"Justificació: {resultat.justificacio}")

Mètriques de RAG

Per a agents amb recuperació de documents (pip install ragas):

from ragas import evaluate
from ragas.metrics import (
    answer_relevancy,    # La resposta és rellevant a la pregunta?
    faithfulness,        # La resposta es basa en els documents recuperats?
    context_precision,   # Els documents recuperats són rellevants?
    context_recall,      # Es recuperen tots els documents necessaris?
)
from datasets import Dataset

dades = {
    "question": ["Què és LangGraph?", "Com funciona RAG?"],
    "answer": [
        "LangGraph és un framework per construir agents amb estat usant grafs.",
        "RAG recupera documents rellevants i els usa com a context per generar."
    ],
    "contexts": [
        ["LangGraph allows building stateful multi-actor applications..."],
        ["RAG combines retrieval with generation to ground LLM responses..."]
    ],
    "ground_truth": [
        "LangGraph és una llibreria per crear aplicacions LLM amb estat.",
        "RAG augmenta els LLM amb recuperació d'informació externa."
    ]
}

dataset = Dataset.from_dict(dades)
resultats = evaluate(dataset, metrics=[
    answer_relevancy, faithfulness, context_precision, context_recall
])
print(resultats)

---

🔍 Observabilitat: Veure Dins de l'Agent

Logging Bàsic amb LangChain

import logging
import time
from langchain_openai import ChatOpenAI
from langchain.callbacks.base import BaseCallbackHandler
from langchain_core.outputs import LLMResult

# ── CALLBACK PERSONALITZAT ─────────────────────────────────────
class AgentLogger(BaseCallbackHandler):
    """Captura tots els events de l'agent per a debugging."""

    def __init__(self):
        self.events = []
        self.temps_inici = None

    def on_llm_start(self, serialized, prompts, **kwargs):
        self.temps_inici = time.time()
        event = {
            "tipus": "LLM_START",
            "model": serialized.get("name", "desconegut"),
            "num_prompts": len(prompts),
            "timestamp": time.time()
        }
        self.events.append(event)
        logging.info(f"🤖 LLM iniciat | Model: {event['model']}")

    def on_llm_end(self, response: LLMResult, **kwargs):
        durada = time.time() - self.temps_inici
        tokens = response.llm_output.get("token_usage", {}) if response.llm_output else {}
        event = {
            "tipus": "LLM_END",
            "durada_s": round(durada, 2),
            "tokens_entrada": tokens.get("prompt_tokens", 0),
            "tokens_sortida": tokens.get("completion_tokens", 0),
            "cost_estimat_eur": tokens.get("total_tokens", 0) * 0.000003
        }
        self.events.append(event)
        logging.info(f"✅ LLM finalitzat | "
                     f"{event['durada_s']}s | "
                     f"{event['tokens_entrada']}+{event['tokens_sortida']} tokens | "
                     f"~{event['cost_estimat_eur']:.4f}€")

    def on_tool_start(self, serialized, input_str, **kwargs):
        event = {"tipus": "TOOL_START", "eina": serialized.get("name"), "input": input_str}
        self.events.append(event)
        logging.info(f"🔧 Eina: {event['eina']} | Input: {input_str[:60]}...")

    def on_tool_end(self, output, **kwargs):
        event = {"tipus": "TOOL_END", "output_len": len(str(output))}
        self.events.append(event)
        logging.info(f"📤 Eina finalitzada | Output: {len(str(output))} caràcters")

    def on_agent_action(self, action, **kwargs):
        logging.info(f"🧠 Acció: {action.tool} | Input: {str(action.tool_input)[:80]}")

    def resum(self) -> dict:
        """Genera un resum de l'execució."""
        llm_events = [e for e in self.events if e["tipus"] == "LLM_END"]
        tool_events = [e for e in self.events if e["tipus"] == "TOOL_START"]
        return {
            "total_crides_llm": len(llm_events),
            "total_eines_usades": len(tool_events),
            "temps_total_llm": sum(e["durada_s"] for e in llm_events),
            "tokens_totals": sum(
                e["tokens_entrada"] + e["tokens_sortida"] for e in llm_events
            ),
            "cost_total_eur": sum(e["cost_estimat_eur"] for e in llm_events),
        }

# ── ÚS DEL LOGGER ─────────────────────────────────────────────
logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s | %(message)s")

logger = AgentLogger()
llm = ChatOpenAI(model="gpt-4o-mini", callbacks=[logger])

# ... executar l'agent normalment ...

print("\n📊 RESUM D'EXECUCIÓ:")
resum = logger.resum()
for clau, valor in resum.items():
    print(f"  {clau}: {valor}")

---

LangSmith: Observabilitat Professional

LangSmith és la plataforma oficial de LangChain per a traces, avaluació i monitoratge en producció.

# Configuració de LangSmith (sense canviar el codi de l'agent!)
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__..."        # Clau de LangSmith
os.environ["LANGCHAIN_PROJECT"] = "curs-agents-ia"  # Nom del projecte

# A partir d'aquí, TOTES les execucions de LangChain/LangGraph
# es registren automàticament a LangSmith:
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub

llm = ChatOpenAI(model="gpt-4o-mini")
# ... definir eines i agent ...
# Cada execució apareixerà a https://smith.langchain.com amb:
# - Traça completa de cada pas
# - Tokens i cost
# - Latència per node
# - Errors i excepcions

Què ofereix LangSmith:

Funcionalitat	Descripció
Traces	Cada pas de l'agent amb inputs/outputs complets
Datasets	Conjunts de dades per a avaluació sistemàtica
Evaluators	Avaluadors automàtics per a cada run
Monitoring	Alertes quan la qualitat baixa
Playground	Provar prompts de forma interactiva

---

🧪 Suite d'Avaluació Sistemàtica

Un framework complet per avaluar agents de forma reproducible:

# Suite d'avaluació completa
# pip install langchain langchain-openai python-dotenv

import json
import time
from dataclasses import dataclass, field
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

@dataclass
class CasDeTest:
    id: str
    pregunta: str
    resposta_esperada: str
    criteris: list[str] = field(default_factory=list)
    # Criteris específics per a aquest test (opcional)

@dataclass
class ResultatTest:
    cas_id: str
    resposta_obtinguda: str
    puntuacio: float       # 0.0 - 1.0
    passes: bool
    temps_s: float
    tokens_usats: int
    feedback: str

class SuiteAvaluacio:
    """Framework per avaluar un agent de forma sistemàtica."""

    def __init__(self, agent_func, llm_jutge_model: str = "gpt-4o"):
        self.agent_func = agent_func
        self.llm_jutge = ChatOpenAI(model=llm_jutge_model, temperature=0)
        self.resultats: list[ResultatTest] = []

    def _avaluar_amb_llm(self, cas: CasDeTest,
                          resposta: str, temps: float) -> ResultatTest:
        """Usa un LLM per avaluar la resposta."""
        criteris_text = "\n".join(f"- {c}" for c in cas.criteris) if cas.criteris else ""
        prompt = (
            f"Pregunta: {cas.pregunta}\n"
            f"Resposta esperada: {cas.resposta_esperada}\n"
            f"Resposta obtinguda: {resposta}\n"
        )
        if criteris_text:
            prompt += f"Criteris específics:\n{criteris_text}\n"
        prompt += (
            "\nPuntua la resposta de 0.0 a 1.0 i indica si passa (true/false).\n"
            'Respon en JSON: {"puntuacio": 0.0-1.0, "passa": true/false, "feedback": "..."}'
        )

        resp = self.llm_jutge.invoke([
            SystemMessage(content="Ets un avaluador expert. Respon SEMPRE en JSON vàlid."),
            HumanMessage(content=prompt)
        ])
        text = resp.content.strip()
        if "```json" in text:
            text = text.split("```json")[1].split("```")[0].strip()
        elif "```" in text:
            text = text.split("```")[1].split("```")[0].strip()
        dades = json.loads(text)
        return ResultatTest(
            cas_id=cas.id,
            resposta_obtinguda=resposta,
            puntuacio=dades["puntuacio"],
            passes=dades["passa"],
            temps_s=temps,
            tokens_usats=0,  # Simplificat
            feedback=dades["feedback"]
        )

    def executar_cas(self, cas: CasDeTest) -> ResultatTest:
        """Executa un cas de test i retorna el resultat."""
        print(f"  🔄 Executant: {cas.id}...")
        inici = time.time()
        try:
            resposta = self.agent_func(cas.pregunta)
            temps = time.time() - inici
            resultat = self._avaluar_amb_llm(cas, resposta, temps)
        except Exception as e:
            temps = time.time() - inici
            resultat = ResultatTest(
                cas_id=cas.id, resposta_obtinguda=f"ERROR: {e}",
                puntuacio=0.0, passes=False, temps_s=temps,
                tokens_usats=0, feedback=f"Excepció: {type(e).__name__}"
            )
        estat = "✅" if resultat.passes else "❌"
        print(f"  {estat} {cas.id}: {resultat.puntuacio:.2f} | {resultat.temps_s:.1f}s")
        self.resultats.append(resultat)
        return resultat

    def executar_suite(self, casos: list[CasDeTest]) -> dict:
        """Executa tots els casos i retorna un resum."""
        print(f"\n🧪 Iniciant suite d'avaluació ({len(casos)} casos)...")
        for cas in casos:
            self.executar_cas(cas)

        aprovats = sum(1 for r in self.resultats if r.passes)
        puntuacio_mitjana = sum(r.puntuacio for r in self.resultats) / len(self.resultats)
        temps_total = sum(r.temps_s for r in self.resultats)

        resum = {
            "total_casos": len(casos),
            "aprovats": aprovats,
            "fallats": len(casos) - aprovats,
            "taxa_exit": aprovats / len(casos),
            "puntuacio_mitjana": puntuacio_mitjana,
            "temps_total_s": round(temps_total, 1),
        }

        print(f"\n{'='*50}")
        print(f"📊 RESUM: {aprovats}/{len(casos)} aprovats "
              f"({resum['taxa_exit']:.1%}) | "
              f"Puntuació: {puntuacio_mitjana:.2f}")
        print(f"{'='*50}")
        return resum


# ── EXEMPLE D'ÚS ──────────────────────────────────────────────
llm_simple = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def agent_simple(pregunta: str) -> str:
    """Agent de demostració."""
    resposta = llm_simple.invoke([HumanMessage(content=pregunta)])
    return resposta.content

casos_de_test = [
    CasDeTest(
        id="T001",
        pregunta="Quina és la capital de Catalunya?",
        resposta_esperada="Barcelona",
        criteris=["Ha de mencionar Barcelona", "Resposta breu i directa"]
    ),
    CasDeTest(
        id="T002",
        pregunta="Explica el concepte de recursivitat en programació",
        resposta_esperada="Funció que es crida a si mateixa amb un cas base",
        criteris=["Ha de mencionar 'cas base'",
                  "Ha de mencionar que la funció es crida a si mateixa"]
    ),
    CasDeTest(
        id="T003",
        pregunta="Quants dies té un any de traspàs?",
        resposta_esperada="366",
        criteris=["Ha de dir 366"]
    ),
]

suite = SuiteAvaluacio(agent_func=agent_simple)
resum = suite.executar_suite(casos_de_test)

---

📊 Panell de Monitoratge en Producció

# Monitoratge continu d'un agent en producció
import time
from collections import deque
from datetime import datetime

class MonitorAgent:
    """Monitoratge en temps real d'un agent en producció."""

    def __init__(self, finestra_temps: int = 3600):
        # Guarda les últimes N execucions (finestra lliscant)
        self.execucions = deque(maxlen=1000)
        self.finestra_temps = finestra_temps  # en segons
        self.alertes: list[str] = []

    def registrar(self, pregunta: str, resposta: str,
                  temps_s: float, tokens: int, error: bool = False):
        """Registra una execució de l'agent."""
        self.execucions.append({
            "timestamp": time.time(),
            "pregunta_len": len(pregunta),
            "resposta_len": len(resposta),
            "temps_s": temps_s,
            "tokens": tokens,
            "error": error,
        })
        self._comprovar_alertes(temps_s, error)

    def _comprovar_alertes(self, temps_s: float, error: bool):
        """Genera alertes si es detecten anomalies."""
        stats = self.estadistiques()
        if error:
            self.alertes.append(
                f"[{datetime.now():%H:%M:%S}] ❌ Error detectat"
            )
        if temps_s > 30:
            self.alertes.append(
                f"[{datetime.now():%H:%M:%S}] ⚠️ Latència alta: {temps_s:.1f}s"
            )
        if stats["taxa_error"] > 0.1:
            self.alertes.append(
                f"[{datetime.now():%H:%M:%S}] 🚨 Taxa d'error >10%: "
                f"{stats['taxa_error']:.1%}"
            )

    def estadistiques(self) -> dict:
        """Calcula estadístiques de la finestra temporal actual."""
        ara = time.time()
        recents = [e for e in self.execucions
                   if ara - e["timestamp"] < self.finestra_temps]
        if not recents:
            return {}
        return {
            "total_execucions": len(recents),
            "taxa_error": sum(1 for e in recents if e["error"]) / len(recents),
            "latencia_mitjana_s": sum(e["temps_s"] for e in recents) / len(recents),
            "latencia_p95_s": sorted(e["temps_s"] for e in recents)[
                int(len(recents) * 0.95)
            ],
            "tokens_per_execucio": sum(e["tokens"] for e in recents) / len(recents),
        }

    def informe(self):
        stats = self.estadistiques()
        print(f"\n📊 PANELL DE MONITORATGE (última hora)")
        print(f"  Execucions: {stats.get('total_execucions', 0)}")
        print(f"  Taxa d'error: {stats.get('taxa_error', 0):.1%}")
        print(f"  Latència mitjana: {stats.get('latencia_mitjana_s', 0):.2f}s")
        print(f"  Latència P95: {stats.get('latencia_p95_s', 0):.2f}s")
        print(f"  Tokens/execució: {stats.get('tokens_per_execucio', 0):.0f}")
        if self.alertes:
            print(f"\n🚨 ALERTES RECENTS:")
            for alerta in self.alertes[-5:]:
                print(f"  {alerta}")

---

🔄 Millora Contínua: el Cicle d'Avaluació

Producció
    ↓
Recollir traces amb errors o puntuació baixa
    ↓
Afegir als datasets d'avaluació
    ↓
Identificar patrons de fallada
    ↓
Millorar el prompt / eines / lògica
    ↓
Executar suite d'avaluació → confirmar millora
    ↓
Desplegar nova versió
    ↓
[Torna a Producció]

Regla dels 80/20 en Avaluació

El 80% dels errors prové del 20% de les causes. Identifica els casos de fallada més freqüents primer i resol-los abans de perseguir casos extrems.

---

✅ Activitats de Consolidació

Exercici 4.4.1 — Logger Personalitzat

Afegeix el callback AgentLogger a un agent ReAct de la unitat 4.2 i: 1. Analitza quantes crides al LLM fa per a 3 preguntes diferents 2. Calcula el cost estimat de cada consulta 3. Identifica quina eina es crida amb més freqüència

Exercici 4.4.2 — Suite d'Avaluació

Crea una suite d'avaluació de 5 casos per a un agent de Q&A senzill: 1. Defineix els casos amb preguntes de l'assignatura (ASIX/DAW) 2. Executa la suite i analitza els resultats 3. Modifica el prompt del sistema per millorar els casos que fallen

Exercici 4.4.3 — Comparació de Models

Usa la SuiteAvaluacio per comparar dos models (gpt-4o-mini vs gpt-4o): 1. Executa els mateixos 5 casos amb tots dos models 2. Compara puntuació, temps i cost estimat 3. Determina quin model ofereix millor relació qualitat/preu per al cas d'ús

---

📚 Referències

• LangSmith Documentation. docs.smith.langchain.com
• Es, S. et al. (2023). "RAGAS: Automated Evaluation of Retrieval Augmented Generation". arXiv:2309.15217
• Zheng, L. et al. (2023). "Judging LLM-as-a-Judge with MT-Bench and Chatbot Arena". arXiv:2306.05685
• Liu, Y. et al. (2023). "G-Eval: NLG Evaluation using GPT-4 with Better Human Alignment". arXiv:2303.16634