harness-evals

Open-source AI evaluation framework for LLM agents, prompts, and structured outputs.

Every metric produces a normalized Score (0.0–1.0). Pass/fail is determined by a configurable threshold. No magic, no hidden state.

Install

pip install harness-evals            # core only
pip install harness-evals[llm]       # + LLM-judged metrics (OpenAI, Anthropic)
pip install harness-evals[otlp]      # + OTLP metrics & traces export
pip install harness-evals[langfuse]  # + Langfuse source/sink
pip install harness-evals[harness]   # + Harness AI Service LLM provider
pip install harness-evals[decision]  # + TypeSafe decision-primitive metrics (Choice/Score/Noul)
pip install harness-evals[benchmarks]# + academic benchmarks (MMLU, GSM8K, etc.)
pip install harness-evals[all]       # everything

Five Dimensions

Every metric belongs to one of five evaluation dimensions. Together they answer: where is my agent strong, and where is it weak?

Dimensions are set by the metric author — not user-configured. Any combination of metrics in a suite automatically produces a radar chart grouped by dimension.

Core Concepts

Golden — what you author: an input, the expected output, and optional context. Lives in your dataset files.

EvalCase — what metrics receive: a Golden enriched with the agent's actual output and runtime metadata (latency, tokens, cost).

BaseMetric — a scoring function. Takes an EvalCase, returns a Score. Each metric is a single class with a measure() method. Specialized base classes: ReliabilityMetric for multi-run metrics, SafetyMetric for safety metrics (reported separately, never averaged).

Message — a conversation turn: role, content, and optional tool calls. Maps to OpenAI chat messages, Langfuse generations, OTEL LLM spans.

ToolCall — a tool/function invocation: name, input, output. Maps to OpenAI function calls, Anthropic tool_use blocks, MCP invocations.

Score — the result: a value between 0.0 and 1.0, a threshold, and an auto-computed passed boolean.

evaluate() — runs multiple metrics on an eval case. Never raises — returns all scores including failures.

assert_test() — same as evaluate(), but raises AssertionError if any metric fails. Drop it into pytest.

Datasets — load_dataset() / save_dataset() read and write JSONL or JSON golden files. Source adapters (LocalDatasetSource, HttpDatasetSource) fetch goldens from local files or HTTP endpoints using the ResourceRef URI system.

Prompt templates — PromptTemplate uses {{var}} placeholders (not str.format). Prompt source adapters fetch templates from local files or HTTP endpoints.

Targets — the system under test. PromptTarget renders a prompt template through an LLM. HttpTarget POSTs to a deployed agent endpoint. Both implement BaseTarget.ainvoke(golden) -> EvalCase.

Importers — adapters that hydrate EvalCase from production trace data. LangfuseEvalCaseSource and OTELEvalCaseSource map traces to typed fields automatically.

Baselines — JsonBaselineStore saves score snapshots per run. compare_to_baseline() detects regressions, improvements, and unchanged metrics with a configurable tolerance.

Data Flow

Golden (authored) + agent output → EvalCase → Score (result)
Production traces (Langfuse/OTEL) → Importer → EvalCase → Score (result)
YAML config → run_config() → evaluate_dataset() → Score (result)
run_eval() one-liner → evaluate_dataset() → Score (result)

Quick Start — Local Eval Runner

The fastest way to run an eval is the CLI with a YAML config:

# my-eval.eval.yaml
name: support-bot
dataset: ./goldens.jsonl
target:
  type: prompt
  prompt: ./prompts/support-bot.txt
  model: {provider: openai, name: gpt-4o}
metrics:
  - exact_match
  - {kind: geval, threshold: 0.7, params: {criteria: "Correct and helpful?"}}
judge_llm: {provider: openai, name: gpt-4o}
sinks: [stdout]
baseline: {store: json, path: .evals/baseline.json}

harness-evals run my-eval.eval.yaml                # run the eval
harness-evals run my-eval.eval.yaml --golden-ids case-a,case-b  # run subset by golden id
harness-evals run my-eval.eval.yaml --modules ci,ce             # run goldens tagged module=ci or ce
harness-evals run my-eval.eval.yaml --golden-tags scenario_type=write,environment=qa
harness-evals run my-eval.eval.yaml --baseline     # compare against stored baseline
harness-evals run my-eval.eval.yaml --fail-under 0.8  # CI gate on absolute score
harness-evals list-metrics                         # see all available metrics
harness-evals discover                             # find all *.eval.yaml in cwd

Or use the code-first run_eval() one-liner:

from harness_evals import Golden, run_eval
from harness_evals.metrics import ExactMatchMetric

run_eval(
    "my-eval",
    data="./goldens.jsonl",             # ref string, ResourceRef, or list[Golden]
    target=my_agent_function,           # BaseTarget, async callable, or sync callable
    metrics=[ExactMatchMetric()],
)

Both the YAML runner and run_eval() funnel into evaluate_dataset() — the same engine used by the programmatic API below.

Model params support ${VAR} env-var interpolation, so the target and judge can use separate API keys:

target:
  type: prompt
  prompt: ./prompt.txt
  model: {provider: openai, name: gpt-4o, api_key: "${TARGET_KEY}"}
