Pre-built Evaluation Flows¶

Latent ships a library of composable evaluation flows that handle scoring, statistical analysis, quality gating, and MLflow logging out of the box. All flows live in latent.flows and require the [eval] extra:

pip install "latent[eval]"

Every flow returns a dict with structured results, a StatisticalReport, and rendered Markdown. All flows auto-log metrics to MLflow when an active run exists.

Eval data from Agent Studio

Flow inputs can come from local files or straight from Agent Studio: use a type: agent_studio.Dataset catalog entry to pull a versioned eval set by slug, and a type: agent_studio.Report output to post the scored StatisticalReport back. See Data Catalog.

Choosing a Flow¶

Task	Flow	Input
Score free-text outputs with LLM judge	`judge_flow`	DataFrame
Binary/multi-class classification	`classification_flow`	DataFrame + `ground_truth` column
Compare model A vs. baseline B	`comparison_flow`	DataFrame + baseline score arrays
Multi-turn conversation quality	`conversation_scoring_flow`	`list[Conversation]`
SOP/checklist compliance	`conversation_sop_flow`	`list[Conversation]` + checklist
Turn-score trajectory analysis	`conversation_trajectory_flow`	`list[Conversation]` with scores
Simulate + score conversations	`conversation_simulation_flow`	Agent class + scenarios
End-to-end agent evaluation	`agent_eval_flow`	DataFrame + agent factory + judge
Agent inference only (no scoring)	`agent_inference_flow`	DataFrame + agent factory
Retrieval/RAG quality	`retrieval_eval_flow`	DataFrame + retriever
Score distribution drift	`drift_flow`	Two score dicts
Sweep multiple models (Pareto)	`model_garden_flow`	DataFrame + judge factory + models
Compare multiple agents	`agent_garden_flow`	DataFrame + agent factory + models
Judge scoring + latency/token triad	`instrumented_judge_flow`	DataFrame + judge
Named-entity recognition	`ner_flow`	DataFrame with span columns
Text-to-SQL generation	`text_to_sql_flow`	DataFrame with `generated_sql`
KB coverage/sufficiency	`knowledge_coverage_flow`	DataFrame + retriever
Aggregate raw results into a report	`eval_report_flow`	`list[dict]`
Classify failure modes	`classify_failure_modes`	`list[RecordResult]` + taxonomy
Autonomous code optimization	`autoresearch_agent_flow`	Eval subflow + dataset
Red-team guardrails	`guardrail_redteam`	`list[dict]` + scanners

Scoring Flows¶

`judge_flow`¶

Score every row in a DataFrame with one or more LLM judges, run bootstrap statistical analysis, and check quality gates.

Signature:

async def judge_flow(
    eval_data: pd.DataFrame,
    judges: Judge[T] | list[Judge],
    *,
    concurrency: int = 10,
    gates: dict[str, float] | None = None,
    score_types: dict[str, str] | None = None,
    rubrics: dict | None = None,
    n_resamples: int = 10_000,
    confidence_level: float = 0.95,
    seed: int | None = None,
    summary_model: str | None = None,
    summary_label: str | None = None,
) -> dict[str, Any]

Returns: scored_data (DataFrame), report (StatisticalReport), markdown (str), all_passed (bool).

judges accepts a single Judge or a list of judges (multi-judge scoring). Each judge scores rows concurrently (bounded by concurrency) and runs sequentially across judges; every judge's output columns are concatenated onto one scored_data frame. A judge exposing an is_eligible(row) -> bool method is only applied to the rows where it returns True (e.g. a faithfulness judge can skip rows without context); to avoid column collisions between judges that share a field name, set a distinct column_prefix= on at least one of them.

Score types and rubrics are auto-detected from each judge's output_type annotations. Pass score_types or rubrics explicitly to override.

Parameter	Default	Purpose
`concurrency`	`10`	Rows scored concurrently per judge (an `asyncio.Semaphore` bound)
`summary_model`	`None`	Model used to generate a natural-language summary of the report
`summary_label`	`None`	Label/title attached to the generated summary

from typing import Annotated
from latent.agents import Judge
from latent.agents.scores import ScoredModel, OrdinalScore
from latent.flows import judge_flow

