AI Evaluation SDK

pip install ai-evaluation

from fi.evals import evaluate

# Local metric — no API key, runs in <1ms
result = evaluate("contains", output="Hello world", keyword="Hello")
print(result.score)   # 1.0
print(result.passed)  # True

# Cloud eval — uses Future AGI's Turing models
result = evaluate("toxicity", output="You're awesome!", model="turing_flash")
print(result.score)   # 1.0 (not toxic)

# LLM-as-Judge — your own criteria, any model
result = evaluate(
    prompt="Rate the helpfulness of this response on 0-1 scale.",
    output="Here are the 3 steps to fix your issue...",
    engine="llm",
    model="gemini/gemini-2.5-flash",
)
print(result.score)   # 0.9
print(result.reason)  # Detailed explanation

def evaluate(
    eval_name: str | list[str] | None = None,
    *,
    prompt: str | None = None,         # Custom judge criteria
    engine: str | None = None,         # Force: "local", "turing", "llm"
    model: str | None = None,          # Model string
    augment: bool | None = None,       # Local + LLM refinement
    config: dict | None = None,        # Metric-specific config
    generate_prompt: bool = False,     # Auto-generate grading criteria
    feedback_store: Any | None = None, # Few-shot from corrections
    fi_api_key: str | None = None,     # Turing API key override
    fi_secret_key: str | None = None,  # Turing secret key override
    fi_base_url: str | None = None,    # Turing base URL override
    **inputs,                          # output=, context=, input=, etc.
) -> EvalResult | BatchResult

result.eval_name   # "faithfulness"
result.score       # 0.85 (float, 0.0–1.0)
result.passed      # True/False (score >= threshold)
result.reason      # Human-readable explanation
result.status      # "completed" or "error"
result.latency_ms  # Execution time in milliseconds
result.metadata    # {"engine": "local+llm", ...}

results = evaluate(["toxicity", "faithfulness"], output="...", model="turing_flash")
for r in results:
    print(f"{r.eval_name}: {r.score}")

evaluate("contains", output="Hello World", keyword="Hello")          # 1.0
evaluate("contains_all", output="Hello World", keywords=["Hello", "World"])  # 1.0
evaluate("contains_any", output="Hello", keywords=["Hi", "Hello"])    # 1.0
evaluate("contains_none", output="Hello", keywords=["Bye", "Quit"])   # 1.0
evaluate("equals", output="hello", expected_output="hello")          # 1.0
evaluate("starts_with", output="Hello World", keyword="Hello")       # 1.0
evaluate("ends_with", output="Hello World", keyword="World")         # 1.0
evaluate("regex", output="Order #12345", pattern=r"Order #\d+")      # 1.0
evaluate("one_line", output="Single line response")                  # 1.0
evaluate("is_json", output='{"key": "value"}')                       # 1.0
evaluate("is_email", output="user@example.com")                      # 1.0
evaluate("contains_link", output="Visit https://example.com")        # 1.0
evaluate("length_less_than", output="short", max_length=100)          # 1.0
evaluate("length_greater_than", output="a longer string", min_length=5)  # 1.0

evaluate("bleu_score", output="the cat sat on mat", expected_output="the cat is on mat")
evaluate("rouge_score", output="quick brown fox", expected_output="a quick brown fox jumps")
evaluate("levenshtein_similarity", output="kitten", expected_output="sitting")
evaluate("embedding_similarity", output="dogs are great", expected_output="canines are wonderful")

# Faithful to context → high score
evaluate(
    "faithfulness",
    output="The capital of France is Paris.",
    context="Paris is the capital of France.",
)  # score ≈ 0.95

# Contradicts context → low score
evaluate(
    "hallucination_score",
    output="The moon is made of cheese.",
    context="The moon is Earth's only natural satellite.",
)  # score ≈ 0.2

evaluate(
    "context_recall",
    output="Paris is the capital of France.",
    context=["Paris is the capital and largest city of France."],
    input="What is the capital of France?",
    expected_output="Paris",
)