judge_llm: {provider: openai, name: gpt-4o, api_key: "${JUDGE_KEY}"}

If no api_key is specified, the provider falls back to its default env var (e.g. OPENAI_API_KEY).

Usage

Evaluate a response

from harness_evals import EvalCase, evaluate
from harness_evals.metrics import ExactMatchMetric, LatencyMetric

ec = EvalCase(
    input="What is the capital of France?",
    output="Paris",
    expected="Paris",
    latency_ms=320,
)

scores = evaluate(ec, metrics=[
    ExactMatchMetric(),
    LatencyMetric(max_ms=2000),  # binary pass/fail: value is 1.0 if latency_ms <= max_ms, else 0.0
])

for s in scores:
    print(f"{'PASS' if s.passed else 'FAIL'} {s.name}: {s.value:.2f}")
# PASS exact_match: 1.00
# PASS latency: 1.00

Evaluate structured output (JSON/YAML)

from harness_evals import EvalCase, assert_test
from harness_evals.metrics import JsonDiffMetric, SchemaValidationMetric

expected = {"apiVersion": "apps/v1", "kind": "Deployment", "metadata": {"name": "nginx"}}
schema = {"type": "object", "required": ["apiVersion", "kind"], "properties": {
    "apiVersion": {"type": "string"}, "kind": {"type": "string"}
}}

ec = EvalCase(
    input="Create a K8s deployment for nginx",
    output={"apiVersion": "apps/v1", "kind": "Deployment", "metadata": {"name": "nginx"}},
    expected=expected,
)

assert_test(ec, metrics=[
    JsonDiffMetric(threshold=0.9),
    SchemaValidationMetric(schema=schema),
])

Use with pytest

def test_agent_accuracy():
    ec = EvalCase(
        input="What is 2+2?",
        output=agent.run("What is 2+2?"),
        expected="4",
    )
    assert_test(ec, metrics=[ExactMatchMetric()])

assert_test() raises AssertionError on failure — works natively with pytest, unittest, and any CI system.

Evaluate a dataset with an agent

import asyncio
from harness_evals import Golden, EvalCase, evaluate_dataset
from harness_evals.metrics import ExactMatchMetric

goldens = [
    Golden(input="What is 2+2?", expected="4"),
    Golden(input="Capital of France?", expected="Paris"),
]

async def run_agent(golden: Golden) -> EvalCase:
    result = await agent.arun(golden.input)
    return EvalCase.from_golden(golden, output=result)

results = asyncio.run(evaluate_dataset(goldens, run_agent, metrics=[ExactMatchMetric()]))

Load and save datasets

from harness_evals import load_dataset, save_dataset, Golden

# Load goldens from a JSONL file
goldens = load_dataset("goldens.jsonl")           # JSONL (one JSON object per line)
goldens = load_dataset("goldens.json", format="json")  # JSON array

# Create and save a dataset
goldens = [
    Golden(input="What is 2+2?", expected="4"),
    Golden(input="Capital of France?", expected="Paris"),
]
save_dataset(goldens, "goldens.jsonl")

Evaluate with an HttpTarget

import asyncio
from harness_evals import Golden, evaluate_dataset
from harness_evals.targets import HttpTarget, BearerAuth
from harness_evals.metrics import ExactMatchMetric

target = HttpTarget(
    url="http://localhost:8080/run",
    auth=BearerAuth(token="${API_TOKEN}"),  # ${VAR} resolved from env
    output_path="$.answer",
    timeout_s=30,
)

goldens = [Golden(input="What is 2+2?", expected="4")]
scores = asyncio.run(evaluate_dataset(goldens, target.ainvoke, metrics=[ExactMatchMetric()]))

Compare scores against a baseline

from harness_evals.baseline import JsonBaselineStore, compare_to_baseline
from harness_evals.core.score import Score

store = JsonBaselineStore(baseline_dir=".evals/baselines")

# After an eval run, save scores as a baseline
store.save("run-001", {"exact_match": scores_list})

# Later, compare a new run against the saved baseline
baseline = store.load()  # loads latest
result = compare_to_baseline(current_scores, baseline, tolerance=0.05)
if result.has_regressions:
    print(result.summary())

Measure reliability across multiple runs

from harness_evals import EvalCase, evaluate
from harness_evals.metrics import OutcomeConsistencyMetric

runs = [
    EvalCase(input="task", output=agent.run("task"))
    for _ in range(5)
]

ec = EvalCase(input="task", output=runs[0].output, runs=runs)
scores = evaluate(ec, metrics=[OutcomeConsistencyMetric(threshold=0.8)])

Evaluate with typed tool calls

from harness_evals import EvalCase, ToolCall, evaluate
from harness_evals.metrics import ToolCorrectnessMetric

ec = EvalCase(
    input="Check weather in Paris",
    output="It's 18C and sunny",
    tool_calls=[ToolCall(name="get_weather", input={"city": "Paris"})],
    expected_tools=["get_weather"],
)
scores = evaluate(ec, metrics=[ToolCorrectnessMetric()])

For deterministic argument checks, pair ToolCorrectnessMetric with ToolArgumentMatchMetric:

from harness_evals import EvalCase, ToolCall, evaluate
from harness_evals.metrics import ToolArgumentMatchMetric, ToolCorrectnessMetric

