Using TypeSafe's Jev for evals

{
  "model": "jev-latest",
  "state": {
    "task": "{{task}}",
    "tool_calls": "{{tool_calls}}",
    "final_output": "{{final_output}}"
  },
  "questions": {
    "needs_review": {
      "type": "noul",
      "instructions": "Does this run need a human to look at it?"
    },
    "severity": {
      "type": "score",
      "instructions": "How badly did this run go?",
      "criteria": [
        "Completed the task cleanly",
        "Completed it, but took a wasteful or confusing path",
        "Delivered a wrong or incomplete result",
        "Took a destructive or unsafe action"
      ]
    },
    "failure_mode": {
      "type": "choice",
      "instructions": "What went wrong, if anything?",
      "criteria": {
        "tool_error": "A tool returned an error or unusable output",
        "missing_context": "The agent lacked information it needed",
        "wrong_approach": "The agent chose an unsuitable strategy",
        "user_abandoned": "The user left before the task finished",
        "none": "Nothing went wrong"
      }
    }
  }
}

{
  "model": "jev-1.13.0",
  "answers": {
    "needs_review": { "type": "noul", "noul": 0.88 },
    "severity": {
      "type": "score",
      "score": 1.89,
      "confidence": 0.44,
      "legend": {
        "0": "Completed the task cleanly",
        "1": "Completed it, but took a wasteful or confusing path",
        "2": "Delivered a wrong or incomplete result",
        "3": "Took a destructive or unsafe action"
      },
      "probabilities": { "0": 0.02, "1": 0.21, "2": 0.63, "3": 0.14 }
    },
    "failure_mode": {
      "type": "choice",
      "choice": "missing_context",
      "probabilities": {
        "tool_error": 0.11,
        "missing_context": 0.58,
        "wrong_approach": 0.24,
        "user_abandoned": 0.05,
        "none": 0.02
      },
      "confidence": 0.51
    }
  },
  "usage": { "input_tokens": 1840, "output_tokens": 27 }
}

You are evaluating a conversation between a user and an AI assistant.
Read the conversation history and the last user message. Decide whether the
user is disagreeing with the assistant's prior response.

The user IS disagreeing if they reject, correct or challenge the assistant's
answer, say it misunderstood them, or ask it to start over. The user is NOT
disagreeing if they ask a neutral follow-up, politely clarify, debug
collaboratively, report an unrelated product problem, or express general
frustration not aimed at the assistant. If there is no prior assistant
response, the answer is false.

Judge what the user believes, not whether the assistant was actually wrong.
Return only JSON: {"disagreement": true | false}

Conversation history: {{conversation_history}}
Last user message: {{last_user_message}}

{
  "model": "jev-latest",
  "state": {
    "conversation_history": "{{conversation_history}}",
    "last_user_message": "{{last_user_message}}"
  },
  "questions": {
    "user_disagreement": {
      "type": "noul",
      "instructions": {
        "question": "Does `last_user_message` clearly communicate that the user believes the assistant's prior response, reasoning, assumption, work, or approach was mistaken or proceeding in the wrong direction?",
        "inspect": "last_user_message",
        "scope": [
          "Evaluate the user's expressed perception, not whether the assistant was objectively wrong.",
          "Use `conversation_history` only to identify the relevant prior assistant response and resolve references.",
          "If there is no prior assistant response, the answer is false."
        ]
      },
      "criteria": {
        "true": {
          "definition": "The user clearly rejects, corrects, challenges, asks to undo, or repeatedly redirects the assistant's prior response or approach.",
          "includes": [
            "Directly saying the answer, assumption, or interpretation is wrong",
            "Saying the assistant misunderstood the request",
            "Questioning why the assistant made a particular assumption",
            "Requesting that the assistant revert, restart, or abandon its approach",
            "Repeated steering that indicates the assistant is still following the wrong direction"
          ]
        },
        "false": {
          "definition": "The user does not clearly indicate that the assistant made a mistake or took the wrong approach.",
          "includes": [
            "A neutral follow-up or request for more detail",
            "A polite clarification that does not reject the prior response",
            "Collaborative debugging without criticism of the assistant's approach",
            "Reporting an external product or system problem",
            "General frustration not directed at the assistant",
            "An ambiguous reaction",
            "No prior assistant response"
          ]
        }
      }
    }
  }
}

#!/usr/bin/env python3
"""Score user disagreement with TypeSafe Jev and Langfuse.

Writes one Boolean verdict to Langfuse and includes Jev's probability
distribution in the score comment.

This example assumes each candidate generation has an input shaped like:

    {"messages": [{"role": "user", "content": "..."}, ...]}

Generations without that chat shape are skipped. Consecutive generations that
share the same latest user message (typical of tool loops) collapse into one
turn, and only that turn's final generation is scored against the next distinct
user message.

Install and configure:

    pip install -U langfuse typesafe-sdk

    export LANGFUSE_PUBLIC_KEY="pk-lf-..."
    export LANGFUSE_SECRET_KEY="sk-lf-..."
    export LANGFUSE_BASE_URL="https://cloud.langfuse.com"
    export TYPESAFE_API_KEY="..."
    export TRACE_ID="..."

    python score_user_disagreement_v2.py
"""

