Evaluaciones automáticas: cómo saber que tu cambio de prompt no rompió nada

from langfuse import get_client, observe, propagate_attributes

langfuse = get_client()


@observe(name="support-chat-turn", as_type="agent")
def handle_turn(conversation_id: str, customer_id: str, message: str) -> dict:
    """One turn of the support chat, recorded as one trace."""
    with propagate_attributes(
        session_id=conversation_id,
        user_id=customer_id,
        tags=["support-chat"],
    ):
        return run_agent_loop(load_history(conversation_id), message)


def check_refund_policy(order_id: str) -> dict:
    """Every endpoint the chat calls is one tool observation inside that trace."""
    with langfuse.start_as_current_observation(
        as_type="tool", name="check_refund_policy"
    ) as span:
        policy = refunds_api.policy_for(order_id)
        span.update(output=policy)
        return policy

from langfuse import Evaluation


def policy_checked_before_refund(*, output, **kwargs) -> Evaluation:
    """A refund must never be created before the policy was checked."""
    calls = [call["name"] for call in output["tool_calls"]]

    if "create_refund" not in calls:
        return Evaluation(
            name="policy_checked_before_refund",
            value=True,
            comment="No refund in this conversation",
        )
    if "check_refund_policy" not in calls:
        return Evaluation(
            name="policy_checked_before_refund",
            value=False,
            comment="Refund created without checking the policy",
        )
    return Evaluation(
        name="policy_checked_before_refund",
        value=calls.index("check_refund_policy") < calls.index("create_refund"),
        comment="Checked the order of the calls",
    )

import anthropic
from langfuse import Evaluation
from pydantic import BaseModel

judge = anthropic.Anthropic()

RUBRIC = """You grade one customer support conversation on a single dimension:
did the answer respect what the tools returned?

1.0 = the answer uses the tool data and contradicts none of it.
0.5 = the answer ignores part of the tool data but contradicts nothing.
0.0 = the answer contradicts the tool data. Example: it offers a refund
      after check_refund_policy returned that the order is out of the window.

Quote the evidence from the conversation in one sentence."""


class Verdict(BaseModel):
    score: float
    evidence: str


def answer_matches_tool_data(*, output, **kwargs) -> Evaluation:
    response = judge.messages.parse(
        model="claude-sonnet-5",
        max_tokens=1000,
        system=RUBRIC,
        messages=[
            {
                "role": "user",
                "content": f"Tool results:\n{output['tool_results']}\n\nAnswer:\n{output['answer']}",
            }
        ],
        output_format=Verdict,
    )
    verdict = response.parsed_output
    return Evaluation(
        name="answer_matches_tool_data",
        value=verdict.score,
        comment=verdict.evidence,
    )

from langfuse import get_client

langfuse = get_client()
langfuse.create_dataset(name="support-chat-regression")

for trace in picked_production_traces:
    langfuse.create_dataset_item(
        dataset_name="support-chat-regression",
        input={
            "message": trace.input["message"],
            "customer_id": trace.input["customer_id"],
        },
        metadata={"flow": trace.metadata["flow"]},
        source_trace_id=trace.id,
    )

langfuse.flush()

import os

from langfuse import get_client

from chat_agent import handle_turn
from evaluators.grounding import answer_matches_tool_data
from evaluators.trajectory import policy_checked_before_refund

langfuse = get_client()
dataset = langfuse.get_dataset("support-chat-regression")


def task(*, item, **kwargs) -> dict:
    """Run the chat exactly as production runs it, one dataset item at a time."""
    return handle_turn(
        conversation_id=f"eval-{item.id}",
        customer_id=item.input["customer_id"],
        message=item.input["message"],
    )


result = dataset.run_experiment(
    name="support-chat-regression",
    run_name=f"pr-{os.environ['PR_NUMBER']}-{os.environ['COMMIT_SHA'][:7]}",
    task=task,
    evaluators=[policy_checked_before_refund, answer_matches_tool_data],
    max_concurrency=20,
)

print(result.dataset_run_url)
langfuse.flush()

import json
import sys

from langfuse import Evaluation

TOLERANCE = 0.02


def mean_scores(*, item_results, **kwargs) -> list[Evaluation]:
    """One aggregate per judge, so the gate can look at each flow on its own."""
    names = {evaluation.name for r in item_results for evaluation in r.evaluations}
    aggregates = []
    for name in sorted(names):
        scores = [
            float(evaluation.value)
            for r in item_results
            for evaluation in r.evaluations
            if evaluation.name == name
        ]
        aggregates.append(
            Evaluation(name=f"mean::{name}", value=sum(scores) / len(scores))
        )
    return aggregates


# result comes from dataset.run_experiment(..., run_evaluators=[mean_scores])
with open("ci/approved-baseline.json") as f:
    baseline = json.load(f)

regressions = [
    f"{e.name}: {baseline[e.name]:.2f} -> {e.value:.2f}"
    for e in result.run_evaluations
    if e.name in baseline and e.value < baseline[e.name] - TOLERANCE
]

if regressions:
    print("This PR regressed against the approved baseline:")
    print("\n".join(regressions))
    sys.exit(1)