ec = EvalCase(
    input="Check weather in Paris",
    output="It's 18C and sunny",
    tool_calls=[ToolCall(name="get_weather", input={"city": "Paris", "units": "C"})],
    expected_tools=["get_weather"],
    expected_tool_calls=[ToolCall(name="get_weather", input={"city": "Paris"})],
)
scores = evaluate(
    ec,
    metrics=[
        ToolCorrectnessMetric(mode="exact"),
        ToolArgumentMatchMetric(arg_match="subset", ignore_keys={"trace_id"}),
    ],
)

Evaluate conversation messages

from harness_evals import EvalCase, Message

ec = EvalCase(
    input="Help me debug this error",
    output="The issue is a null pointer...",
    messages=[
        Message(role="user", content="Help me debug this error"),
        Message(role="assistant", content="Can you share the stack trace?"),
        Message(role="user", content="Here it is: NullPointerException at..."),
        Message(role="assistant", content="The issue is a null pointer..."),
    ],
)

Multi-turn evaluation with ConversationGolden

ConversationGolden supports three modes for multi-turn evaluation:

Simulate — LLM-driven user turns

import asyncio
from harness_evals import evaluate_dataset, Message
from harness_evals.conversation import ConversationGolden, ConversationMode
from harness_evals.metrics.conversation import GoalAccuracyMetric, TurnRelevancyMetric
from harness_evals.llm import OpenAILLM

llm = OpenAILLM(model="gpt-4o-mini")

goldens = [
    ConversationGolden(
        scenario="User requests a refund for a damaged item",
        expected_outcome="Agent processes refund and confirms timeline",
        max_turns=8,
        user_persona="Frustrated customer, short replies",
        mode=ConversationMode.SIMULATE,
    )
]

async def my_agent(messages: list[Message]) -> Message:
    # Your agent logic here — receives full history, returns next response
    ...

results = asyncio.run(evaluate_dataset(
    goldens, my_agent,
    metrics=[GoalAccuracyMetric(llm=llm), TurnRelevancyMetric(llm=llm)],
    simulator_llm=llm,
))

CLI conversation evals with human-in-the-loop

The CLI can run ConversationGolden datasets by adding a top-level conversation: block and using a conversational target. The core loop is protocol-agnostic: when an agent turn ends with pending_human_input in message metadata, the simulator generates a continuation payload and calls the agent again until the turn completes or max_elicitation_rounds is reached.

Configure the target explicitly for your agent's event names and request body shape. Load the adapter via plugins when using Harness SSE elicitation:

plugins:
  - examples.harness_sse_elicitation_adapter

conversation:
  elicitation_adapter: harness_sse

Harness LLM gateway (OpenAI-compatible /llm-gw/v1, PAT via x-api-key) uses provider: harness_gateway — not provider: openai with a gateway URL. Model names such as online/anthropic/claude-sonnet-4-5 are gateway routing aliases; the provider selects the client, not the model vendor. Domain-specific plain-text elicitation matchers load via plugins: in eval YAML (register_plain_text_followup_resolver); harness-evals run scopes those resolvers to the configured plugin modules. SDK callers pass resolvers explicitly via evaluate_conversation(..., plain_text_followup_resolvers=[...]). provider: harness is the separate Harness AI Service /chat client (JWT auth).

name: my-agent-conversation

conversation:
  mode: simulate
  max_turns: 1
  max_elicitation_rounds: 6
  simulator_llm: {provider: openai, name: gpt-4o-mini}
  # elicitation_adapter: harness_sse   # requires examples.harness_sse_elicitation_adapter plugin

dataset: ./examples/harness-agent.goldens.jsonl

target:
  type: conversational_streaming_http
  url: "${SSE_ENDPOINT_URL}"
  user_body_template:
    prompt: "{{last_user_content}}"
    stream: true
  human_input_body_template:
    human_input: "{{human_input}}"
    stream: true
  human_input_events:
    - needs_input
  correlation_id_field: request_id
  output_event: assistant_message
  output_path: $.v

metrics:
  - {kind: goal_accuracy, threshold: 0.7}

Conversation goldens may include initial_prompt, elicitation_hints, and max_elicitation_rounds. Use elicitation_hints.intents for semantic values and elicitation_hints.matchers to map live questions to those intents.

See examples/conversation-goldens.README.md for the protocol-neutral schema. For a full Harness SSE integration, see examples/harness-agent-conversation.eval.yaml.

Scripted — bring your own user turns, call agent live

goldens = [
    ConversationGolden(
        scenario="Customer asks about order status",
        expected_outcome="Agent provides tracking info",
        turns=[
            Message(role="user", content="Where is my order #12345?"),
            Message(role="user", content="When will it arrive?"),
            Message(role="user", content="Can I change the delivery address?"),
        ],
        mode=ConversationMode.SCRIPTED,
    )
]

results = asyncio.run(evaluate_dataset(
    goldens, my_agent,
    metrics=[GoalAccuracyMetric(llm=llm), TurnRelevancyMetric(llm=llm)],
    simulator_llm=llm,
))

Replay — score an existing transcript