class QAScores(ScoredModel):
    quality: Annotated[int, OrdinalScore(scale=(1,2,3,4,5), pass_threshold=3)]

judge = Judge("qa_judge", model="gpt-4o", output_type=QAScores)
result = await judge_flow(eval_data=df, judges=judge, gates={"quality": 3.5})

print(result["markdown"])       # Statistical report with CIs
assert result["all_passed"]     # Gate check

Pass a list to score with several judges in one pass:

result = await judge_flow(
    eval_data=df,
    judges=[quality_judge, faithfulness_judge],
    concurrency=20,
    gates={"quality": 3.5, "faithfulness": 4.0},
    summary_model="gpt-4o",
)

Auto-detected configuration

ScoredModel fields annotated with OrdinalScore, BinaryScore, or ContinuousScore are automatically extracted as score types and rubrics. You rarely need to pass score_types manually.

`comparison_flow`¶

Score a dataset and compare against baseline scores. Wraps judge_flow with paired comparison statistics.

Signature:

async def comparison_flow(
    eval_data: pd.DataFrame,
    judge: Judge[T],
    *,
    baseline_scores: dict[str, np.ndarray] | None = None,
    gates: dict[str, float] | None = None,
    score_types: dict[str, str] | None = None,
    rubrics: dict | None = None,
    n_resamples: int = 10_000,
    confidence_level: float = 0.95,
    seed: int | None = None,
) -> dict[str, Any]

Returns: scored_data, report, markdown, all_passed.

If baseline_scores is None, the flow runs in first-run mode (no comparison, just scoring).

from latent.flows import comparison_flow
import numpy as np

# First run: establish baseline
baseline_result = await judge_flow(eval_data=df_v1, judges=judge)
baseline = {
    "quality": baseline_result["scored_data"]["quality"].values
}

# Second run: compare against baseline
result = await comparison_flow(
    eval_data=df_v2,
    judge=judge,
    baseline_scores=baseline,
    gates={"quality": 3.5},
)
# result["report"] includes paired CIs, effect sizes, p-values

Classification Flows¶

`classification_flow`¶

Run a Classifier on a dataset and compute accuracy, precision, recall, and F1 with confidence intervals.

Signature:

async def classification_flow(
    eval_data: pd.DataFrame,
    classifier: Classifier[T],
    *,
    labels: list[str] | None = None,
    gates: dict[str, float] | None = None,
    confidence_level: float = 0.95,
    n_resamples: int = 10_000,
    seed: int | None = None,
) -> dict[str, Any]

Returns: predictions_data (DataFrame), metrics (dict of MetricResult), all_passed (bool).

Required column

The input DataFrame must have a ground_truth column. The classifier's output model must have a prediction field.

from pydantic import BaseModel
from latent.agents import Classifier
from latent.flows import classification_flow

class Intent(BaseModel):
    prediction: str

classifier = Classifier(
    "intent",
    model="gpt-4o",
    output_type=Intent,
    prompt_template="Classify intent: {text}\nLabels: {labels}",
)

result = await classification_flow(
    eval_data=df,  # must have "ground_truth" column
    classifier=classifier,
    labels=["billing", "support", "sales"],
    gates={"accuracy": 0.85, "f1": 0.80},
)
print(result["metrics"]["accuracy"].point_estimate)

Conversation Flows¶

`conversation_scoring_flow`¶

Score multi-turn conversations at the turn level, aggregate per conversation, and run statistical analysis.

Signature:

async def conversation_scoring_flow(
    conversations: list[Conversation],
    judge: Judge[T],
    *,
    aggregation: str = "mean",  # "mean" | "min" | "max" | "last"
    gates: dict[str, float] | None = None,
    n_resamples: int = 10_000,
    confidence_level: float = 0.95,
    seed: int | None = None,
) -> dict[str, Any]

Returns: scored_conversations (list), aggregated_scores (dict), report, markdown.

Only assistant turns are scored. The aggregation parameter controls how per-turn scores collapse into a single per-conversation value.

from latent.stats.conversation import Conversation, Turn
from latent.flows import conversation_scoring_flow

convos = [
    Conversation(turns=[
        Turn(role="user", content="How do I reset my password?"),
        Turn(role="assistant", content="Go to Settings > Security > Reset Password."),
    ]),
]

