Skip to content

tdfs4ds.agent.evaluation

tdfs4ds.agent.evaluation

Agent evaluation: scenario generation, run, LLM-as-judge, and reporting.

Self-contained evaluation harness for the tdfs4ds consumer agent. It does NOT use an external eval framework (RAGAS etc.) — the judge reuses the configured instruct model via the agent's own _llm_call seam, and scenario "expected answers" are MCP/Teradata-grounded structured facts (catalog rows), not gold prose.

Public surface

  • :func:generate_scenarios — build a scenario set grounded in the live catalog.
  • :func:run_evaluation — run the agent over every turn and judge each.
  • :func:build_report — render a Markdown or HTML synthesis + detail report.
  • :func:evaluate_agent — programmatic entry point: load a scenario file, run, judge, and write the report. Re-exported as tdfs4ds.evaluate_agent.
  • :func:load_scenarios / :func:save_scenarios / :func:merge_scenarios — import/export helpers (replace vs append).

The on-disk results schema mirrors the existing evaluation_*.json format: {type, timestamp, model, tdfs4ds_version, n_scenarios, n_turns, summary[], turns[]} with five 1–5 metrics (faithfulness, completeness, relevance, robustness, efficiency).

logger module-attribute

logger = getLogger(__name__)

METRICS module-attribute

METRICS = ['faithfulness', 'completeness', 'relevance', 'robustness', 'efficiency']

_NUM_WORDS module-attribute

_NUM_WORDS = {0: ('zero', 'zéro'), 1: ('one', 'un', 'une'), 2: ('two', 'deux'), 3: ('three', 'trois'), 4: ('four', 'quatre'), 5: ('five', 'cinq'), 6: ('six',), 7: ('seven', 'sept'), 8: ('eight', 'huit'), 9: ('nine', 'neuf'), 10: ('ten', 'dix'), 11: ('eleven', 'onze'), 12: ('twelve', 'douze'), 13: ('thirteen', 'treize'), 14: ('fourteen', 'quatorze'), 15: ('fifteen', 'quinze'), 16: ('sixteen', 'seize'), 17: ('seventeen', 'dix-sept'), 18: ('eighteen', 'dix-huit'), 19: ('nineteen', 'dix-neuf'), 20: ('twenty', 'vingt')}

_AS_OF_MARKERS module-attribute

_AS_OF_MARKERS = ('as it was on', "tel qu'il était", 'valid time', 'temps de validité')

_DENIAL module-attribute

_DENIAL = ('not found', 'not registered', 'does not exist', 'did not exist', 'no such', "n'est pas enregistr", "n'existait pas", "n'existe pas", 'introuvable', 'non trouvé')

_fetch_catalog_facts

_fetch_catalog_facts() -> dict

Pull ground-truth catalog facts used to generate scenarios + expected_facts.

One read each of the feature catalog, process catalog, and follow-up history, materialised into plain dicts. Fail-open: any read that fails yields an empty structure so generation still produces what it can.

_historian_facts

_historian_facts(col: dict) -> dict

Expected facts stating that a documented name is a TAG of a deep (historian) table.

Such a name is a value of the table's tag column, read from its payload column by its id columns and timestamp. The agent says so (that is the documented layout), and a judge that only sees 'column X of table T' scored it as invented. Empty when the column is an ordinary one.

_generate_fact_first

_generate_fact_first(facts: dict, *, facts_per_kind: int = 2, multi_turn: bool = True, start_id: int = 0) -> list

Fact-first ("Jeopardy") scenarios for non-feature OBJECTS and COLUMNS.

Harvests the answer (a grounded fact) first, then writes the question around it: object/column DEFINITION + LINEAGE, plus a column→object→lineage COMPOUND turn. Grounded in BD_OBJECTS / BD_COLUMNS + FS_LINEAGE_GRAPH neighbors. Returns a list of scenario dicts; empty when no non-feature objects/columns are documented.

generate_scenarios

generate_scenarios(skills: Optional[list] = None, *, multi_turn: bool = True, use_pronouns: bool = True, per_skill: int = 1, categories: Optional[list] = None, fact_first: bool = True, facts_per_kind: int = 2, temporal: bool = True, seed: int = 0, include_writes: bool = False, mcp_enabled: bool = False, store: Optional[dict] = None, store_at=None) -> dict