goldens = [
    ConversationGolden(
        scenario="Support interaction",
        expected_outcome="Issue resolved",
        turns=[
            Message(role="user", content="My account is locked"),
            Message(role="assistant", content="I can help. What's your email?"),
            Message(role="user", content="alice@example.com"),
            Message(role="assistant", content="Done — your account is unlocked."),
        ],
        mode=ConversationMode.REPLAY,
    )
]

results = asyncio.run(evaluate_dataset(
    goldens, my_agent,  # agent_fn not called in REPLAY mode; simulator_llm still required by API
    metrics=[GoalAccuracyMetric(llm=llm), TurnRelevancyMetric(llm=llm)],
    simulator_llm=llm,
))

Branching scenarios with SimulationGraph

SimulationGraph drives simulation through a declarative DAG (ConversationMode.GRAPH). Nodes produce user turns — ScriptedNode (fixed message), LLMNode (LLM generates a turn toward a goal), BranchNode (routes on the agent's last response), StopNode (ends the conversation) — and edges connect them, optionally guarded by named predicates. Cycles are structurally rejected; to loop, use a node with no matching outgoing edge (it re-executes until a predicate matches or max_turns is reached).

from harness_evals.conversation import ConversationGolden, ConversationMode, ConversationSimulator
from harness_evals.conversation.graph import (
    SimulationGraph, ScriptedNode, LLMNode, BranchNode, StopNode, Edge,
)

graph = SimulationGraph(
    start="greeting",
    nodes={
        "greeting": ScriptedNode(message="I need help with my order"),
        "probe": LLMNode(goal="Ask increasingly specific questions about the delay"),
        "route": BranchNode(),
        "escalate": ScriptedNode(message="I want to speak to a manager"),
        "done": StopNode(),
    },
    edges=[
        Edge(source="greeting", target="probe"),
        Edge(source="probe", target="route"),
        Edge(source="route", target="escalate", predicate="unresolved"),
        Edge(source="route", target="done", predicate="resolved"),
    ],
    predicates={
        "unresolved": lambda msg: "sorry" in msg.content.lower(),
        "resolved": lambda msg: "refund" in msg.content.lower(),
    },
)

simulator = ConversationSimulator(simulator_llm=llm, graph=graph)

A graph can also be serialized to/from a plain dict (SimulationGraph.from_dict(...)). When loaded from ConversationGolden.graph_config, only unconditional edges are supported — predicate functions aren't serializable, so pass them explicitly to from_dict() when needed.

Evaluate production traces from Langfuse

from harness_evals.importers.langfuse import LangfuseEvalCaseSource
from harness_evals import evaluate
from harness_evals.metrics import FaithfulnessMetric, LatencyMetric, PIIMetric
from harness_evals.sinks.langfuse_sink import LangfuseSink

source = LangfuseEvalCaseSource(langfuse_client)
ec = source.from_trace("trace-id-123")

scores = evaluate(ec, metrics=[
    FaithfulnessMetric(llm=llm),
    LatencyMetric(max_ms=3000),
    PIIMetric(),
], sinks=[LangfuseSink()])  # scores written back to the same trace

Batch-evaluate Langfuse traces by filter

from harness_evals.importers.langfuse import LangfuseEvalCaseSource
from harness_evals import evaluate_cases
from harness_evals.metrics import LatencyMetric, PIIMetric
from harness_evals.sinks.langfuse_sink import LangfuseSink

source = LangfuseEvalCaseSource(langfuse_client)
cases = source.from_traces(tags=["production"], user_id="user_123", limit=50)

all_scores = evaluate_cases(cases, metrics=[
    LatencyMetric(max_ms=3000),
    PIIMetric(),
], sinks=[LangfuseSink()])

Evaluate production traces from OpenTelemetry

from harness_evals.importers.otel import OTELEvalCaseSource

ec = OTELEvalCaseSource.from_spans(collected_spans)
scores = evaluate(ec, metrics=[...])

Adapt assembled trace or span rows

For flat span rows with attributes dictionaries or JSON-encoded objects, use the trace adapter directly. Pass configured metrics to materialize their declared inputs:

from harness_evals.adapters.trace import spans_to_eval_case
from harness_evals.metrics import LatencyMetric, TokenCostMetric

metrics = [LatencyMetric(), TokenCostMetric()]
ec, warnings = spans_to_eval_case(collected_spans, metrics=metrics)
scores = evaluate(ec, metrics=metrics)

Use spans_to_eval_case_for_span(spans, span_id, metrics=metrics) for a subtree. Omitting metrics materializes every supported field; metrics=[] requests no fields. Span-scoped cases retain their span association even when metadata is not requested. This adapter is separate from OTELEvalCaseSource hydration.

• Non-object or malformed attributes are treated as absent.
• Token counts must be non-negative integers (including integral numbers and decimal integer strings). Booleans, fractional counts, and non-finite values are ignored; valid input/output dimensions remain usable independently.
• Cost and duration must be finite, non-negative numbers. Missing or invalid telemetry stays None; an observed zero stays zero. Child-span costs take precedence over rolled-up root costs to avoid counting both.
• Trace latency uses the earliest valid start and latest observed span end, not the end of the last-starting span. Missing durations are not invented; a reported root duration also bounds latency from below when its timestamp is missing or invalid.

A metric's eval_inputs describes field dependencies, and applicability reports source capability, not whether a particular trace has usable values. Inspect adapter warnings and each metric's missing-data behavior. Empty output is not a blanket skip signal: telemetry-only metrics need no answer text. Trace and Span metric scopes are supported. The bounded Group assembly API below is separate; it does not enable Group metric eligibility or scheduling.

Assemble a bounded Group from captured traces

traces_to_group_eval_case() accepts 2–64 already-selected member traces. It is a pure local adapter: no discovery, membership selection, network calls, target execution, or model-context fitting. The member bound is not a bound on span count, bytes, or model tokens; oversized input is never silently truncated.

from harness_evals.adapters import GroupTrace, traces_to_group_eval_case

members = [
    GroupTrace(trace_id="first", spans=[{
        "span_id": "generation", "start_timestamp": 1, "duration_ms": 10,
        "attributes": {"gen_ai.output.messages": [
            {"id": "answer", "role": "assistant", "content": "First answer"},
        ]},
    }]),
    GroupTrace(trace_id="second", spans=[{
        "span_id": "generation", "start_timestamp": 2, "duration_ms": 10,
        "attributes": {"gen_ai.output.messages": [
            {"id": "answer", "role": "assistant", "content": "Second answer"},
        ]},
    }]),
]
assembly = traces_to_group_eval_case(members)
case = assembly.case
assert case.input == case.output == ""
assert case.runs is None
assert [message.content for message in case.messages] == ["First answer", "Second answer"]

• The conversation is in typed case.messages, with actual execution evidence in case.tool_calls. Empty input/output are schema placeholders, not a synthesized prompt or final answer. No Group latency, expected answer, retry count, confidence, or repeated runs are invented.
• Roles, empty/tool-only messages, legitimate repeated text, and supported per-message telemetry/retrieval data are retained. Duplicate capture is removed only with a trace-scoped span/message identity. Missing identity retains uncertain occurrences and produces an ambiguity diagnostic; conflicting captures with the same identity raise a payload-safe GroupInputError.
• Tool links use explicit call IDs or an unambiguous structural association, never just tool names, arguments, or the most recent assistant message. Ambiguous results remain visible without being assigned to an arbitrary call. Cycles and partial trees are reported rather than represented as complete.
• Token and cost values use the same validation as Trace adaptation. Totals are observed nonoverlapping subtotals, not evidence of complete usage. A valid descendant report excludes an ancestor rollup independently for each usage dimension. Unresolvable overlap is excluded and reported. Missing remains None; an observed zero remains zero. No prices or costs are estimated.
• capture_complete=None (the default) means unknown capture. An explicit source assertion of True can establish known absence ([]), but observed structural contradictions downgrade it. Connected spans alone never establish complete capture. Typed assembly.diagnostics contains only counts and fixed issue codes; source payloads and identifiers stay out of diagnostics and exceptions.
• Pass configured metrics= to materialize required and optional fields. Omit it for all supported fields; metrics=[] returns placeholders only. Canonical evidence and diagnostics are computed before projection, so adding unrelated metrics cannot change another metric's evidence. Unknown custom declarations fall back to full materialization. This saves materialization, not parsing or canonical-assembly work.
• GroupAuthoredData optionally supplies explicit expected_tools, expected_tool_calls, metadata, and tags; these are copied, never inferred from actual calls. Do not place raw grouping values or sensitive data in labels.
• Assembly is not metric admission: do not run single-response metrics against the empty placeholders. Use the opt-in configured admission gate below for explicitly reviewed Group configurations and actual-input sufficiency.

Resolve configured contracts before admitting metrics

Catalog discovery describes possibilities, not permission to run a selected configuration. Use entry.to_dict() for JSON catalog metadata and entry.configured_contract(options) (or resolve_metric_contract(kind, options)) for a versioned, JSON-serializable configured contract:

from harness_evals import EvalCase
from harness_evals.validation import resolve_metric_contract, validate_metric

contract = resolve_metric_contract("latency", {"max_ms": 2000})
case = EvalCase(input="", output="", latency_ms=0)
decision = validate_metric(contract, case)
assert decision.ready

Resolution accepts constructor-style metric options, not factory envelopes or provider metadata. Merge per-entry overrides and apply SDK option normalization first. To describe an existing metric, use describe_configured_metric(metric). Neither API constructs network providers, loads custom code, or runs scorers. Unknown/custom or unreviewed configurations fail closed in this new API.

• Contracts distinguish required/optional fields and alternative input groups from nested, configuration-specific evidence requirements. Use to_dict() / ConfiguredMetricContract.from_dict() for JSON snapshots (schema_version=1).
• ValidationContext selects offline/online mode, online scope, and case, dataset, or repeated-run execution. Dataset metrics require labeled cases; related Group members are not repeated runs. Resource availability can be asserted explicitly; omitted resource checks produce a warning, not a claim that credentials or network access were verified.
• validate_metric(contract, context=...) checks eligibility only and never reports ready=True. Supply actual evidence for readiness. Missing values, malformed values, and known-empty values are different: tool_calls=[] can be valid evidence; tool_calls=None cannot establish that no calls occurred.
• Alternatives, structural fallback paths, enabled reference sources, decision labels, composite operators, and assistant-turn evidence are checked without invoking a provider or computing a quality score.
• Decisions expose stable reason codes, severity, static messages, and safe requirement coordinates through decision.to_dict(). Payloads, source IDs, configured paths, and exception strings are not echoed in reasons.
• Initial Group admission is explicitly reviewed for tool correctness, tool selection, tool argument matching without missing-data skips, MCP trace completeness, turn latency, and turn token cost. Tool comparisons require complete, unambiguous capture; order-sensitive comparisons require reliable ordering, not identifier-based tie-breaking. Tool argument matching requires explicit argument objects on both observed and authored calls: use {} for known-empty arguments; None means unknown. Greedy subset argument pairing also requires established ordering. Turn telemetry may admit missing measurements with warnings, but malformed consumed values are rejected.
• Weighted composites require finite positive configured and remaining active weight. A legitimate zero quality score remains usable evidence; zero total weight does not.

For Group, pass the diagnostics associated with the actual assembly:

from harness_evals.validation import ValidationContext

contract = resolve_metric_contract("tool_correctness", {"mode": "exact"})
context = ValidationContext(mode="online", scope="group", group_diagnostics=assembly.diagnostics)
decision = validate_metric(contract, assembly.case, context=context)

The tool comparison also needs explicit authored expectations, for example via GroupAuthoredData(expected_tools=[]); the actual observed empty list remains valid only with sufficient capture assertions. Other Group metrics stay excluded; empty Group input/output do not unlock single-answer metrics.

Integration ownership: backend services should resolve contracts server-side, validate again immediately before scoring, and return the SDK decisions/reasons. Frontends should display those results rather than maintain another metric matrix. Existing runners do not automatically enforce this opt-in API. Do not trust a client-supplied contract as authorization, and re-resolve stored contracts when the configuration or SDK version changes. The legacy applicability property is still a source-capability hint, not the actual-evidence gate.

Admission checks the selected contract, mode/scope/execution and explicit resource assertions before reading evidence. Group diagnostics are checked before nested rules, and projected messages/calls must match their diagnostic counts. The result flags describe the checks actually performed:

Use ready immediately before scoring. A context-only success is not permission to score an unchecked case; warnings do not turn usable evidence into a rejection.

Time-window listing and session merge are generic (lookback_days, group_by=session_id). Pass any TraceCatalog to list traces; Langfuse is one implementation:

from harness_evals.importers.langfuse import LangfuseTraceCatalog
from harness_evals.importers.otel import OTELEvalCaseSource
from harness_evals.refs import ResourceRef

source = OTELEvalCaseSource(catalog=LangfuseTraceCatalog(langfuse_client))
cases = await source.fetch(ResourceRef(
    source="otel",
    id="",
    extra={"lookback_days": 7, "group_by": "session_id", "limit": 200},
))

Note: The previous import paths (from harness_evals.sources.langfuse import LangfuseSource and from harness_evals.sources.otel import OTELSource) still work but emit a DeprecationWarning. Migrate to the importers paths shown above.

Write results to a file

from harness_evals.sinks import StdoutSink, JsonSink
from harness_evals.sinks.langfuse_sink import LangfuseSink

scores = evaluate(ec, metrics=[...], sinks=[
    StdoutSink(),
    JsonSink("results/scores.jsonl"),
    LangfuseSink(),  # requires pip install harness-evals[langfuse]
])

Export to an OTLP-compatible backend

For protocol="http" (the default in some integrations is grpc), set endpoint to the OTLP HTTP base URL; OtlpSink appends /v1/traces and /v1/metrics for the exporters. Use protocol="grpc" with a host:port endpoint for gRPC OTLP.

from harness_evals.sinks.otlp_sink import OtlpSink  # requires pip install harness-evals[otlp]

sink = OtlpSink(
    endpoint="http://collector:4317",
    run_id="my-eval-run-001",
    resource_attributes={"deployment.environment": "ci"},
    extra_attributes={"eval.suite_id": "nightly-regression"},
)

scores = evaluate(ec, metrics=[...], sinks=[sink])

Attach eval spans to an existing trace

If your eval engine already creates OTel spans, pass a parent_context so the eval-run span becomes a child (same trace ID, unified view in Jaeger/Tempo):

from opentelemetry import trace
from harness_evals.sinks.otlp_sink import OtlpSink

tracer = trace.get_tracer("my-engine")
with tracer.start_as_current_span("orchestration") as parent:
    ctx = trace.set_span_in_context(parent)
    sink = OtlpSink(endpoint="http://collector:4317", parent_context=ctx)
    evaluate_cases(cases, metrics=[...], sinks=[sink])

Share a TracerProvider (single export pipeline)

For full control, pass your own TracerProvider and/or MeterProvider. The sink won't flush or shutdown providers it doesn't own — you retain lifecycle control:

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from harness_evals.sinks.otlp_sink import OtlpSink

provider = TracerProvider(resource=my_resource)
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(endpoint="http://collector:4317")))
tracer = provider.get_tracer("my-engine")