result = await conversation_scoring_flow(
    convos,
    judge=judge,
    aggregation="mean",
    gates={"quality": 3.0},
)

Aggregation	Behavior
`mean`	Average of all turn scores
`min`	Worst turn determines conversation score
`max`	Best turn determines conversation score
`last`	Final assistant turn score only

`conversation_sop_flow`¶

Check conversations against an SOP (Standard Operating Procedure) compliance checklist.

Signature:

async def conversation_sop_flow(
    conversations: list[Conversation],
    judge: Judge[T],
    *,
    sop_checklist: list[dict[str, Any]],
    confidence_level: float = 0.95,
    n_resamples: int = 10_000,
    seed: int | None = None,
) -> dict[str, Any]

Returns: compliance_results (list of per-conversation dicts), aggregate_rates (per-item stats with CIs), report, markdown.

Each checklist item is a dict with name, description, and required keys. The judge evaluates each item against the full conversation text.

from latent.flows import conversation_sop_flow

checklist = [
    {"name": "greeting", "description": "Agent greets the customer", "required": True},
    {"name": "verify_identity", "description": "Agent verifies customer identity", "required": True},
    {"name": "resolution", "description": "Agent provides a resolution or next steps", "required": True},
]

result = await conversation_sop_flow(
    conversations=convos,
    judge=judge,
    sop_checklist=checklist,
)
# result["aggregate_rates"]["greeting"]["mean"] -> 0.92

`conversation_trajectory_flow`¶

Analyze how scores evolve across turns. Computes L2 distances between conversation trajectories (or against a reference pattern).

Signature:

def conversation_trajectory_flow(
    conversations: list[Conversation],
    *,
    score_key: str = "score",
    reference_pattern: list[float] | None = None,
    confidence_level: float = 0.95,
    n_resamples: int = 10_000,
    seed: int | None = None,
) -> dict[str, Any]

Returns: trajectories (list of score sequences), distances (ndarray), report, markdown.

Conversations must have turn scores in their metadata (typically from a prior conversation_scoring_flow run). If reference_pattern is provided, each trajectory is compared to it; otherwise pairwise distances are computed.

from latent.flows import conversation_trajectory_flow

# Ideal pattern: quality should stay high across 5 turns
result = conversation_trajectory_flow(
    scored_conversations,
    score_key="quality",
    reference_pattern=[4.0, 4.0, 4.0, 4.0, 4.0],
)

`conversation_simulation_flow`¶

Simulate conversations between an agent under test and a synthetic human, then optionally score the results.

Signature:

async def conversation_simulation_flow(
    agent_class: type,
    agent_kwargs: dict[str, Any] | None = None,
    *,
    scenarios: pd.DataFrame | None = None,
    human_class: type | None = None,
    human_kwargs: dict[str, Any] | None = None,
    judge_class: Callable[[], Judge[T]] | None = None,
    judge_scoring: JudgeScoringConfig | None = None,
    simulation_guardrails: SimulationGuardrails | None = None,
    context_column: str = "context",
) -> dict[str, Any]

Returns: simulation_results (DataFrame with turn data), report (scoring dict or None).

If scenarios is not provided, the flow loads them from the catalog. If judge_class is provided, conversations are scored automatically after simulation.

from latent.flows import conversation_simulation_flow

result = await conversation_simulation_flow(
    agent_class=MyAgent,
    agent_kwargs={"model": "gpt-4o"},
    scenarios=scenarios_df,          # must have "context" column
    judge_class=lambda: judge,       # factory for scoring
    judge_scoring={"scope": "conversation", "gates": {"quality": 3.5}},
    simulation_guardrails={"max_turns_per_conversation": 20},
)

Scoring scope

Set judge_scoring["scope"] to "conversation" to score each conversation as a whole, or "turn" to score each assistant turn individually and aggregate. Quality gates are auto-derived from the judge's output_type pass_threshold annotations and can be overridden via judge_scoring["gates"].

Agent Evaluation Flows¶

`agent_eval_flow`¶

Full end-to-end agent evaluation pipeline: inference, LLM judging, optional deterministic scorers, category breakdowns, and failure mode classification.

Signature:

