LLM as a Jury: How To Implement

!pip install -qqq litellm arize-otel arize-phoenix-evals pandas openai openinference-instrumentation-litellm


			

import os

# Add Arize info to see Judge Call Tracing
os.environ["SPACE_ID"] = ''
os.environ["API_KEY"] = ''

# Add API Keys for Judge Models
os.environ["OPENAI_API_KEY"] = ''
os.environ["ANTHROPIC_API_KEY"] = ''


			

import os
from typing import List, Dict, Any
from concurrent.futures import ThreadPoolExecutor
from collections import Counter
import pandas as pd
from phoenix.evals import llm_classify, LiteLLMModel
from arize.otel import register
from openinference.instrumentation.litellm import LiteLLMInstrumentor
from datetime import datetime

# Register OpenTelemetry with Arize
tracer_provider = register(
    space_id=os.environ["SPACE_ID"],
    api_key=os.environ["API_KEY"],
    project_name=f"llm-jury-evaluation",
)

# Instrument LiteLLM
LiteLLMInstrumentor().instrument(tracer_provider=tracer_provider)

# Define your responses to evaluate
my_responses = [
    {
        "input": "What is the capital of France?",
        "reference": "Paris is the capital of France.",
        "output": "Paris",
    },
    # Add more examples here...
]

# Convert responses to DataFrame for Phoenix evals
df = pd.DataFrame(my_responses)

# Define the judge models
JUDGE_MODELS = [
    "claude-3-5-sonnet-latest",
    "gpt-3.5-turbo",
    "gpt-4o-mini",
]

# Define the judge prompt template
JUDGE_PROMPT_TEMPLATE = """You are an expert judge evaluating the quality of AI responses.

Task: Evaluate if the AI's response is correct and helpful based on the input and reference.

Input: {input}
Reference Answer: {reference}
AI Response: {output}

Evaluate the response based on:
1. Correctness: Is the information accurate and matches the reference?
2. Helpfulness: Is the response clear, complete, and useful?

Respond with one of the following labels:
- "correct": The response is both correct and helpful
- "incorrect": The response is incorrect or misleading

Provide a brief explanation for your judgment."""

# Define valid response options (rails)
JUDGE_RAILS = ["correct", "incorrect"]

def run_judge(model: str, df: pd.DataFrame) -> Dict[str, List[str]]:
    """Run a single judge model on all responses using Phoenix evals."""
    # Initialize the judge model with tracing
    llm_judge_model = LiteLLMModel(
        model=model,
    )

    # Run evaluation using Phoenix with tracing
    eval_results = llm_classify(
        dataframe=df,
        template=JUDGE_PROMPT_TEMPLATE,
        model=llm_judge_model,
        provide_explanation=True,
        rails=JUDGE_RAILS,
    )

    return {
        "labels": eval_results["label"].tolist(),
        "explanations": eval_results["explanation"].tolist()
    }

def safe_majority_vote(votes: List[str]) -> str:
    """Aggregate votes with tie handling."""
    count = Counter(votes)
    most_common = count.most_common()
    if len(most_common) == 1 or most_common[0]
[1] > most_common[1]
[1]:
        return most_common[0]
[0]
    return "tie"

def main():
    # Run all judges in parallel
    with ThreadPoolExecutor(max_workers=len(JUDGE_MODELS)) as pool:
        juror_results = list(pool.map(
            lambda model: run_judge(model, df),
            JUDGE_MODELS
        ))

    # Extract labels from each judge's results
    juror_votes = [result["labels"] for result in juror_results]

    # Aggregate votes for each response
    verdicts = [
        safe_majority_vote(votes)
        for votes in zip(*juror_votes)
    ]

    # Prepare explanations for logging
    juror_explanations = {
        model: results["explanations"]
        for model, results in zip(JUDGE_MODELS, juror_results)
    }

    print(f"Verdicts: {verdicts}")
    print(f"Explanations: {juror_explanations}")

if __name__ == "__main__":
    main()