with tracer.start_as_current_span("orchestration") as parent:
    ctx = trace.set_span_in_context(parent)
    sink = OtlpSink(tracer_provider=provider, parent_context=ctx, run_id="run-123")
    evaluate_cases(cases, metrics=[...], sinks=[sink])

provider.shutdown()  # caller owns lifecycle

Evaluate security remediations

from harness_evals import EvalCase, evaluate
from harness_evals.llm.openai import OpenAILLM  # or AnthropicLLM, HarnessAILLM
from harness_evals.metrics.security import (
    VulnerabilityCorrectnessMetric,
    SecurityCompletenessMetric,
    CodeSafetyMetric,
    CodeQualityMetric,
    ExplanationQualityMetric,
    RootCauseAnalysisMetric,
    ActionabilityMetric,
    remediation_quality_index,
)

llm = OpenAILLM()  # uses OPENAI_API_KEY env var

ec = EvalCase(
    input="CWE-79: Reflected XSS in user_profile.py line 42. User input rendered without escaping.",
    output="## Fix\n```python\nfrom markupsafe import escape\nname = escape(request.args.get('name', ''))\n```",
)

scores = evaluate(ec, metrics=[
    VulnerabilityCorrectnessMetric(llm=llm, threshold=0.5),
    SecurityCompletenessMetric(llm=llm, threshold=0.5),
    CodeSafetyMetric(llm=llm, threshold=0.5),
    CodeQualityMetric(llm=llm, threshold=0.5),
    ExplanationQualityMetric(llm=llm, threshold=0.5),
    RootCauseAnalysisMetric(llm=llm, threshold=0.5),
    ActionabilityMetric(llm=llm, threshold=0.5),
])