async def agent_eval_flow(
    eval_data: pd.DataFrame,
    agent_factory: Callable[[], BaseAgent],
    judge: Judge[T],
    *,
    question_column: str = "question",
    expected_column: str | None = None,
    category_column: str | None = None,
    id_column: str | None = None,
    context_columns: list[str] | None = None,
    context_formatter: Callable[[dict], str] | None = None,
    gates: dict[str, float] | None = None,
    scorers: list[Callable] | None = None,
    failure_taxonomy: dict[str, str] | None = None,
    failure_filter_fn: Callable | None = None,
    concurrency: int = 1,
    n_resamples: int = 10_000,
    confidence_level: float = 0.95,
    seed: int | None = None,
) -> dict[str, Any]

Returns: inference_results (list), scored_data (DataFrame), report (StatisticalReport), markdown (str), all_passed (bool).

This flow orchestrates five stages:

Inference -- run the agent on every row via agent_inference_flow
Judging -- score outputs with an LLM judge via judge_flow
Deterministic scoring -- run optional scorer functions
Category breakdown -- slice metrics by category column
Failure classification -- classify failure modes on low-scoring records

from latent.flows import agent_eval_flow

result = await agent_eval_flow(
    eval_data=df,
    agent_factory=lambda: MyRAGAgent(model="gpt-4o"),
    judge=judge,
    question_column="question",
    expected_column="expected_answer",
    category_column="topic",
    gates={"quality": 3.5, "faithfulness": 4.0},
    failure_taxonomy={
        "hallucination": "Agent fabricated information not in sources",
        "refusal": "Agent refused to answer when it should have",
        "incomplete": "Agent gave a partial answer missing key details",
    },
    concurrency=5,
)

Deterministic scorers

The scorers parameter accepts callables with signature (scored_data: DataFrame, inference_results: list[dict]) -> dict[str, MetricResult]. Use these for exact-match, BLEU, ROUGE, or any non-LLM metric.

`agent_inference_flow`¶

Run an agent on every row of a DataFrame and collect outputs, tool calls, token counts, and latency. Uses Prefect task.map() for concurrent execution.

Signature:

async def agent_inference_flow(
    eval_data: pd.DataFrame,
    agent_factory: Callable[[], BaseAgent],
    *,
    question_column: str = "question",
    context_columns: list[str] | None = None,
    context_formatter: Callable[[dict], str] | None = None,
    id_column: str | None = None,
    concurrency: int = 1,
    include_events: bool = False,
) -> dict[str, Any]

Returns: inference_results (list of dicts with output, thinking, tool_calls, tool_results, latency_ms, input_tokens, output_tokens, error), summary (aggregate counts and mean latency).

from latent.flows import agent_inference_flow

result = await agent_inference_flow(
    eval_data=df,
    agent_factory=lambda: MyAgent(model="gpt-4o"),
    question_column="prompt",
    concurrency=10,
)
print(result["summary"])
# {"total_count": 200, "success_count": 198, "error_count": 2, ...}

Agent factory isolation

Each Prefect task creates its own agent via agent_factory(). Never pass a shared agent instance -- the factory pattern prevents mutable state leaking between concurrent tasks.

RAG Evaluation Flows¶

`retrieval_eval_flow`¶

Evaluate retrieval quality independently of generation. For each question+expected answer pair, retrieves chunks and optionally classifies answerability, grades coverage, and labels chunk relevance.

Signature:

async def retrieval_eval_flow(
    eval_data: pd.DataFrame,
    retriever: Any,
    *,
    question_column: str = "question",
    expected_column: str = "expected",
    id_column: str | None = None,
    k: int = 10,
    judge_model: str = "claude-sonnet-4-5-20250929",
    answerability_check: bool = True,
    chunk_relevance_labels: bool = True,
    coverage_grading: bool = True,
    n_resamples: int = 10_000,
    confidence_level: float = 0.95,
    seed: int | None = None,
) -> dict[str, Any]

Returns: results (list of per-row dicts), report (StatisticalReport with coverage metrics), markdown (str), relevant_chunk_ids (dict mapping row ID to relevant chunk IDs).

The retriever must implement a .search(query, k=k) method returning objects with content, source, and optionally id and score attributes.

from latent.flows import retrieval_eval_flow

