Async & Batch Evaluations

A single synchronous call blocks until the result is ready.

from fi.evals import evaluate

result = evaluate(
    "groundedness",
    output="The Eiffel Tower is in Paris, France.",
    context="The Eiffel Tower is a wrought-iron lattice tower on the Champ de Mars in Paris.",
    model="turing_small",
)

print(f"Score: {result.score}  Passed: {result.passed}")
print(f"Reason: {result.reason}")

You should see:

Score: 1.0  Passed: True
Reason: The output is fully supported by the provided context.

This is fine for single items. For 50+ items it becomes slow because each call waits for the server response before the next one starts.

Use Evaluator.evaluate() with is_async=True. The call returns immediately with an eval_id you can poll later.

import os
from fi.evals import Evaluator

evaluator = Evaluator(
    fi_api_key=os.environ["FI_API_KEY"],
    fi_secret_key=os.environ["FI_SECRET_KEY"],
)

result = evaluator.evaluate(
    eval_templates="groundedness",
    inputs={
        "output": "The Eiffel Tower is in Paris, France.",
        "context": "The Eiffel Tower is a wrought-iron lattice tower on the Champ de Mars in Paris.",
    },
    model_name="turing_small",
    is_async=True,
)

eval_id = result.eval_results[0].eval_id
print(f"Submitted async eval (eval_id: {eval_id})")

You should see:

Submitted async eval (eval_id: abc123-def456-...)

Use get_eval_result(eval_id) to retrieve the result once processing completes.

import time

for attempt in range(15):
    poll_result = evaluator.get_eval_result(eval_id)
    inner = poll_result.get("result", {})

    if isinstance(inner, dict) and inner.get("eval_status") == "completed":
        eval_data = inner["result"]
        print("Evaluation complete")
        print(f"  Metric:  {eval_data['name']}")
        print(f"  Value:   {eval_data['value']}")
        print(f"  Runtime: {eval_data['runtime'] / 1000:.1f}s")
        print(f"  Reason:  {eval_data['reason'][:120]}...")
        break

    print(f"  Attempt {attempt + 1}/15: still processing...")
    time.sleep(5)
else:
    print("Timed out waiting for result")

You should see:

  Attempt 1/15: still processing...
  Attempt 2/15: still processing...

Evaluation complete
  Metric:  groundedness
  Value:   Passed
  Runtime: 24.2s
  Reason:  The output is fully supported by the provided context. The Eiffel Tower being in Paris, France is...

The runtime, reason text, and attempt count above are illustrative and will vary with your account tier and network conditions.

eval_status moves from pending to completed once the server finishes scoring; polling on a short interval is what turns the fire-and-forget submission from step 2 into a usable result.

Use concurrent.futures.ThreadPoolExecutor to submit many evaluations concurrently. Each thread calls Evaluator.evaluate() independently.

import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from fi.evals import Evaluator

evaluator = Evaluator(
    fi_api_key=os.environ["FI_API_KEY"],
    fi_secret_key=os.environ["FI_SECRET_KEY"],
)

# Sample dataset, 50 items
test_cases = [
    {
        "output": f"Response {i}: The capital of France is Paris.",
        "context": "Paris is the capital and most populous city of France.",
        "input": f"Question {i}: What is the capital of France?",
    }
    for i in range(50)
]

def evaluate_one(index, test_case):
    result = evaluator.evaluate(
        eval_templates="groundedness",
        inputs=test_case,
        model_name="turing_small",
    )
    return index, result

results = [None] * len(test_cases)
completed = 0
failed = 0
start = time.time()

with ThreadPoolExecutor(max_workers=8) as executor:
    futures = {
        executor.submit(evaluate_one, i, tc): i
        for i, tc in enumerate(test_cases)
    }

    for future in as_completed(futures):
        idx = futures[future]
        try:
            idx, result = future.result(timeout=60)
            results[idx] = result
            completed += 1
        except Exception as exc:
            # One slow or errored item shouldn't sink the whole batch.
            print(f"  Item {idx} failed: {exc}")
            failed += 1
        if (completed + failed) % 10 == 0:
            elapsed = time.time() - start
            print(f"Progress: {completed + failed}/{len(test_cases)} ({elapsed:.1f}s)")

elapsed = time.time() - start
print(f"Done in {elapsed:.1f}s. Succeeded: {completed}, Failed: {failed}")

scored = sum(
    1 for r in results
    if r and r.eval_results and r.eval_results[0].output is not None
)
print(f"Scored: {scored}/{len(test_cases)}")

You should see:

Progress: 10/50 (3.2s)
Progress: 20/50 (5.8s)
  Item 27 failed: TimeoutError
Progress: 30/50 (8.1s)
Progress: 40/50 (10.5s)
Progress: 50/50 (12.9s)

Done in 12.9s. Succeeded: 49, Failed: 1
Scored: 49/50

Timings, the failing item, and the succeeded/failed split are illustrative and vary with your account tier and network conditions. The batch completes even when individual items time out or error, because each future’s exception is caught and counted rather than left to crash the loop.

For maximum throughput, submit every item with is_async=True first, then poll each returned eval_id in a loop until all complete, instead of waiting on each one in turn.

import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from fi.evals import Evaluator

evaluator = Evaluator(
    fi_api_key=os.environ["FI_API_KEY"],
    fi_secret_key=os.environ["FI_SECRET_KEY"],
)

test_cases = [
    {
        "output": f"Response {i}: Python is a programming language.",
        "context": "Python is a high-level, general-purpose programming language.",
    }
    for i in range(50)
]

def submit_async(index, test_case):
    result = evaluator.evaluate(
        eval_templates="groundedness",
        inputs=test_case,
        model_name="turing_small",
        is_async=True,
    )
    eval_id = result.eval_results[0].eval_id
    return index, eval_id

eval_ids = {}
with ThreadPoolExecutor(max_workers=8) as executor:
    futures = {
        executor.submit(submit_async, i, tc): i
        for i, tc in enumerate(test_cases)
    }
    for future in as_completed(futures):
        idx, eval_id = future.result()
        eval_ids[idx] = eval_id

print(f"Submitted {len(eval_ids)} async evaluations")

results = {}
max_polls = 15

for poll_round in range(max_polls):
    still_pending = {
        idx: eid for idx, eid in eval_ids.items() if idx not in results
    }
    if not still_pending:
        break

    for idx, eid in still_pending.items():
        poll_result = evaluator.get_eval_result(eid)
        inner = poll_result.get("result", {})
        if isinstance(inner, dict) and inner.get("eval_status") == "completed":
            results[idx] = poll_result

    print(f"  Poll {poll_round + 1}: {len(results)}/{len(eval_ids)} completed")
    if len(results) < len(eval_ids):
        time.sleep(3)

if len(results) < len(eval_ids):
    print(f"Poll budget exhausted: {len(results)}/{len(eval_ids)} evaluations completed, {len(eval_ids) - len(results)} still pending")
else:
    print(f"Completed {len(results)}/{len(eval_ids)} evaluations")

You should see:

Submitted 50 async evaluations
  Poll 1: 12/50 completed
  Poll 2: 34/50 completed
  Poll 3: 50/50 completed

Completed 50/50 evaluations

The poll counts and round timing above are illustrative and vary with your account tier and network conditions.

Unlike step 4, submission here never blocks on a response: every item is submitted before any result is awaited, so the polling loop is the only place this step waits.