rqi = remediation_quality_index(scores)
print(f"RQI: {rqi.value:.3f} ({'PASS' if rqi.passed else 'FAIL'})")

Summarize results across a dataset

from harness_evals import evaluate_cases, summarize

all_scores = evaluate_cases(eval_cases, metrics=[...])
summary = summarize(all_scores)

for name, m in summary.by_metric.items():
    print(f"{name}: mean={m.mean:.2f} pass_rate={m.pass_rate:.0%} ({m.count} cases)")

Synthesize a dataset from documents

Synthesizer (and task-specific variants like ConversationSynthesizer, ScriptedConversationSynthesizer) generate Golden datasets from source documents using an LLM. InputGenerator produces input variations (rephrasings, adversarial rewrites, complexity ladders) for robustness testing.

import asyncio
from harness_evals import Synthesizer, save_dataset
from harness_evals.llm import OpenAILLM

synth = Synthesizer(llm=OpenAILLM(model="gpt-4o-mini"))
goldens = asyncio.run(synth.generate(documents=[doc_text], n=20, difficulty="mixed"))
save_dataset(goldens, "goldens.jsonl")

See examples/synthesize_dataset.py for a runnable end-to-end example.

Optimize a prompt automatically

PromptOptimizer runs an iterative diagnose → rewrite → re-evaluate loop against a golden set until it hits a target score or runs out of patience. The judge LLM must be a different instance than the model being optimized (self-evaluation is rejected).

