¶`ciel.eval` — Evaluación y testing reproducible (Fase 18)

Capa de evaluación offline-safe por defecto: las métricas propias son deterministas y no requieren red ni extras. DeepEval/RAGAS/TruLens son opt-in vía el extra eval; si no están instalados, Evaluator degrada a métricas propias.

¶Componentes

Símbolo	Tipo	Descripción
`EvalCase`	dataclass	Un caso: `query`, `expected`, `context`, `gold`, `metadata`.
`Evaluator`	clase	Corre un dataset sobre un agente/callable y acumula KPIs.
`Evaluator.run` / `arun`	método	Ejecuta (sync/async) y devuelve `List[EvalResult]`.
`Evaluator.kpis`	método	Agrega `pass_rate` y medias de cada métrica.
`Evaluator.export`	método	Exporta `results.json` (kpis + resultados).
`load_dataset`	función	Carga un dataset YAML de casos.
`exact_match`, `contains`, `f1_token`	métricas	Coincidencia cerrada.
`faithfulness`, `context_relevance`	métricas	RAG (tokens comunes respuesta↔contexto; usa `Retriever` si se pasa).
`answer_relevance`	métrica	Heurística de diagnóstico (no gating por defecto).

¶Uso mínimo

import asyncio
from ciel.eval import Evaluator, load_dataset
from ciel.providers import MockProvider
from ciel.runtime import ChatRequest, ChatMessage

cases = load_dataset("tests/eval/smoke.yaml")

async def agent(query, **kw):
    p = MockProvider(mode="fixed", response="París")
    r = await p.complete(ChatRequest(messages=[ChatMessage(role="user", content=query)]))
    return r.choice.message.content

ev = Evaluator(agent=agent)
ev.run(cases, threshold=0.8)
print(ev.kpis())
ev.export("results.json")

¶CLI

# Correr un dataset con MockProvider (offline)
ciel evaluate run --dataset tests/eval/smoke.yaml --provider mock --threshold 0.8

# Modo mapa explícito
ciel evaluate run --dataset ds.yaml --provider mock \
  --mock-map "capital de Francia=París,2+2=4"

# Regression gate contra un baseline
ciel evaluate regression --baseline results.json --dataset ds.yaml --provider mock

# Red-teaming (prompt injection / fuga de tenant) con MockProvider
ciel evaluate redteam --dataset adversarial.yaml --provider mock

El extra eval habilita métricas de terceros (se degradan a None si no están instaladas):

pip install "mana-ciel[eval]"

¶ `ciel.eval`

ciel.eval — capa de evaluación y testing reproducible (Fase 18).

Offline-safe por defecto: métricas deterministas propias funcionan sin red ni extras. DeepEval/RAGAS/TruLens son opt-in vía el extra eval; si no están instalados, Evaluator degrada a métricas propias (igual que LiteLLM/RAG).

¶ `all = ['EvalCase', 'Evaluator', 'EvalResult', 'exact_match', 'contains', 'f1_token', 'faithfulness', 'context_relevance', 'answer_relevance', 'load_dataset']` `module-attribute`

¶ `EvalCase` `dataclass`

Un caso de evaluación.

Parameters:

Name	Type	Description	Default
`query`	`str`	entrada que se le pasa al agente/evaluable.	required
`expected`	`Optional[str]`	respuesta esperada (para métricas cerradas como `exact_match` o `contains`). Opcional si solo se evalúa `faithfulness`/`answer_relevance`.	`None`
`context`	`Optional[str]`	contexto RAG recuperado (para `faithfulness`/`context_relevance`).	`None`
`gold`	`Optional[str]`	respuesta "gold" alternativa para `f1_token` (por defecto usa `expected`).	`None`
`metadata`	`Dict[str, Any]`	metadatos libres (tenant, tags, etc.).	`dict()`

Source code in src/ciel/eval/eval_case.py

@dataclass
class EvalCase:
    """Un caso de evaluación.

    Args:
        query: entrada que se le pasa al agente/evaluable.
        expected: respuesta esperada (para métricas cerradas como ``exact_match``
            o ``contains``). Opcional si solo se evalúa ``faithfulness``/``answer_relevance``.
        context: contexto RAG recuperado (para ``faithfulness``/``context_relevance``).
        gold: respuesta "gold" alternativa para ``f1_token`` (por defecto usa ``expected``).
        metadata: metadatos libres (tenant, tags, etc.).
    """

    query: str
    expected: Optional[str] = None
    context: Optional[str] = None
    gold: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

    def gold_text(self) -> Optional[str]:
        """Texto gold efectivo (``gold`` o ``expected``)."""
        return self.gold if self.gold is not None else self.expected

¶ `gold_text() -> Optional[str]`

Texto gold efectivo (gold o expected).

Source code in src/ciel/eval/eval_case.py

def gold_text(self) -> Optional[str]:
    """Texto gold efectivo (``gold`` o ``expected``)."""
    return self.gold if self.gold is not None else self.expected

¶ `EvalResult` `dataclass`

Resultado de un caso individual.

Source code in src/ciel/eval/evaluator.py

@dataclass
class EvalResult:
    """Resultado de un caso individual."""

    query: str
    response: str
    expected: Optional[str]
    scores: Dict[str, float] = field(default_factory=dict)
    passed: bool = True
    error: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

¶ `Evaluator`

Evalúa un agente/callable sobre un dataset y acumula KPIs.