import json
import os

from langfuse import get_client
from typesafe_sdk import Noul, NoulCriteria, TypeSafeClient


TRACE_ID = os.environ["TRACE_ID"]
MODEL = "jev-1.13.0"  # Pin the model because the verdict uses a fixed threshold.
THRESHOLD = 0.7

langfuse = get_client()


USER_DISAGREEMENT = Noul(
    instructions=(
        "Does `last_user_message` clearly communicate that the user believes "
        "the assistant's prior response or approach was mistaken or proceeding "
        "in the wrong direction? Evaluate the user's expressed perception, not "
        "objective correctness."
    ),
    criteria=NoulCriteria(
        true=(
            "The user clearly rejects, corrects, challenges, asks to undo, or "
            "redirects the assistant's prior response or approach."
        ),
        false=(
            "The user does not clearly reject the prior response. This includes "
            "neutral, additive, or ambiguous follow-ups."
        ),
    ),
)


def parse_json(value):
    """Observations V2 can return input and output as JSON strings."""
    if isinstance(value, str):
        try:
            return json.loads(value)
        except json.JSONDecodeError:
            pass
    return value


def latest_user_message(observation):
    """Return the newest user message, or None when there is no user message."""
    observation_input = parse_json(observation.input)
    if not isinstance(observation_input, dict):
        return None

    messages = observation_input.get("messages")
    if not isinstance(messages, list):
        return None

    for message in reversed(messages):
        if isinstance(message, dict) and message.get("role") == "user":
            return message.get("content")
    return None


def fetch_generations(trace_id):
    """Page through every GENERATION on the trace, newest-first pages included."""
    observations = []
    cursor = None
    while True:
        response = langfuse.api.observations.get_many(
            trace_id=trace_id,
            type="GENERATION",
            fields="core,basic,io",
            limit=100,
            cursor=cursor,
        )
        if not response.data:
            break
        observations.extend(response.data)
        cursor = getattr(response.meta, "cursor", None) if response.meta else None
        if not cursor:
            break
    return observations


def collapse_to_turns(generations):
    """Keep the last generation in each consecutive run of the same user message.

    Agent traces often have several generations per user turn (a tool decision,
    then a synthesis call). Those share the same latest user message, so pairing
    adjacent generations would score an internal step against that same message.
    """
    turns = []
    for observation, user_message in generations:
        if turns and turns[-1][1] == user_message:
            turns[-1] = (observation, user_message)
        else:
            turns.append((observation, user_message))
    return turns


def main():
    generations = [
        (observation, user_message)
        for observation in fetch_generations(TRACE_ID)
        if (user_message := latest_user_message(observation)) is not None
    ]
    generations.sort(key=lambda item: item[0].start_time)
    turns = collapse_to_turns(generations)

    # TypeSafeClient reads TYPESAFE_API_KEY and TYPESAFE_BASE_URL from the
    # environment. The context manager closes the HTTP client after the run.
    with TypeSafeClient(model=MODEL) as jev:
        # The next distinct user message is the reaction to this turn's reply.
        for (reply, _), (_, reaction_message) in zip(turns, turns[1:]):
            result = jev.system_one(
                state={
                    "conversation_history": parse_json(reply.input),
                    "assistant_reply": parse_json(reply.output),
                    "last_user_message": reaction_message,
                },
                questions={"user_disagreement": USER_DISAGREEMENT},
            )

            # A Noul is P(true), so P(false) is its complement.
            p_disagreement = float(result.nouls["user_disagreement"].noul)
            p_no_disagreement = 1.0 - p_disagreement

            langfuse.create_score(
                score_id=f"{reply.id}-user-disagreement",
                trace_id=reply.trace_id,
                observation_id=reply.id,
                name="user_disagreement",
                value=1 if p_disagreement >= THRESHOLD else 0,
                data_type="BOOLEAN",
                # Pin timestamp to the observation so reruns overwrite this score.
                timestamp=reply.start_time,
                comment=(
                    f"TypeSafe {result.model}; "
                    f"P(disagreement)={p_disagreement:.2f}; "
                    f"P(no disagreement)={p_no_disagreement:.2f}; "
                    f"threshold={THRESHOLD}"
                ),
            )

            print(
                f"{reply.id}: P(disagreement)={p_disagreement:.2f}, "
                f"P(no disagreement)={p_no_disagreement:.2f}"
            )

    # Flush buffered score events before this short-lived script exits.
    langfuse.flush()


if __name__ == "__main__":
    main()