result = await retrieval_eval_flow(
    eval_data=df,         # columns: question, expected
    retriever=my_chroma,  # .search(query, k) -> list[Chunk]
    k=10,
    judge_model="gpt-4o",
)

# Per-row coverage grades (1-5)
for r in result["results"]:
    print(f"{r['id']}: relevance={r.get('relevance_grade')}, "
          f"info_match={r.get('info_match_grade')}")

Evaluation dimensions:

Dimension	Toggle	What it measures
Answerability	`answerability_check`	Whether the question needs retrieval at all
Coverage grading	`coverage_grading`	Relevance (1-5) and info completeness (1-5) of retrieved chunks
Chunk relevance	`chunk_relevance_labels`	Per-chunk labels: `relevant`, `contradicts`, or `not_relevant`

Drift Detection¶

`drift_flow`¶

Detect distribution drift between a reference (baseline) and current set of scores.

Signature:

def drift_flow(
    reference_scores: dict[str, np.ndarray],
    current_scores: dict[str, np.ndarray],
    *,
    score_types: dict[str, str] | None = None,
    threshold: float = 0.05,
    confidence_level: float = 0.95,
    seed: int | None = None,
) -> dict[str, Any]

Returns: drift_results (list of DriftResult), drift_detected (bool), report_text (str).

Drift is flagged when any metric's p-value falls below threshold (default 0.05). Each DriftResult includes delta, p_value, effect_size, and severity.

import numpy as np
from latent.flows import drift_flow

result = drift_flow(
    reference_scores={"quality": np.array([4, 5, 4, 3, 5, 4])},
    current_scores={"quality": np.array([3, 2, 3, 4, 2, 3])},
    threshold=0.05,
)

if result["drift_detected"]:
    print(result["report_text"])

More Eval Flows¶

Specialized flows for multi-system sweeps, structured-output tasks, instrumentation, and red-teaming. All share the common statistical parameters (n_resamples, confidence_level, seed) and return a report plus rendered markdown unless noted otherwise.

`model_garden_flow`¶

Evaluate the same dataset across multiple models with one judge per model, then compute a Pareto frontier over the accuracy/latency/tokens triad. Each model runs through instrumented_judge_flow.

async def model_garden_flow(
    eval_data, judge_factory, models, *,
    gates=None, metric_directions=None, score_types=None, rubrics=None,
    n_resamples=10_000, confidence_level=0.95, seed=None,
    pareto_x=None, pareto_y=None, save_plots=None,
) -> dict

judge_factory is Callable[[str], Judge] (model name -> judge). Returns: per_model, pareto, summary_table, figures.

from latent.flows import model_garden_flow
result = await model_garden_flow(
    df, judge_factory=lambda m: Judge("scorer", model=m, output_type=Scores),
    models=["gpt-4o", "gpt-4o-mini"], pareto_x="latency_seconds", pareto_y="quality",
)

`agent_garden_flow`¶

The agent-under-test analogue of model_garden_flow: runs agent_eval_flow for each agent model and merges the per-variant reports into one report_mode="comparison" StatisticalReport. The first model is the baseline.

async def agent_garden_flow(
    eval_data, agent_factory_for_model, judge, models, *,
    question_column="question", expected_column=None, category_column=None,
    id_column=None, context_columns=None, context_formatter=None,
    gates=None, scorers=None, failure_taxonomy=None, failure_filter_fn=None,
    concurrency=1, n_resamples=10_000, confidence_level=0.95, seed=None,
    summary_model=None, summary_label=None,
) -> dict

agent_factory_for_model is Callable[[str], Callable[[], BaseAgent]] (the double-callable mirrors agent_eval_flow). Returns: per_model, report, markdown, all_passed.

from latent.flows import agent_garden_flow
result = await agent_garden_flow(
    df, agent_factory_for_model=lambda m: lambda: make_agent(model=m),
    judge=judge, models=["gemini/gemini-2.5-flash", "gemini/gemini-2.5-flash-lite"],
    gates={"accuracy": 0.6},
)

`instrumented_judge_flow`¶

Like judge_flow, but additionally captures per-call latency and token usage, reporting accuracy + latency + tokens with CIs (mean plus p50/p90/p99 quantiles). Latency/token metrics are auto-marked lower_is_better.