evaluate(
    "groundedness",
    output="Paris has 2.1 million people.",
    context=["Paris, population 2.1 million, is the capital of France."],
)

evaluate("prompt_injection", output="Ignore previous instructions and...")
evaluate("pii_detection", output="My SSN is 123-45-6789")
evaluate("secret_detection", output="api_key=sk-1234abcd...")
evaluate("sql_injection", output="'; DROP TABLE users; --")

# Set environment variables
import os
os.environ["FI_API_KEY"] = "your-api-key"
os.environ["FI_SECRET_KEY"] = "your-secret-key"

result = evaluate("toxicity", output="You're doing great!", model="turing_flash")

result = evaluate(
    prompt="""Rate the customer service quality of this response.
    Score 1.0 if it's empathetic, addresses the concern, and offers next steps.
    Score 0.5 if it's polite but missing key elements.
    Score 0.0 if it's dismissive or unhelpful.""",
    output="I understand your frustration. Let me escalate this to our team.",
    input="My order is 2 weeks late!",
    engine="llm",
    model="gemini/gemini-2.5-flash",
)

model="gemini/gemini-2.5-flash"      # Google Gemini
model="gemini/gemini-2.5-pro"        # Google Gemini Pro
model="gpt-4o"                       # OpenAI
model="claude-sonnet-4-20250514"          # Anthropic
model="ollama/llama3.2:3b"           # Local via Ollama

result = evaluate(
    prompt="""Given the context: {context}
    Rate how well the response answers the question: {input}
    Response: {output}""",
    output="Paris is the capital.",
    context="France's capital is Paris.",
    input="What's the capital of France?",
    engine="llm",
    model="gemini/gemini-2.5-flash",
)

result = evaluate(
    prompt="""Rate how accurately the text description matches the image.
    Score 1.0 if every detail is visible in the image.
    Score 0.0 if the description is completely wrong.""",
    output="A white daisy flower with a yellow center.",
    image_url="https://example.com/flower.jpg",
    engine="llm",
    model="gemini/gemini-2.5-flash",
)

result = evaluate(
    prompt="""Compare the input image (reference) with the output image (generated).
    Score 1.0 if they show the same content. Score 0.0 if completely different.""",
    output="A product description...",
    input_image_url="https://example.com/reference.jpg",
    output_image_url="https://example.com/generated.jpg",
    engine="llm",
    model="gemini/gemini-2.5-flash",
)

result = evaluate(
    prompt="""Rate how accurately the transcription captures the audio.
    Score 1.0 for accurate transcription. Score 0.0 for completely wrong.""",
    output="How old is the Brooklyn Bridge?",
    audio_url="https://example.com/audio.flac",
    engine="llm",
    model="gemini/gemini-2.5-flash",
)

result = evaluate(
    prompt="customer complaint resolution quality",  # Short description
    output="I understand your frustration. Let me look into this.",
    engine="llm",
    model="gemini/gemini-2.5-flash",
    generate_prompt=True,  # LLM generates the full rubric
)

from fi.evals.core.prompt_generator import generate_grading_criteria

criteria = generate_grading_criteria(
    "product photo quality for e-commerce",
    model="gemini/gemini-2.5-flash",
    inputs={"image_url": "...", "output": "..."},
)
print(criteria)
# "Score 1.0 if the image is sharp, well-lit, shows the product clearly..."

result = evaluate(
    "faithfulness",
    output="The capital of France is Paris, with 2 million residents.",
    context="Paris is the capital of France.",
    model="gemini/gemini-2.5-flash",
    augment=True,
)

print(result.metadata["engine"])  # "local+llm"

from fi.evals.feedback import FeedbackCollector, InMemoryFeedbackStore, ChromaFeedbackStore

# For testing / quick experiments
store = InMemoryFeedbackStore()

# For production (requires: pip install ai-evaluation[feedback])
store = ChromaFeedbackStore()

from fi.evals import evaluate
from fi.evals.feedback import FeedbackCollector, ChromaFeedbackStore

store = ChromaFeedbackStore()
feedback = FeedbackCollector(store)