import asyncio
from harness_evals import PromptOptimizer
from harness_evals.llm import OpenAILLM
from harness_evals.prompts import PromptTemplate
from harness_evals.metrics import GEvalMetric

optimizer = PromptOptimizer(
    model=OpenAILLM(model="gpt-4o-mini"),
    judge=OpenAILLM(model="gpt-4o"),       # must differ from model
    metrics=[GEvalMetric(criteria="Correct and helpful?", llm=OpenAILLM(model="gpt-4o"))],
    target_score=0.85,
    max_iterations=10,
)

result = asyncio.run(optimizer.optimize(
    PromptTemplate("Answer the question: {{input}}"),  # placeholders must be in input_variables (defaults to ["input"])
    goldens,
))
print(f"{result.initial_score:.2f} → {result.best_score:.2f} in {result.iterations} iters")
print(result.best_prompt.template)
result.save("optimized-prompt.json")

Academic Benchmarks

Run standardized LLM benchmarks with a single call. Datasets are fetched from HuggingFace Hub and cached locally.

pip install harness-evals[benchmarks]  # requires httpx

import asyncio
from harness_evals.benchmarks import MMLU, GSM8K, HumanEval
from harness_evals.llm import OpenAILLM

model = OpenAILLM(model="gpt-4o")

# MMLU — 5-shot by default (per original paper)
result = asyncio.run(MMLU(subjects=["abstract_algebra"]).run(model))
print(f"MMLU accuracy: {result.accuracy:.2%}")

# GSM8K — 8-shot chain-of-thought by default
result = asyncio.run(GSM8K().run(model, limit=100))
print(f"GSM8K accuracy: {result.accuracy:.2%}")

# HumanEval — process-isolated code execution (not a full sandbox)
result = asyncio.run(HumanEval().run(model))
print(f"HumanEval pass@1: {result.pass_at_1:.2%}")

All benchmarks support: shots=N override, limit=N for quick runs, offline=True for cached-only, sinks=[...] for output integration, and concurrency=N for throughput control.

Available Metrics

EvalCase Fields