Source code in src/ciel/eval/evaluator.py

class Evaluator:
    """Evalúa un agente/callable sobre un dataset y acumula KPIs."""

    def __init__(
        self,
        *,
        agent: Any = None,
        tenant_id: Optional[str] = None,
        use_third_party: bool = False,
        retriever: Any = None,
    ) -> None:
        self.agent = agent
        self.tenant_id = tenant_id
        self.use_third_party = use_third_party
        self.retriever = retriever
        self.results: List[EvalResult] = []

    # -- invocación del evaluable -------------------------------------------
    async def _call(self, case: EvalCase) -> str:
        agent = self.agent
        if agent is None:
            raise ValueError("Evaluator requiere un agent/callable (pasar agent=...)")
        # ciel.Agent
        if hasattr(agent, "arun") and callable(getattr(agent, "arun")):
            resp = await agent.arun(case.query, tenant_id=self.tenant_id)
            text = getattr(resp, "text", None)
            if text is None and hasattr(resp, "raw"):
                text = resp.raw.response.choice.message.text()
            return text or ""
        # corutina
        if asyncio.iscoroutinefunction(agent):
            return await agent(case.query, tenant_id=self.tenant_id) or ""
        # función síncrona / callable
        return agent(case.query) or ""

    # -- métricas -----------------------------------------------------------
    def _score(self, case: EvalCase, response: str) -> Dict[str, float]:
        scores: Dict[str, float] = {}
        gold = case.gold_text()
        if gold is not None:
            scores["exact_match"] = M.exact_match(response, gold)
            scores["contains"] = M.contains(response, gold)
            scores["f1_token"] = M.f1_token(response, gold)
        if case.context is not None:
            scores["faithfulness"] = M.faithfulness(response, case.context)
            scores["context_relevance"] = M.context_relevance(
                case.query, case.context, retriever=self.retriever, tenant_id=self.tenant_id
            )
        scores["answer_relevance"] = M.answer_relevance(response, case.query)

        # Opt-in terceros (extra eval). Degradan a None si no instalados.
        if self.use_third_party and case.context is not None:
            de = M.deepeval_faithfulness(response, case.context)
            if de is not None:
                scores["deepeval_faithfulness"] = de
            rg = M.ragas_faithfulness(case.query, response, case.context)
            if rg is not None:
                scores["ragas_faithfulness"] = rg
        return scores

    # -- run ----------------------------------------------------------------
    async def arun(self, dataset: Sequence[EvalCase], *, threshold: float = 0.8) -> List[EvalResult]:
        out: List[EvalResult] = []
        for case in dataset:
            try:
                response = await self._call(case)
                scores = self._score(case, response)
                # ``answer_relevance`` es una métrica de diagnóstico (heurística
                # sobre tokens compartidos con la query); por defecto NO debe
                # hundir el caso en respuestas cerradas. El gating usa las demás.
                gating = {k: v for k, v in scores.items() if k != "answer_relevance"}
                passed = all(v >= threshold for v in gating.values()) if gating else True
                out.append(
                    EvalResult(
                        query=case.query,
                        response=response,
                        expected=case.expected,
                        scores=scores,
                        passed=passed,
                        metadata=dict(case.metadata),
                    )
                )
            except Exception as exc:  # degradación: el caso falla, no el eval
                out.append(
                    EvalResult(
                        query=case.query,
                        response="",
                        expected=case.expected,
                        scores={},
                        passed=False,
                        error=f"{type(exc).__name__}: {exc}",
                        metadata=dict(case.metadata),
                    )
                )
        self.results = out
        return out

    def run(self, dataset: Sequence[EvalCase], *, threshold: float = 0.8) -> List[EvalResult]:
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            return asyncio.run(self.arun(dataset, threshold=threshold))
        raise RuntimeError("Evaluator.run() no puede usarse dentro de un event loop; usa arun()")

    # -- KPIs ---------------------------------------------------------------
    def kpis(self) -> Dict[str, Any]:
        if not self.results:
            return {"n": 0, "passed": 0, "failed": 0, "pass_rate": 0.0, "metrics": {}}
        n = len(self.results)
        passed = sum(1 for r in self.results if r.passed)
        agg: Dict[str, List[float]] = {}
        for r in self.results:
            for k, v in r.scores.items():
                agg.setdefault(k, []).append(v)
        metrics = {k: (sum(v) / len(v) if v else 0.0) for k, v in agg.items()}
        return {
            "n": n,
            "passed": passed,
            "failed": n - passed,
            "pass_rate": passed / n,
            "metrics": metrics,
        }

    def export(self, path: str) -> Dict[str, Any]:
        """Exporta resultados + KPIs a ``results.json``."""
        payload = {
            "kpis": self.kpis(),
            "results": [asdict(r) for r in self.results],
        }
        os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True)
        with open(path, "w", encoding="utf-8") as fh:
            json.dump(payload, fh, indent=2, ensure_ascii=False)
        return payload

¶ `export(path: str) -> Dict[str, Any]`

Exporta resultados + KPIs a results.json.

Source code in src/ciel/eval/evaluator.py

def export(self, path: str) -> Dict[str, Any]:
    """Exporta resultados + KPIs a ``results.json``."""
    payload = {
        "kpis": self.kpis(),
        "results": [asdict(r) for r in self.results],
    }
    os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True)
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(payload, fh, indent=2, ensure_ascii=False)
    return payload