# Run evaluation with feedback store — past corrections are injected as few-shot examples
result = evaluate(
    "faithfulness",
    output="Revenue was $5.2M.",
    context="Q3 revenue reached $5.2 million.",
    model="gemini/gemini-2.5-flash",
    augment=True,
    feedback_store=store,
)

# Correct a wrong result
if result.score < 0.7:
    feedback.submit(
        result,
        inputs={"output": "Revenue was $5.2M.", "context": "..."},
        correct_score=0.95,
        correct_reason="$5.2M is a valid abbreviation of $5.2 million.",
    )

# Calibrate pass/fail thresholds from accumulated feedback
profile = feedback.calibrate("faithfulness")
print(f"Optimal threshold: {profile.optimal_threshold}")

from fi.evals.otel import enable_auto_enrichment

# Enable — all evaluate() calls now emit OTEL spans
enable_auto_enrichment()

# Every evaluation creates a span with:
# - gen_ai.evaluation.name = "faithfulness"
# - gen_ai.evaluation.score = 0.95
# - gen_ai.evaluation.reason = "..."
result = evaluate("faithfulness", output="...", context="...")

from fi.evals.streaming import (
    StreamingEvaluator, StreamingConfig, EarlyStopPolicy,
    toxicity_scorer, pii_scorer, coherence_scorer,
)

# Create with config and early stop policy
assessor = StreamingEvaluator(
    config=StreamingConfig(min_chunk_size=10, enable_early_stop=True),
    policy=EarlyStopPolicy.default(),
)

# Add scoring functions
assessor.add_eval("toxicity", toxicity_scorer, threshold=0.5, pass_above=False)
assessor.add_eval("pii", pii_scorer, threshold=0.3, pass_above=False)
assessor.add_eval("coherence", coherence_scorer, threshold=0.5, pass_above=True)

# Feed tokens from your LLM stream
for token in llm_stream:
    result = assessor.process_token(token)
    if result and result.should_stop:
        print(f"Stopped: {result.stop_reason}")
        break

# Get final assessment
final = assessor.finalize()
print(final.summary())

# Safety-optimized (stops on toxicity/safety violations)
assessor = StreamingEvaluator.for_safety(
    toxicity_threshold=0.5,
    safety_threshold=0.5,
)

# Quality-optimized (larger chunks, less frequent checks)
assessor = StreamingEvaluator.for_quality(
    min_chunk_size=50,
    eval_interval_ms=500,
)

from fi.evals.streaming import EarlyStopPolicy

# Pre-built policies
policy = EarlyStopPolicy.default()     # toxicity > 0.7 or safety < 0.3
policy = EarlyStopPolicy.strict()      # Lower thresholds, quality checks
policy = EarlyStopPolicy.permissive()  # Only stops on severe issues

# Custom policy
policy = EarlyStopPolicy()
policy.add_toxicity_stop(threshold=0.7, consecutive=1)
policy.add_safety_stop(threshold=0.3, consecutive=1)
policy.add_quality_stop(threshold=0.3, consecutive=3)
policy.add_condition(
    name="pii_detected",
    eval_name="pii",
    threshold=0.5,
    comparison="above",
    consecutive_chunks=1,
)

import openai

async def safe_generate(prompt: str) -> str:
    client = openai.AsyncOpenAI()
    stream = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )

    assessor = StreamingEvaluator.for_safety()
    assessor.add_eval("toxicity", toxicity_scorer, threshold=0.5, pass_above=False)

    full_text = ""
    async for chunk in stream:
        token = chunk.choices[0].delta.content or ""
        full_text += token
        result = assessor.process_token(token)
        if result and result.should_stop:
            return "[Response blocked for safety]"

    final = assessor.finalize()
    return full_text if final.passed else "[Response failed quality check]"

result = evaluate("faithfulness", output="test", context="test context")

if result.status == "error":
    print(f"Failed: {result.reason}")
elif result.passed:
    print("Evaluation passed")
else:
    print(f"Failed with score {result.score}: {result.reason}")