async def instrumented_judge_flow(
    eval_data, judge, *,
    gates=None, metric_directions=None, score_types=None, rubrics=None,
    n_resamples=10_000, confidence_level=0.95, seed=None,
) -> dict

Returns: scored_data, report, markdown, all_passed, call_metrics (per-row CallMetrics), supplementary (quantile CIs).

from latent.flows import instrumented_judge_flow
result = await instrumented_judge_flow(df, judge, gates={"quality": 3.5})

`ner_flow`¶

Span-level named-entity-recognition evaluation: precision, recall, and F1 with confidence intervals. Input DataFrame columns hold lists of span dicts (start, end, label). Synchronous.

def ner_flow(
    eval_data, *,
    predicted_column="predicted_spans", gold_column="gold_spans",
    match_mode="exact", iou_threshold=0.5, labels=None, gates=None,
    confidence_level=0.95, n_resamples=10_000, seed=None,
) -> dict

match_mode is "exact", "overlap", or "iou". Returns: metrics, report, markdown, all_passed.

from latent.flows import ner_flow
result = ner_flow(df, match_mode="iou", iou_threshold=0.5)

`text_to_sql_flow`¶

Text-to-SQL evaluation: SQL validity (parse and optional execution) plus optional result-set matching. Input DataFrame must have a generated_sql column. Synchronous.

def text_to_sql_flow(
    eval_data, *,
    dialect="ansi", connection=None, numeric_tolerance=0.0,
    gold_results_column=None, predicted_results_column=None, gates=None,
    confidence_level=0.95, n_resamples=10_000, seed=None,
) -> dict

Pass gold_results_column + predicted_results_column to enable result-set matching. Returns: metrics, report, markdown, all_passed.

from latent.flows import text_to_sql_flow
result = text_to_sql_flow(df, dialect="postgres", connection=conn)

`knowledge_coverage_flow`¶

Two-pass KB-sufficiency eval: extract atomic claims from expected answers, then ground each claim against retrieved chunks, yielding a per-row coverage_score (supported / total claims).

async def knowledge_coverage_flow(
    eval_data, retriever, *,
    expected_column="expected", question_column="question", id_column=None,
    k=20, model="claude-sonnet-4-5-20250929", batch_size=5,
    n_resamples=10_000, confidence_level=0.95, seed=None,
) -> dict

retriever exposes .search(query, k). Returns: coverage_results, report, markdown, relevant_chunk_ids.

from latent.flows import knowledge_coverage_flow
result = await knowledge_coverage_flow(df, retriever=my_retriever, k=20)

`eval_report_flow`¶

Aggregate raw evaluation result dicts into a complete StatisticalReport — metrics with CIs, category breakdowns, failure modes, gates, optional drift vs. baseline, and optional system comparisons. Synchronous.

def eval_report_flow(
    eval_results, *,
    score_columns=None, question_column="question", output_column="output",
    expected_column=None, category_column=None, id_column=None,
    failure_taxonomy=None, gates=None, baseline_results=None,
    comparison_results=None, n_resamples=10_000, confidence_level=0.95, seed=None,
) -> dict

score_columns maps column -> "binary"|"ordinal"|"continuous" (auto-detected if omitted). Returns: report, markdown, all_passed.

from latent.flows import eval_report_flow
result = eval_report_flow(records, gates={"quality_mean": 3.5})

`classify_failure_modes`¶

LLM-based failure-mode classifier: labels filtered records against a taxonomy. Lower-level helper used internally by agent_eval_flow.

async def classify_failure_modes(
    records, taxonomy, *,
    filter_fn=None, model="claude-sonnet-4-5-20250929",
) -> tuple[list[RecordResult], list[FailureModeSummary]]

records is list[RecordResult]; taxonomy is {mode: description}. filter_fn defaults to records with failure_mode is None. Returns: (updated_records, summaries) — a tuple, not a dict.

from latent.flows import classify_failure_modes
records, summaries = await classify_failure_modes(records, taxonomy)

`autoresearch_agent_flow`¶

Autonomous code optimization: drives an AutoResearchOptimizer over a user-supplied eval subflow with a git keep/discard loop, stratified subsampling, and progressive confirmation.