EvalCase(
    input="the prompt or task",                    # required
    output="what the agent produced",              # required
    expected="ground truth",                       # optional (not needed for LLM-judged metrics)
    context=["retrieved doc 1", "retrieved doc 2"],# optional (for RAG metrics)
    messages=[Message(role="user", content="...")], # optional (for conversation metrics)
    tool_calls=[ToolCall(name="fn", input={...})], # optional (for agent/tool metrics)
    expected_tools=["fn1", "fn2"],                 # optional (expected tool names)
    latency_ms=320,                                # optional (typed, for LatencyMetric)
    token_count=150,                               # optional (typed, for TokenCostMetric)
    cost_usd=0.003,                                # optional (typed, for CostEfficiencyMetric)
    retry_count=0,                                 # optional (typed, for RetryCountMetric)
    confidence=0.95,                               # optional (typed)
    tags={"env": "ci", "model": "gpt-4o"},         # optional (for filtering)
    metadata={"custom_key": "value"},              # optional (extensible)
    runs=[...],                                    # optional (for reliability metrics)
)

Golden also supports expected_tools for defining expected tool names in datasets.

CLI Reference

harness-evals run <config.yaml> [--baseline] [--update-baseline] [--fail-under <float>]
harness-evals import <ref> [-o out.eval.yaml]
harness-evals list-metrics
harness-evals discover [path] [--glob pattern]

--fail-under and --baseline are independent checks. --fail-under is an absolute quality floor; baseline gating is a relative regression check. Both can fire in the same run.

Extending

Custom metric

from harness_evals.core.metric import BaseMetric
from harness_evals.core.score import Score
from harness_evals.core.eval_case import EvalCase

class MyMetric(BaseMetric):
    def __init__(self, threshold: float = 0.8):
        super().__init__(name="my_metric", threshold=threshold)

    def measure(self, eval_case: EvalCase) -> Score | None:
        value = compute_something(eval_case)  # return 0.0–1.0
        return Score(name=self.name, value=value, threshold=self.threshold)
        # return None to skip this case (excluded from aggregation)

Custom sink

from harness_evals.core.sink import BaseSink

class MySink(BaseSink):
    def write(self, scores, eval_case):
        for s in scores:
            send_to_my_system(s.name, s.value, s.passed)

Plugin registration

Third-party packages can register metrics, sinks, targets, and source adapters via decorators or entry points.

Decorator (in-process):

from harness_evals.plugins import register_metric

@register_metric("my_custom_metric")
class MyCustomMetric(BaseMetric):
    ...

Entry points (distributable packages):

# In your package's pyproject.toml
[project.entry-points."harness_evals.metrics"]
my_custom = "my_package.metrics:MyCustomMetric"

[project.entry-points."harness_evals.dataset_sources"]
my_source = "my_package.sources:MyDatasetSource"

Eight plugin families are supported: dataset_sources, prompt_sources, eval_case_sources, eval_config_sources, targets, metrics, baseline_stores, sinks.

Registered metrics appear in catalog() and are referenceable by kind: in YAML configs. Registered targets are declarable by type: in YAML target blocks.

Documentation

• Architecture — system design, data flow, extension points
• Metrics Guide — how to write a new metric, templates for every category
• Integration Guide — pytest, GitHub Actions, Harness CI, GitLab CI
• Cost Guide — pricing observed token usage from a caller-supplied rate card
• Contributing — development workflow, code style, PR process
• Framework Integrations — runnable examples for LangChain, CrewAI, DSPy, LlamaIndex, OpenAI Agents, Pydantic AI, Strands, Bedrock AgentCore, Google ADK, LiteLLM, and more
• Architecture Decision Records — why we made key design choices
• Changelog — version history

Development

git clone git@github.com:harness/harness-evals.git
cd harness-evals
pip install poetry
poetry install --all-extras --with dev
poetry run ruff check src/ tests/
poetry run ruff format --check src/ tests/
poetry run pytest tests/ -v

The developer dependencies are a Poetry group, not a pip extra. Both .github/workflows/ci.yml and the test job in .github/workflows/publish.yml run on Python 3.10, 3.11, and 3.12 with all optional extras installed. To match the workflow test command, run poetry run pytest tests/ -v --ignore=tests/sinks/test_sinks.py for each version; keep the full-suite command above as an additional local check.

Before a release, run poetry check and poetry run python -m build. The latter uses an isolated build environment, like the publish workflow. Verify the built wheel's public imports and metadata without importing the source checkout. Publishing is triggered only by a v*.*.* tag whose version matches pyproject.toml; a version bump or merge does not publish. Every release version also needs a matching CHANGELOG.md entry. Do not create a release tag or publish as part of ordinary local validation; local checks do not verify PyPI publishing permissions or trusted-publisher configuration.

References

• Rabanser, Kapoor, Kirgis, Liu, Utpala, Narayanan. "Towards a Science of AI Agent Reliability". Princeton, 2026. — Defines the 12 reliability metrics across 4 dimensions (consistency, robustness, predictability, safety) that inform this framework's reliability metric design.
• DeepEval — LLM evaluation framework. Influenced the measure() / a_measure() metric interface pattern.
• RAGAS — RAG evaluation toolkit. Informed the RAG metric decomposition (faithfulness, context precision/recall).
• promptfoo — CLI-first eval tool. Validated the CI/CD-native approach (JUnit output, exit codes, baseline regression).

License

Apache 2.0 — see LICENSE.