Evaluating an AI Coach with LLM-as-a-Judge

import langwatch
import pandas as pd
import json

# Authenticate with LangWatch
langwatch.login()

# Create a sample evaluation dataset (or load one from [LangWatch Datasets](https://docs.langwatch.ai/evaluations/experiments/sdk#use-langwatch-datasets)). In a real workflow, you would load this
# from a CSV or directly from LangWatch Datasets.
data = [
    {
        "input": "I feel stuck in my career and don't know what to do next.",
        "output": "That sounds challenging. What's one small step you think you could explore this week?",
        "conversation_history": "[]", # Start of conversation
        "expected_did_ask_question": True,
        "expected_did_not_loop": True,
    },
    {
        "input": "I'm not sure. I guess I could update my resume.",
        "output": "That sounds like a good starting point. What's one small step you could take to begin?",
        "conversation_history": json.dumps([
            {"role": "user", "content": "I feel stuck in my career and don't know what to do next."},
            {"role": "assistant", "content": "That sounds challenging. What's one small step you think you could explore this week?"}
        ]),
        # This output is repetitive, so we expect the 'looping' judge to fail.
        "expected_did_ask_question": True,
        "expected_did_not_loop": False,
    },
]
df = pd.DataFrame(data)
print("Sample evaluation data:")
print(df)

from pydantic import BaseModel
from openai import OpenAI

class JudgeAnswer(BaseModel):
    result: bool

def run_stacking_judge_llm(model_output: str) -> JudgeAnswer:
    """LLM judge: Does the response include an open-ended question?"""
    prompt = "You are an evaluator checking whether the AI coach response includes at least one open-ended question "

    response = client.responses.parse(
        model="gpt-5",
        instructions=prompt,
        response_format=JudgeAnswer,
        input={"role": "user", "content": f"AI Response: {model_output}"},
    )
    return response.output

# This judge needs the full conversation history to detect repetition.
def run_looping_judge_llm(model_output: str, history_json: str) -> bool:
    """LLM judge: Is the response a repetition of the previous assistant message?"""
    prompt = "You are an evaluator checking for repetition in an AI coach's behavior. "

    conversation_history = json.loads(history_json)
    messages = [{"role": "user", "content": f"Response: {model_output}"}]
    if conversation_history:
        messages.append({
            "role": "user",
            "content": f"Previous conversation:\n{json.dumps(conversation_history, indent=2)}"
        })

    response = client.responses.parse(
        model="gpt-5",
        instructions=prompt,
        response_format=JudgeAnswer,
        input=messages,
    )
    return response.output

# Initialize a new evaluation run in LangWatch
evaluation = langwatch.experiment.init("ai-coach-quality-v3-run-001")

# Use evaluation.loop() with evaluation.submit() for parallel execution.
# This speeds things up, as each judge can run independently.
for idx, row in evaluation.loop(df.iterrows(), threads=4):

    # Define a function to evaluate a single row from the dataset
    def evaluate_sample(index, data_row):
        # --- Run our custom judges ---
        actual_did_ask_question = run_stacking_judge(data_row["output"])
        actual_did_not_loop = run_looping_judge(data_row["output"], data_row["conversation_history"])

        # --- Log the result for the 'Stacking Judge' ---
        stacking_judge_passed = (actual_did_ask_question == data_row["expected_did_ask_question"])
        evaluation.log(
            "stacking_judge_passed",
            index=index,
            passed=stacking_judge_passed,
            data={
                "input": data_row["input"],
                "output": data_row["output"],
                "actual_value": actual_did_ask_question,
                "expected_value": data_row["expected_did_ask_question"],
            }
        )

        # --- Log the result for the 'Looping Judge' ---
        looping_judge_passed = (actual_did_not_loop == data_row["expected_did_not_loop"])
        evaluation.log(
            "looping_judge_passed",
            index=index,
            passed=looping_judge_passed,
            data={
                "input": data_row["input"],
                "output": data_row["output"],
                "actual_value": actual_did_not_loop,
                "expected_value": data_row["expected_did_not_loop"],
                "conversation_history": data_row["conversation_history"],
            }
        )

    # Submit the function to run in a separate thread
    evaluation.submit(evaluate_sample, idx, row)

print("\nEvaluation complete! Check your results in the LangWatch dashboard.")