async def autoresearch_agent_flow(
    eval_flow, brief, dataset, *,
    deployment_name=None, stratify_column="category", sample_size=60,
    confirmation_size=150, min_per_stratum=2, repo_root=Path("."),
    max_iterations=20, patience=5, checks=None, agent_model="claude-opus-4-6",
    agent_max_turns=15, allowed_tools=None, tracker=None, checkpoint_path=None,
    check_timeout=300, eval_timeout=600, secondary_metrics=None,
    system_prompt=None, scope_paths=None,
) -> dict

eval_flow is a Prefect @flow returning a StatisticalReport; brief is a ResearchBrief. Returns: result, best_score, best_label, num_experiments.

from latent.flows import autoresearch_agent_flow
result = await autoresearch_agent_flow(eval_flow=my_eval, brief=brief, dataset=df)

`guardrail_redteam`¶

Red-team a list of adversarial prompts against input/output scanners. Unit mode (scanner-only, no LLM) when agent=None; E2E mode (full agent + GuardrailMiddleware) when an agent is provided. Not exported from latent.flows — import from the submodule.

from latent.flows.guardrail_redteam.flow import guardrail_redteam

async def guardrail_redteam(
    inputs, *,
    agent=None, pre_scanners=None, post_scanners=None,
) -> list[dict]

inputs is a list of {"prompt", "expected_blocked", "label"}. Returns: a list[dict] (not a report) with blocked, violated_rules, score (and response in E2E mode) per case.

from latent.flows.guardrail_redteam.flow import guardrail_redteam
from latent.guardrails.scanners.builtin import LanguageScanner

results = await guardrail_redteam(
    inputs=[{"prompt": "bonjour", "expected_blocked": True, "label": "fr"}],
    pre_scanners=[LanguageScanner(allowed_languages=["en"])],
)

RAG research & optimization flows¶

rag_research_flow (auto-tune chunking/retrieval configs) and rag_optimization_flow (end-to-end RAG sweep) are documented in RAG Evaluation. Both are async and imported from their submodules: from latent.flows.rag_research_flow import rag_research_flow.

Dataset helpers¶

Two utilities from latent.flows support the flows above rather than being eval flows themselves:

validate_dataset(eval_data, ...) — check a DataFrame has the columns/shape a flow expects before running it.
results_to_conversations(results) — convert agent-inference/simulation results into the list[Conversation] the conversation flows consume (used in the simulation-to-scoring pattern).

Quality Gates¶

All scoring flows support quality gates via the gates parameter -- a dict mapping metric names to minimum thresholds. The flow checks each gate against the point estimate from statistical analysis and sets all_passed accordingly.

result = judge_flow(
    eval_data=df,
    judge=judge,
    gates={
        "quality": 3.5,        # mean quality must be >= 3.5
        "faithfulness": 4.0,   # mean faithfulness must be >= 4.0
    },
)

if not result["all_passed"]:
    for gate in result["report"].gates:
        if not gate.passed:
            print(f"FAILED: {gate.metric_name} = {gate.observed:.2f} < {gate.threshold}")

Gate strictness

Flow-level gates use the point estimate by default. For stricter gating based on the lower CI bound, use latent.stats.threshold_gate directly with strictness="lower_ci".

Statistical Analysis Integration¶

Every flow that produces scores calls latent.stats.analyze() internally, which provides:

Bootstrap confidence intervals for all metrics
Paired comparisons when baseline scores are available (via comparison_flow)
Effect sizes (Cohen's d, odds ratio)
Gate evaluation against thresholds
MLflow logging of all metrics, comparisons, and gate results

All flows accept these common statistical parameters:

Parameter	Default	Purpose
`n_resamples`	`10_000`	Bootstrap iterations for CI estimation
`confidence_level`	`0.95`	Confidence interval width
`seed`	`None`	Random seed for reproducibility

Common Patterns¶

Chaining flows¶

Use retrieval_eval_flow to validate retrieval quality before running the full agent_eval_flow:

from latent.flows import retrieval_eval_flow, agent_eval_flow

# Step 1: Check retrieval quality
retrieval_result = await retrieval_eval_flow(
    eval_data=df,
    retriever=my_retriever,
    k=10,
)

# Step 2: Only proceed if coverage is acceptable
coverage = retrieval_result["report"].metrics[0].point_estimate
if coverage >= 3.5:
    eval_result = await agent_eval_flow(
        eval_data=df,
        agent_factory=lambda: MyRAGAgent(retriever=my_retriever),
        judge=judge,
        gates={"quality": 3.5},
    )

Simulation-to-scoring pipeline¶

Simulate conversations, then run detailed scoring and trajectory analysis:

from latent.flows import (
    conversation_simulation_flow,
    conversation_scoring_flow,
    conversation_trajectory_flow,
    results_to_conversations,
)

# Simulate
sim = await conversation_simulation_flow(
    agent_class=MyAgent,
    scenarios=scenarios_df,
)

# Convert to Conversation objects
convos = results_to_conversations(sim["simulation_results"])

# Score
scored = await conversation_scoring_flow(convos, judge=judge, aggregation="mean")

# Analyze trajectory patterns
traj = conversation_trajectory_flow(
    scored["scored_conversations"],
    score_key="quality",
    reference_pattern=[4.0, 4.0, 4.0],
)

Drift monitoring across runs¶

Store baseline scores and check for regression on each new evaluation:

import numpy as np
from latent.flows import judge_flow, drift_flow

# Current run
current = await judge_flow(eval_data=df, judges=judge)

# Load baseline from previous run (e.g., from MLflow or disk)
baseline_scores = {"quality": np.load("baseline_quality.npy")}
current_scores = {"quality": current["scored_data"]["quality"].values}

drift = drift_flow(
    reference_scores=baseline_scores,
    current_scores=current_scores,
    threshold=0.05,
)

if drift["drift_detected"]:
    print("Quality regression detected")
    print(drift["report_text"])

Custom deterministic scorers with `agent_eval_flow`¶

from latent.stats.models import MetricResult

def exact_match_scorer(scored_data, inference_results):
    matches = sum(
        1 for r in inference_results
        if r.get("output", "").strip() == r.get("expected", "").strip()
    )
    total = len(inference_results)
    return {
        "exact_match": MetricResult(
            name="exact_match",
            point_estimate=matches / total if total else 0.0,
            sample_size=total,
        )
    }

result = await agent_eval_flow(
    eval_data=df,
    agent_factory=lambda: MyAgent(),
    judge=judge,
    scorers=[exact_match_scorer],
)

Pre-built Evaluation Flows¶

Choosing a Flow¶

Scoring Flows¶

judge_flow¶

comparison_flow¶

Classification Flows¶

classification_flow¶

Conversation Flows¶

conversation_scoring_flow¶

conversation_sop_flow¶

conversation_trajectory_flow¶

conversation_simulation_flow¶

Agent Evaluation Flows¶

agent_eval_flow¶

agent_inference_flow¶

RAG Evaluation Flows¶

retrieval_eval_flow¶

Drift Detection¶

drift_flow¶

More Eval Flows¶

model_garden_flow¶

agent_garden_flow¶

instrumented_judge_flow¶

ner_flow¶

text_to_sql_flow¶

knowledge_coverage_flow¶

eval_report_flow¶

classify_failure_modes¶

autoresearch_agent_flow¶

guardrail_redteam¶

RAG research & optimization flows¶

Dataset helpers¶

Quality Gates¶

Statistical Analysis Integration¶

Common Patterns¶

Chaining flows¶

Simulation-to-scoring pipeline¶

Drift monitoring across runs¶

Custom deterministic scorers with agent_eval_flow¶

See Also¶

`judge_flow`¶

`comparison_flow`¶

`classification_flow`¶

`conversation_scoring_flow`¶

`conversation_sop_flow`¶

`conversation_trajectory_flow`¶

`conversation_simulation_flow`¶

`agent_eval_flow`¶

`agent_inference_flow`¶

`retrieval_eval_flow`¶

`drift_flow`¶

`model_garden_flow`¶

`agent_garden_flow`¶

`instrumented_judge_flow`¶

`ner_flow`¶

`text_to_sql_flow`¶

`knowledge_coverage_flow`¶

`eval_report_flow`¶

`classify_failure_modes`¶

`autoresearch_agent_flow`¶

`guardrail_redteam`¶

Custom deterministic scorers with `agent_eval_flow`¶