Generate a scenario set grounded in the store, including time (0.3.1.240).

Questions are chosen from data (see :mod:tdfs4ds.agent.eval_scenarios): every turn carries a fact_spec and the expected_facts built from it by :func:tdfs4ds.agent.eval_facts.build_facts, plus a time_context (the valid time the turn is about, under the agent's own date rules).

Parameters:

Name Type Description Default
skills Optional[list]

Intent names to cover (default: every intent; ALGORITHM and BUILD_DATASET are reported as skipped unless possible).

None
multi_turn bool

Also emit follow-up scenarios.

True
use_pronouns bool

Follow-up turns use a pronoun (no explicit name).

True
per_skill int

Distinct objects per skill.

1
categories Optional[list]

Restrict to standard, multi_turn, temporal, object, column; default all.

None
fact_first bool

Add the object / column scenarios from the business dictionary.

True
temporal bool

Add the time scenarios (dates between registrations, before the store, a date kept then cleared, freshness / lineage at a date, a dataset's own time, a year alone).

True
seed int

Deterministic choice of the objects asked about.

0
include_writes bool

Also generate BUILD_DATASET (creates a dataset).

False
store, store_at

The store (eval_facts.collect_eval_facts()) and a callable(as_of) reading it at a valid time; read from the database when omitted (tests pass frames).

required

Returns:

Type Description
dict

``{"scenarios": [...], "meta": {..., "coverage": {intent: "generated" |

dict

"skipped: "}}}``.

_scenario_hash

_scenario_hash(scenarios: list) -> str

Short content fingerprint of a scenario set (stable across reorderings).

Covers the questions AND their expected facts: a set whose facts were edited or regenerated on a changed store gets another hash.

load_scenarios

load_scenarios(path: Union[str, Path]) -> dict

Load a scenario set from JSON. Tolerates the legacy turns[]-only shape.

save_scenarios

save_scenarios(scenario_set: dict, path: Union[str, Path]) -> str

Write a scenario set to JSON; returns the path.

merge_scenarios

merge_scenarios(base: dict, incoming: dict, mode: str = 'append') -> dict

Merge two scenario sets. mode='replace' returns incoming; 'append' unions by scenario id (incoming wins on conflict).

_verified_names_from_context

_verified_names_from_context(graph_context: dict) -> set

Collect every real object/feature/view/dataset name from a turn's accumulated graph context, so the judge can tell real-but-unlisted names apart from fabrications.

The agent records what it actually grounded on in accumulated_context_graph (features/datasets/objects/columns + per-feature view_name). Those names are authoritative — expected_facts is only a focal spot-check, not the complete truth, so anything here must NOT be judged a hallucination. Returns an UPPER-cased set; empty when no context (fail-open).

_has_number

_has_number(answer: str, n) -> bool

A count in the answer: digits (also '1,234' / '1 234') or a word, EN or FR.

_has_date

_has_date(answer: str, value) -> bool

_has_name

_has_name(answer: str, name) -> bool

_body

_body(answer: str) -> str

The answer without its valid-time header line (which carries a date that would otherwise match any expected date).

deterministic_checks

deterministic_checks(answer: str, facts: dict, time_context: Optional[dict] = None, actual_as_of: Optional[str] = None) -> dict

Facts checked by code, before the judge reads the answer. Pure.

Returns {"hits", "misses", "violations", "as_of_match", "as_of_stated"}: hits / misses for required names, figures and dates; violations for a name that must not appear (an object that did not exist at the turn's valid time, a catalog name in an out-of-scope answer). Wording is free: a figure counts in digits or words, a date in ISO or textual EN/FR form.

_judge_turn

_judge_turn(user: str, answer: str, expected_facts: dict, expected_intent: str, actual_intent: str, tokens: int, graph_context: Optional[dict] = None, time_context: Optional[dict] = None, actual_as_of: Optional[str] = None, run_now: Optional[str] = None, store_first: Optional[str] = None, acceptable_intents: Optional[list] = None) -> dict

Score one turn on the five metrics with an LLM-as-judge grounded on facts.

Deterministic signals (intent match, fact presence in the answer) are computed first and injected into the judge prompt so the LLM grades against ground truth rather than its own priors.

graph_context is the turn's accumulated_context_graph — the names the agent actually grounded on. It is the second authoritative source: a name the answer mentions that is in EITHER expected_facts OR this verified set is real, not a hallucination. expected_facts alone is a focal spot-check, not a closed-world checklist (so set-returning skills — SUMMARY/FEASIBILITY/DATASET — are not penalised for correctly enumerating many real assets).

_time_rules

_time_rules(tc: dict, facts: dict, run_now: Optional[str], store_first: Optional[str]) -> str

The judge's time rules for this turn: the same as the agent's, stated.

_refresh_turn_facts

_refresh_turn_facts(scenarios: list, mode: str, log) -> dict

Rebuild each turn's facts from its fact_spec on the store as it is now (at the spec's valid time for a dated turn). Returns {(sid, turn): info} with stale (facts changed), facts (the rebuilt ones) or error. mode='off' (or a legacy file without specs) does nothing.

_db_now

_db_now() -> Optional[str]

_store_first

_store_first(scenarios: list) -> Optional[str]

When the store was first populated: the day after a 'before_store' date.

run_evaluation

run_evaluation(scenario_set: dict, *, mcp_enabled: bool = False, progress=None, refresh_facts: str = 'check') -> dict

Run the agent over every scenario turn and judge each.

Parameters:

Name Type Description Default
scenario_set dict

{"scenarios": [...]} (from :func:generate_scenarios or :func:load_scenarios).

required
mcp_enabled bool

Pass-through to the agent (enables MCP-gated skills).

False
progress

Optional callable(msg: str) for streaming progress lines.

None
refresh_facts str

'check' (default) rebuilds each turn's facts from its fact_spec and flags the turns whose facts no longer match the store (judged on the stored facts, marked STALE); 'refresh' judges on the rebuilt facts; 'off' uses the stored facts as they are.

'check'

Returns:

Type Description
dict

Results dict in the llm_as_judge_evaluation schema (summary +

dict

turns).

_intent_from_trace

_intent_from_trace(trace: str) -> str

_as_of_from_trace

_as_of_from_trace(trace: str) -> Optional[str]

The valid time the agent answered at (the trace's **As of** line); 'current' for now, None when the trace predates that line.

_tokens_from_trace

_tokens_from_trace(trace: str) -> int

build_report

build_report(results: dict, fmt: str = 'md') -> str

Render a synthesis + detail report. fmt is 'md' or 'html'.

_avg

_avg(values: list) -> Optional[float]

_build_report_md

_build_report_md(results: dict) -> str

_render_graph_context_md

_render_graph_context_md(acc: dict) -> str

Compact Markdown rendering of the accumulated graph context for a turn.

_score_color

_score_color(s) -> str

Map a 1-5 score to a CSS color string.

_metric_chip

_metric_chip(label: str, score) -> str

Tiny colored badge for a single metric score.

_build_report_html

_build_report_html(results: dict) -> str

Build a rich visual HTML report with metric cards, per-skill charts, and collapsible turns.

_md_to_html

_md_to_html(md: str, results: dict) -> str

Render evaluation results as a rich styled HTML page (calls _build_report_html).

evaluate_agent

evaluate_agent(scenario_file: Union[str, Path], out_path: Optional[Union[str, Path]] = None, *, fmt: str = 'md', mcp_enabled: bool = False, results_path: Optional[Union[str, Path]] = None, progress=None, refresh_facts: str = 'check') -> dict

Load a scenario file, run the agent, judge, and write the report.

refresh_facts: see :func:run_evaluation ('check' flags turns whose facts no longer match the store).

Parameters:

Name Type Description Default
scenario_file Union[str, Path]

Path to a scenario JSON (generate_scenarios output or a legacy evaluation_*.json).

required
out_path Optional[Union[str, Path]]

Where to write the report. Defaults to agent_eval_report_<ts>.<fmt> next to scenario_file.

None
fmt str

'md' or 'html'.

'md'
mcp_enabled bool

Enable MCP-gated skills during the run.

False
results_path Optional[Union[str, Path]]

Optional path to also dump the raw results JSON.

None
progress

Optional callable(msg) for streaming progress.

None

Returns:

Type Description
dict

{"results": <results dict>, "report_path": str, "results_path": str|None}.