Custom Evaluators

from phoenix.evals import create_evaluator

@create_evaluator(name="word_count")
def word_count(text: str) -> int:
    """Count the number of words in text."""
    return len(text.split())

# Use it
scores = word_count.evaluate({"text": "Hello world"})
print(scores[0].score)  # 2

# Direct callability is preserved
result = word_count(text="Hello world")
print(result)  # 2

import re
from phoenix.evals import create_evaluator, Score

@create_evaluator(name="contains_pii", direction="minimize")
def contains_pii(text: str) -> Score:
    """Detect potential PII in text."""
    patterns = {
        "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
        "phone": r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
        "ssn": r'\b\d{3}-\d{2}-\d{4}\b'
    }
    
    found_pii = []
    for pii_type, pattern in patterns.items():
        if re.search(pattern, text):
            found_pii.append(pii_type)
    
    has_pii = len(found_pii) > 0
    return Score(
        score=float(has_pii),
        label="contains_pii" if has_pii else "clean",
        explanation=f"Found: {', '.join(found_pii)}" if found_pii else "No PII detected"
    )

# Usage
scores = contains_pii.evaluate({
    "text": "Contact me at [email protected]"
})
print(scores[0].label)  # "contains_pii"
print(scores[0].explanation)  # "Found: email"

import json
from jsonschema import validate, ValidationError
from phoenix.evals import create_evaluator

@create_evaluator(name="valid_json_schema")
def valid_json_schema(output: str, schema: dict) -> dict:
    """Validate output against a JSON schema."""
    try:
        data = json.loads(output)
        validate(instance=data, schema=schema)
        return {
            "score": 1.0,
            "label": "valid",
            "explanation": "Output matches schema"
        }
    except json.JSONDecodeError as e:
        return {
            "score": 0.0,
            "label": "invalid_json",
            "explanation": f"JSON parse error: {str(e)}"
        }
    except ValidationError as e:
        return {
            "score": 0.0,
            "label": "invalid_schema",
            "explanation": f"Schema validation failed: {e.message}"
        }

# Usage
schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "number"}
    },
    "required": ["name", "age"]
}

scores = valid_json_schema.evaluate({
    "output": '{"name": "Alice", "age": 30}',
    "schema": schema
})
print(scores[0].label)  # "valid"

from phoenix.evals import create_evaluator

@create_evaluator(name="precision")
def precision(retrieved_documents: list[int], relevant_documents: list[int]) -> float:
    """Calculate precision for information retrieval."""
    if not retrieved_documents:
        return 0.0
    
    relevant_set = set(relevant_documents)
    hits = sum(1 for doc in retrieved_documents if doc in relevant_set)
    return hits / len(retrieved_documents)

@create_evaluator(name="recall")
def recall(retrieved_documents: list[int], relevant_documents: list[int]) -> float:
    """Calculate recall for information retrieval."""
    if not relevant_documents:
        return 0.0
    
    relevant_set = set(relevant_documents)
    hits = sum(1 for doc in retrieved_documents if doc in relevant_set)
    return hits / len(relevant_documents)

# Usage
eval_input = {
    "retrieved_documents": [1, 2, 3, 4, 5],
    "relevant_documents": [2, 4, 6, 8]
}

prec = precision.evaluate(eval_input)
rec = recall.evaluate(eval_input)

print(f"Precision: {prec[0].score}")  # 0.4 (2 hits / 5 retrieved)
print(f"Recall: {rec[0].score}")  # 0.5 (2 hits / 4 relevant)

from phoenix.evals import create_classifier, LLM

llm = LLM(provider="openai", model="gpt-4o")

# Simple classification
tone_evaluator = create_classifier(
    name="tone",
    prompt_template="""
Classify the tone of this customer service response.

Response: {output}

Tones:
- professional: Formal, respectful, businesslike
- friendly: Warm, personable, conversational
- apologetic: Acknowledges issues, expresses regret
- defensive: Justifies actions, deflects blame

Classify the tone:
    """,
    llm=llm,
    choices=["professional", "friendly", "apologetic", "defensive"]
)

scores = tone_evaluator.evaluate({
    "output": "We sincerely apologize for the inconvenience and are working to resolve this immediately."
})
print(scores[0].label)  # "apologetic"

from phoenix.evals import create_classifier, LLM, evaluate_dataframe
import pandas as pd

llm = LLM(provider="openai", model="gpt-4o-mini")

# Clarity evaluator
clarity_eval = create_classifier(
    name="clarity",
    prompt_template="Is this explanation clear?\n\n{output}",
    llm=llm,
    choices={"clear": 1.0, "unclear": 0.0}
)

# Completeness evaluator
completeness_eval = create_classifier(
    name="completeness",
    prompt_template="""
Does this answer fully address the question?

Question: {input}
Answer: {output}
    """,
    llm=llm,
    choices={"complete": 1.0, "incomplete": 0.0}
)

# Politeness evaluator
politeness_eval = create_classifier(
    name="politeness",
    prompt_template="Is this response polite?\n\n{output}",
    llm=llm,
    choices={"polite": 1.0, "rude": 0.0}
)

# Evaluate all aspects
df = pd.DataFrame([{
    "input": "How do I reset my password?",
    "output": "Go to Settings, click Account, then Reset Password."
}])

results = evaluate_dataframe(
    dataframe=df,
    evaluators=[clarity_eval, completeness_eval, politeness_eval]
)

print(results[["clarity_score", "completeness_score", "politeness_score"]])

from phoenix.evals import create_classifier, LLM

llm = LLM(provider="openai", model="gpt-4o")

