From f29a14a6e3dd61f20609919b09ca890cfccc6650 Mon Sep 17 00:00:00 2001 From: Nikolas Jack Altran Date: Thu, 2 Jul 2026 15:37:17 +0200 Subject: [PATCH] feat(pipeline): finalize on best iteration, not last Accuracy can regress across retune passes (e.g. 0.39 at iter 3 but 0.37 at the forced proceed), yet each pass overwrote predictions/eval/classifier, so explain + finalize ran on the regressed artifacts. - evaluate node snapshots PREDS/EVAL/CODE to outputs/best/ on each new best - new select_best node on the proceed branch restores that snapshot and redraws sample_for_explanation.csv when the last iteration wasn't the best - extract Manager's sampling into shared write_sample() so the sample format keeps a single definition Gate semantics unchanged: the Manager still decides from the last iteration's report; restore happens after the verdict. --- agents/jack_manager.py | 21 ++++++--- agents/pipeline_graph.py | 57 ++++++++++++++++++++++--- tests/test_pipeline_graph.py | 82 ++++++++++++++++++++++++++++++++++++ 3 files changed, 147 insertions(+), 13 deletions(-) diff --git a/agents/jack_manager.py b/agents/jack_manager.py index 50805a9..eb392f1 100644 --- a/agents/jack_manager.py +++ b/agents/jack_manager.py @@ -235,21 +235,30 @@ def write_retune(state: ManagerState) -> dict: return {"decision_log": [f"iteration {state['iteration']}: wrote retune_request.json"]} -def proceed(state: ManagerState) -> dict: - """sample_for_explanation.csv (Handoff 4) — drawn from predictions_test.csv. - Terminal: hands off to Freddi and waits for explanations.csv.""" +def write_sample(predictions_path: str, sample_size: int = 300) -> int: + """Write sample_for_explanation.csv (Handoff 4) from a predictions file. + Shared by the proceed node and the pipeline's best-iteration restore, so the + sample format has exactly one definition.""" import pandas as pd - preds = pd.read_csv(state.get("predictions_path", "mock_data/predictions_test.csv")) + preds = pd.read_csv(predictions_path) sample = (preds[["article_id", "article_title", "predicted_label", "label", "confidence", "prob_up", "prob_down", "prob_neutral"]] .rename(columns={"label": "actual_label"})) - n = min(len(sample), state.get("sample_size", 300)) + n = min(len(sample), sample_size) if n < len(sample): # only subsample when needed sample = sample.sample(n=n, random_state=42) # representative + reproducible - _write_decision(state) os.makedirs(OUTPUT_DIR, exist_ok=True) sample.to_csv(os.path.join(OUTPUT_DIR, "sample_for_explanation.csv"), index=False) + return n + + +def proceed(state: ManagerState) -> dict: + """sample_for_explanation.csv (Handoff 4) — drawn from predictions_test.csv. + Terminal: hands off to Freddi and waits for explanations.csv.""" + _write_decision(state) + n = write_sample(state.get("predictions_path", "mock_data/predictions_test.csv"), + state.get("sample_size", 300)) return {"decision_log": [f"iteration {state['iteration']}: wrote sample_for_explanation.csv ({n} rows)"]} diff --git a/agents/pipeline_graph.py b/agents/pipeline_graph.py index 0cb7604..bbe8101 100644 --- a/agents/pipeline_graph.py +++ b/agents/pipeline_graph.py @@ -20,7 +20,12 @@ ## Graph shape START → process → classify → evaluate → gate ─(retune)→ classify [CYCLE] - └(proceed)→ explain → finalize → END + └(proceed)→ select_best → explain → finalize → END + +`evaluate` snapshots each new-best iteration's artifacts into `outputs/best/`; +`select_best` restores that snapshot (and redraws the explanation sample) when the +loop's LAST iteration wasn't its best — accuracy can regress across retunes, and +without this the pipeline would finalize on the regressed predictions. `gate` is the Manager: calling it writes either `retune_request.json` (retune) or `sample_for_explanation.csv` (proceed). Its own checkpointer carries the iteration @@ -36,7 +41,9 @@ from __future__ import annotations +import json import os +import shutil from dataclasses import dataclass from typing_extensions import TypedDict @@ -46,13 +53,13 @@ from agents.nadi_classifier import ClassifierAgent from agents.sabina_evaluator import EvaluatorAgent from agents.freddi_explanation import ExplanationAgent - from agents.jack_manager import ManagerAgent, OUTPUT_DIR as OUT + from agents.jack_manager import ManagerAgent, write_sample, OUTPUT_DIR as OUT except ModuleNotFoundError: # running as a bare script with agents/ on sys.path from aurora_processing import ProcessingAgent from nadi_classifier import ClassifierAgent from sabina_evaluator import EvaluatorAgent from freddi_explanation import ExplanationAgent - from jack_manager import ManagerAgent, OUTPUT_DIR as OUT + from jack_manager import ManagerAgent, write_sample, OUTPUT_DIR as OUT # Contract files exchanged between the nodes. All live under the Manager's # OUTPUT_DIR except processed_data.csv, whose path Aurora returns at runtime. @@ -63,6 +70,10 @@ RETUNE = os.path.join(OUT, "retune_request.json") EXPL = os.path.join(OUT, "explanations.csv") +# Snapshot dir for the best iteration's artifacts (internal to the loop, not a +# contract handoff). Holds copies of CODE/PREDS/EVAL from the highest-accuracy pass. +BEST_DIR = os.path.join(OUT, "best") + # A retune cycle is 3 nodes (classify → evaluate → gate); with the process/explain/ # finalize tail, ~5 iterations stays well under this. LangGraph aborts a runaway # cycle at recursion_limit, so we set headroom rather than rely on the default 25. @@ -78,6 +89,8 @@ class PipelineState(TypedDict, total=False): retune_request_path: str | None # None on the first pass; RETUNE on cycle passes final_action: str # the gate's verdict: "retune" | "proceed" iteration: int # Manager's loop counter (for reporting) + last_accuracy: float # accuracy of the most recent evaluate pass + best_accuracy: float # best accuracy seen across cycle passes @dataclass @@ -136,9 +149,19 @@ def classify(state: PipelineState) -> dict: return {} def evaluate(state: PipelineState) -> dict: - """Sabina: score the predictions and write evaluation_report.json.""" + """Sabina: score the predictions and write evaluation_report.json. Also + snapshots this pass's artifacts into BEST_DIR whenever accuracy sets a new + best, so `select_best` can restore them if later retunes regress.""" agents.sabina.run(predictions=PREDS, classifier_code=CODE) - return {} + with open(EVAL, encoding="utf-8") as f: + acc = json.load(f)["accuracy"] + out = {"last_accuracy": acc} + if acc > state.get("best_accuracy", -1.0): + os.makedirs(BEST_DIR, exist_ok=True) + for path in (PREDS, EVAL, CODE): + shutil.copy2(path, os.path.join(BEST_DIR, os.path.basename(path))) + out["best_accuracy"] = acc + return out def gate(state: PipelineState) -> dict: """Manager gate. Invoking it applies the accuracy gate (with convergence + @@ -152,6 +175,24 @@ def gate(state: PipelineState) -> dict: out["retune_request_path"] = RETUNE # fed back into `classify` on the cycle return out + def select_best(state: PipelineState) -> dict: + """The gate proceeded with the LAST iteration's artifacts, which are not + necessarily the best ones (accuracy can regress across retunes). If an + earlier pass scored higher, restore its snapshot over the canonical paths + and redraw the explanation sample from the restored predictions, so + explain/finalize run on the best iteration. Gate semantics are untouched: + the Manager already made its verdict from the last iteration's report.""" + best, last = state.get("best_accuracy", -1.0), state.get("last_accuracy", -1.0) + if best <= last: + return {} + for path in (PREDS, EVAL, CODE): + shutil.copy2(os.path.join(BEST_DIR, os.path.basename(path)), path) + sample_size = getattr(agents.manager, "_defaults", {}).get("sample_size", 300) + write_sample(PREDS, sample_size) + print(f"[pipeline] last iteration regressed ({last:.2f}) — restored best " + f"iteration's artifacts ({best:.2f}) for explanation + finals") + return {} + def explain(state: PipelineState) -> dict: """Freddi: justify each sampled prediction into explanations.csv.""" agents.freddi.run(sample_for_explanation=SAMPLE, output=EXPL) @@ -170,13 +211,15 @@ def route(state: PipelineState) -> str: b = StateGraph(PipelineState) for name, fn in [("process", process), ("classify", classify), ("evaluate", evaluate), - ("gate", gate), ("explain", explain), ("finalize", finalize)]: + ("gate", gate), ("select_best", select_best), ("explain", explain), + ("finalize", finalize)]: b.add_node(name, fn) b.add_edge(START, "process") b.add_edge("process", "classify") b.add_edge("classify", "evaluate") b.add_edge("evaluate", "gate") - b.add_conditional_edges("gate", route, {"retune": "classify", "proceed": "explain"}) + b.add_conditional_edges("gate", route, {"retune": "classify", "proceed": "select_best"}) + b.add_edge("select_best", "explain") b.add_edge("explain", "finalize") b.add_edge("finalize", END) return b.compile(checkpointer=checkpointer) diff --git a/tests/test_pipeline_graph.py b/tests/test_pipeline_graph.py index 65ee882..fa3a368 100644 --- a/tests/test_pipeline_graph.py +++ b/tests/test_pipeline_graph.py @@ -66,6 +66,88 @@ def run(self, *, predictions, classifier_code): return {"output_path": "outputs/evaluation_report.json"} +class MarkedNadi: + """FakeNadi variant that stamps every predictions file with the pass number, + so the test can tell WHICH iteration's artifacts survived to the end.""" + def __init__(self): + self.calls = 0 + + def run(self, *, processed_data, classifier_code, predictions, retune_request=None): + self.calls += 1 + Path(predictions).parent.mkdir(parents=True, exist_ok=True) + shutil.copy(MOCK_PRED, predictions) + import pandas as pd + df = pd.read_csv(predictions) + df["fake_pass"] = self.calls + df.to_csv(predictions, index=False) + Path(classifier_code).write_text(f"THRESHOLD = 0.5 # pass {self.calls}\n") + return {} + + +class PeakSabina: + """Accuracy peaks on pass 2 then regresses, so the best iteration is NOT the + last one the gate proceeds with — exactly the case select_best must fix.""" + ACCS = [0.20, 0.39, 0.30, 0.30, 0.30, 0.30, 0.30, 0.30] + + def __init__(self): + self.calls = 0 + + def run(self, *, predictions, classifier_code): + acc = self.ACCS[self.calls] + self.calls += 1 + report = { + "accuracy": acc, + "below_threshold": True, + "class_accuracy": {"up": 0.3, "down": 0.2, "neutral": 0.5}, + "misclassified_ids": [], + "proposal": {"recommended_action": "retune", "reason": "low", + "focus_labels": ["down"], + "suggested_params": {"threshold": 0.5, "max_length": 128}, + "code_notes": ""}, + } + os.makedirs("outputs", exist_ok=True) + Path("outputs/evaluation_report.json").write_text(json.dumps(report)) + return {"output_path": "outputs/evaluation_report.json"} + + +@needs_langgraph +def test_best_iteration_restored_when_accuracy_regresses(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / "outputs").mkdir() + + import pandas as pd + import agents.pipeline_graph as pg + from agents.jack_manager import ManagerAgent + from agents.freddi_explanation import ExplanationAgent + from langgraph.checkpoint.memory import MemorySaver + + nadi, sabina = MarkedNadi(), PeakSabina() + agents = pg.Agents( + aurora=FakeAurora(), nadi=nadi, sabina=sabina, + manager=ManagerAgent(predictions_path=pg.PREDS, target_accuracy=0.60, + max_iterations=6, patience=2, min_delta=0.01), + freddi=ExplanationAgent(use_ollama=False, output_path=pg.EXPL), + ) + graph = pg.build_pipeline(agents, checkpointer=MemorySaver()) + final = graph.invoke({"retune_request_path": None}, + {"configurable": {"thread_id": "t"}, "recursion_limit": pg.RECURSION_LIMIT}) + + assert final["final_action"] == "proceed" + assert nadi.calls >= 3, "loop must run past the accuracy peak for this test to bite" + + # The artifacts that survive are the BEST pass's (accuracy 0.39 = pass 2), + # not the last pass's — the whole point of select_best. + preds = pd.read_csv(tmp_path / "outputs" / "predictions_test.csv") + assert preds["fake_pass"].iloc[0] == 2, "final predictions should come from the best pass" + report = json.loads((tmp_path / "outputs" / "evaluation_report.json").read_text()) + assert report["accuracy"] == 0.39 + final_report = json.loads((tmp_path / "outputs" / "final_report.json").read_text()) + assert final_report["final_accuracy"] == 0.39 + + # The explanation sample was redrawn from the restored predictions. + assert (tmp_path / "outputs" / "sample_for_explanation.csv").exists() + + @needs_langgraph def test_graph_cycles_then_finalizes(tmp_path, monkeypatch): monkeypatch.chdir(tmp_path)