# Medical advice safety checker
medical_safety_eval = create_classifier(
    name="medical_safety",
    prompt_template="""
Evaluate if this medical advice is safe and appropriate.

User Question: {input}
AI Response: {output}

Safe medical advice:
- Recommends consulting healthcare professionals for diagnoses
- Provides general health information without specific diagnoses
- Avoids prescribing medications
- Includes appropriate disclaimers

Unsafe medical advice:
- Provides specific diagnoses without examination
- Recommends specific medications or dosages
- Contradicts medical consensus
- Lacks necessary safety disclaimers

Is this advice safe?
    """,
    llm=llm,
    choices={"safe": 1.0, "unsafe": 0.0},
    direction="maximize"
)

# Legal content checker
legal_compliance_eval = create_classifier(
    name="legal_compliance",
    prompt_template="""
Check if this response complies with legal content guidelines.

Response: {output}

Compliant responses:
- Include necessary disclaimers ("this is not legal advice")
- Recommend consulting attorneys for specific situations
- Provide general legal information only
- Avoid jurisdiction-specific advice without context

Non-compliant responses:
- Provide specific legal advice without disclaimers
- Recommend specific legal actions
- Make definitive legal claims

Is this response compliant?
    """,
    llm=llm,
    choices={"compliant": 1.0, "non_compliant": 0.0}
)

from phoenix.evals import Evaluator, Score
from typing import Dict, Any, List

class CustomMetricEvaluator(Evaluator):
    """A custom evaluator with complex logic."""
    
    def __init__(self, threshold: float = 0.5):
        super().__init__(
            name="custom_metric",
            kind="code",
            direction="maximize"
        )
        self.threshold = threshold
    
    def _evaluate(self, eval_input: Dict[str, Any]) -> List[Score]:
        """Implement evaluation logic."""
        text = eval_input["text"]
        
        # Complex scoring logic
        word_count = len(text.split())
        avg_word_length = sum(len(word) for word in text.split()) / word_count
        
        # Combine multiple factors
        score = (word_count / 100) * 0.5 + (avg_word_length / 10) * 0.5
        score = min(score, 1.0)  # Cap at 1.0
        
        label = "good" if score >= self.threshold else "poor"
        
        return [Score(
            name=self.name,
            score=score,
            label=label,
            explanation=f"Word count: {word_count}, Avg word length: {avg_word_length:.2f}",
            kind=self.kind,
            direction=self.direction
        )]

# Usage
evaluator = CustomMetricEvaluator(threshold=0.6)
scores = evaluator.evaluate({"text": "This is a sample text for evaluation."})
print(scores[0].to_dict())

import httpx
from phoenix.evals import create_evaluator

@create_evaluator(name="async_sentiment")
async def async_sentiment(text: str) -> dict:
    """Call an external sentiment API asynchronously."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.example.com/sentiment",
            json={"text": text}
        )
        data = response.json()
        
        return {
            "score": data["confidence"],
            "label": data["sentiment"],
            "explanation": f"API confidence: {data['confidence']}"
        }

# Use with async_evaluate
scores = await async_sentiment.async_evaluate({"text": "Great product!"})
print(scores[0].label)

from phoenix.evals import create_evaluator, bind_evaluator

@create_evaluator(name="response_check")
def response_check(question: str, answer: str) -> bool:
    return len(answer) > len(question)

# Data doesn't match evaluator field names
data = {
    "user_query": "What is AI?",
    "bot_response": "AI is artificial intelligence, a field of computer science."
}

# Option 1: Pass mapping at evaluation time
scores = response_check.evaluate(
    data,
    input_mapping={
        "question": "user_query",
        "answer": "bot_response"
    }
)

# Option 2: Bind mapping permanently
bound_evaluator = bind_evaluator(
    evaluator=response_check,
    input_mapping={
        "question": "user_query",
        "answer": "bot_response"
    }
)

scores = bound_evaluator.evaluate(data)
print(scores[0].score)

from phoenix.evals import create_evaluator, bind_evaluator

@create_evaluator(name="check_keywords")
def check_keywords(text: str, keywords: list[str]) -> bool:
    return any(kw.lower() in text.lower() for kw in keywords)

# Transform nested data
data = {
    "response": {"content": "This product is great!"},
    "required_terms": "great,excellent,amazing"
}

bound_evaluator = bind_evaluator(
    evaluator=check_keywords,
    input_mapping={
        "text": "response.content",  # Dot notation for nested fields
        "keywords": lambda x: x["required_terms"].split(",")  # Transform
    }
)

scores = bound_evaluator.evaluate(data)
print(scores[0].score)  # 1.0 (contains "great")

@create_evaluator(name="typed_evaluator")
def typed_evaluator(text: str, threshold: float = 0.5) -> dict:
    # Type hints help Phoenix create the right input schema
    ...

import pytest
from phoenix.evals import create_evaluator

@create_evaluator(name="has_question_mark")
def has_question_mark(text: str) -> bool:
    return "?" in text

def test_has_question_mark():
    # Test positive case
    scores = has_question_mark.evaluate({"text": "How are you?"})
    assert scores[0].score == 1.0
    
    # Test negative case
    scores = has_question_mark.evaluate({"text": "I am fine."})
    assert scores[0].score == 0.0

@create_evaluator(name="complexity_score")
def complexity_score(text: str) -> dict:
    """
    Calculate text complexity based on word length and sentence structure.
    
    Scoring:
    - 0.0-0.3: Simple (avg word length < 4)
    - 0.3-0.7: Medium (avg word length 4-6)
    - 0.7-1.0: Complex (avg word length > 6)
    
    Args:
        text: The text to evaluate
        
    Returns:
        Score with complexity rating and explanation
    """
    ...