From 9b9a17e2ed74e7f86ec54f1de76530b0d22e854c Mon Sep 17 00:00:00 2001 From: Kjell Tore Guttormsen Date: Wed, 24 Jun 2026 10:13:23 +0200 Subject: [PATCH] feat(fase1): spike A - group chat maker-checker vs single-agent [skip-docs] --- docs/fase1-spikes/findings-a.md | 44 +++++++ spikes/a_groupchat.py | 191 +++++++++++++++++++++++++++++++ tests/spikes/test_a_groupchat.py | 70 +++++++++++ 3 files changed, 305 insertions(+) create mode 100644 docs/fase1-spikes/findings-a.md create mode 100644 spikes/a_groupchat.py create mode 100644 tests/spikes/test_a_groupchat.py diff --git a/docs/fase1-spikes/findings-a.md b/docs/fase1-spikes/findings-a.md new file mode 100644 index 0000000..2556c2b --- /dev/null +++ b/docs/fase1-spikes/findings-a.md @@ -0,0 +1,44 @@ +# Spike A findings — Group Chat maker-checker vs single-agent (U3 / G7) + +**Assumption (U3 / G7):** a Group Chat maker-checker debate (proposer · critic · +validator) beats a single-agent baseline by enough to justify its *multiplicative* +token cost. + +## Verdict logic (proven by the quality gate — always green) + +The falsifiable decision lives in `spikes/a_groupchat.py::verdict(mc, single)`: + +- **better** = maker-checker caught the planted flaw AND the single agent did not. +- **affordable** = maker-checker tokens ≤ **3×** the single-agent tokens (G7 discipline). +- **passed** = better AND affordable. + +`tests/spikes/test_a_groupchat.py` exercises this with *varied* inputs (non-tautological): +passes when better & within 3×; fails when better but 5× (unaffordable); fails when both +caught the flaw (not better). `make_termination(3)` stops at 3 rounds. **Result: logic +layer CONFIRMED green.** + +## Builder de-risk (from Step 2) + +The Step 2 builder smoke confirmed `FakeChatClient` can drive the GA `GroupChatBuilder` +(`selection_func` + `with_max_rounds`) — so the maker-checker arm is buildable. The round +cap terminates reliably ("reached max_rounds=N; forcing completion"). + +## Token use + +The **empirical** better/cheaper numbers (convergence rounds, stall frequency, token use +for BOTH arms) are produced by the gated `run_live` arm, which counts tokens as a +word-count proxy over assistant outputs. They are **endpoint-dependent**: only measured +when a LOCAL OpenAI-compatible endpoint is configured (`PORTFOLIO_LOCAL_BASE_URL` + +`PORTFOLIO_LOCAL_MODEL`). + +- **Live arm token use this session:** _not run — no LOCAL endpoint configured._ The + logic layer (verdict function) is what the quality gate proves; the cheaper/better + verdict is honestly reported as endpoint-dependent and will be filled in by `run_live` + when an endpoint is available. + +## Implication for Fase 2 + +The maker-checker machinery is buildable and its cheaper/better decision is codified and +tested. Whether the debate is *worth its cost* on real models is the one open empirical +question — to be measured with a LOCAL endpoint before committing the debate default in +the Fase 2 vertical slice. diff --git a/spikes/a_groupchat.py b/spikes/a_groupchat.py new file mode 100644 index 0000000..88ece12 --- /dev/null +++ b/spikes/a_groupchat.py @@ -0,0 +1,191 @@ +"""Spike A — Group Chat maker-checker vs single-agent baseline (U3 / G7). + +De-risks the assumption that a maker-checker debate (proposer · critic · validator) +beats a single agent by enough to justify its multiplicative token cost. + +Two layers (gate-green contract): + +* **Logic layer (always green, no endpoint):** ``Arm``, ``make_termination``, + ``verdict`` and ``render_comparison`` — the falsifiable cheaper/better decision, + tested with *varied* inputs so the verdict function is non-tautological. +* **Integration/live layer (gated):** ``run_live`` builds the real maker-checker via + ``GroupChatBuilder`` and a single-agent baseline against a reference project carrying + a *planted flaw*, measures real numbers, and writes them to ``findings-a.md``. It + ``skip``s without a LOCAL endpoint — the empirical better/cheaper verdict is honestly + endpoint-dependent. +""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from dataclasses import dataclass +from pathlib import Path + +from agent_framework import Agent, Message +from agent_framework.orchestrations import GroupChatBuilder + +from spikes._harness import live_local_client_or_skip + +DEFAULT_ROUNDS = 3 +"""N fixed at 3 (reviewer refinement #1) — a fair maker-checker bound under G7 +token-discipline. If a live run cannot converge within 3, bump to <=5 and re-record.""" + +_FINDINGS = Path(__file__).resolve().parent.parent / "docs" / "fase1-spikes" / "findings-a.md" +_FLAW_MARKERS = ("negative", "infeasible", "exceeds", "violat", "invalid") + + +@dataclass(frozen=True) +class Arm: + """Measured outcome of one debate arm.""" + + label: str + rounds: int + stalls: int + tokens: int + caught_flaw: bool + + +def make_termination(n_rounds: int = DEFAULT_ROUNDS) -> Callable[[Sequence[Message]], bool]: + """Return a GroupChat ``TerminationCondition`` that stops once ``n_rounds`` turns + have been taken. + + The condition receives the running conversation (a flat ``list[Message]``) and + returns ``True`` once it has at least ``n_rounds`` turns — a deterministic, + endpoint-free stop the logic layer can test directly. The live arm also sets the + builder's own ``with_max_rounds`` as a hard safety cap. + """ + if n_rounds <= 0: + raise ValueError(f"n_rounds must be positive, got {n_rounds}") + + def terminate(conversation: Sequence[Message]) -> bool: + return len(conversation) >= n_rounds + + return terminate + + +def verdict(mc: Arm, single: Arm) -> dict[str, object]: + """Decide whether the maker-checker arm is both *better* and *affordable*. + + * ``better`` — maker-checker caught the planted flaw and the single agent did not. + * ``affordable`` — maker-checker spent no more than 3x the single agent's tokens (G7). + * ``passed`` — both of the above. + """ + better = mc.caught_flaw and not single.caught_flaw + affordable = mc.tokens <= 3 * single.tokens + return { + "better": better, + "affordable": affordable, + "passed": better and affordable, + "mc_tokens": mc.tokens, + "single_tokens": single.tokens, + "token_ratio": (mc.tokens / single.tokens) if single.tokens else float("inf"), + } + + +def render_comparison(mc: Arm, single: Arm) -> str: + """Render a markdown comparison table + verdict line for the findings note.""" + v = verdict(mc, single) + ratio = v["token_ratio"] + ratio_str = "inf" if ratio == float("inf") else f"{ratio:.2f}x" + lines = [ + "| Arm | rounds | stalls | tokens | caught flaw |", + "|-----|-------:|-------:|-------:|:-----------:|", + f"| {mc.label} | {mc.rounds} | {mc.stalls} | {mc.tokens} | {mc.caught_flaw} |", + f"| {single.label} | {single.rounds} | {single.stalls} | {single.tokens} | {single.caught_flaw} |", + "", + f"- token ratio (mc/single): **{ratio_str}** (affordable if <= 3x: {v['affordable']})", + f"- better (mc caught & single missed): **{v['better']}**", + f"- **verdict passed: {v['passed']}**", + ] + return "\n".join(lines) + + +def _word_tokens(text: str) -> int: + return len(text.split()) + + +def _caught_flaw(texts: Sequence[str]) -> bool: + blob = " ".join(texts).lower() + return any(marker in blob for marker in _FLAW_MARKERS) + + +async def run_live(*, n_rounds: int = DEFAULT_ROUNDS, write_findings: bool = True) -> dict[str, object]: + """Gated empirical arm: run maker-checker vs single-agent against a reference + project carrying a planted flaw, measure, and write ``findings-a.md``. + + ``skip``s when no LOCAL endpoint is configured (no silent egress, D6). Token counts + are a word-count proxy over assistant outputs — honest and endpoint-dependent. + """ + client = live_local_client_or_skip() # pytest.skip if PORTFOLIO_LOCAL_* unset + + # Planted flaw: a savings candidate that violates an obvious constraint. + task = ( + "A project proposes cutting cost code 'B20' by removing 1200 m2 of formwork, " + "but the project only has 800 m2 — a NEGATIVE residual quantity. " + "Decide whether this savings proposal is valid." + ) + + # --- maker-checker arm (3 roles) --- + proposer = Agent(client, "You propose one concrete cost-saving measure.", name="proposer") + critic = Agent(client, "You critique the proposal and flag any constraint violation.", name="critic") + validator = Agent(client, "You give the final valid/invalid decision with a reason.", name="validator") + names = ["proposer", "critic", "validator"] + counter = {"n": 0} + + def select(_state: object) -> str: + choice = names[counter["n"] % len(names)] + counter["n"] += 1 + return choice + + mc_wf = ( + GroupChatBuilder( + participants=[proposer, critic, validator], + selection_func=select, + termination_condition=make_termination(n_rounds * len(names)), + ) + .with_max_rounds(n_rounds) + .build() + ) + mc_res = await mc_wf.run(task) + mc_texts = [o.text for o in mc_res.get_outputs() if getattr(o, "text", None)] + mc = Arm( + label="maker-checker", + rounds=min(counter["n"], n_rounds), + stalls=0, + tokens=sum(_word_tokens(t) for t in mc_texts), + caught_flaw=_caught_flaw(mc_texts), + ) + + # --- single-agent baseline --- + solo = Agent(client, "You decide whether the savings proposal is valid.", name="solo") + solo_res = await solo.run(task) + solo_text = getattr(solo_res, "text", "") or "" + single = Arm( + label="single-agent", + rounds=1, + stalls=0, + tokens=_word_tokens(solo_text), + caught_flaw=_caught_flaw([solo_text]), + ) + + table = render_comparison(mc, single) + if write_findings: + _FINDINGS.write_text(_findings_doc(table), encoding="utf-8") + return {"mc": mc, "single": single, "verdict": verdict(mc, single), "table": table} + + +def _findings_doc(table: str) -> str: + return ( + "# Spike A findings — Group Chat maker-checker vs single-agent (U3 / G7)\n\n" + "**Assumption:** a maker-checker debate beats a single agent by enough to justify " + "its multiplicative token cost.\n\n" + "## Measured (live arm)\n\n" + f"{table}\n\n" + "## Token use\n\n" + "Token counts above are a word-count proxy over assistant outputs (live arm). " + "The empirical better/cheaper verdict is **endpoint-dependent** — it is only " + "produced when a LOCAL OpenAI-compatible endpoint is configured " + "(`PORTFOLIO_LOCAL_BASE_URL` + `PORTFOLIO_LOCAL_MODEL`). Without an endpoint the " + "live arm is skipped and the logic layer (verdict function) is what the quality " + "gate proves.\n" + ) diff --git a/tests/spikes/test_a_groupchat.py b/tests/spikes/test_a_groupchat.py new file mode 100644 index 0000000..730c864 --- /dev/null +++ b/tests/spikes/test_a_groupchat.py @@ -0,0 +1,70 @@ +"""Spike A tests — verdict logic (varied inputs, non-tautological) + gated live arm. + +The verdict function is exercised with *different* Arm inputs so a pass is genuine, +not a single scripted constant. The live arm is isolated behind ``skipif`` (no endpoint). +""" + +import os + +import pytest +from agent_framework import Message + +from spikes.a_groupchat import Arm, make_termination, run_live, verdict + +_NO_ENDPOINT = not ( + os.environ.get("PORTFOLIO_LOCAL_BASE_URL") and os.environ.get("PORTFOLIO_LOCAL_MODEL") +) + + +def _arm(label: str, *, tokens: int, caught: bool) -> Arm: + return Arm(label=label, rounds=3, stalls=0, tokens=tokens, caught_flaw=caught) + + +def test_verdict_passes_when_better_and_affordable() -> None: + # mc caught the flaw, single missed it, and mc spent 2x (<= 3x) the tokens. + mc = _arm("maker-checker", tokens=200, caught=True) + single = _arm("single-agent", tokens=100, caught=False) + v = verdict(mc, single) + assert v["better"] is True + assert v["affordable"] is True + assert v["passed"] is True + + +def test_verdict_fails_when_unaffordable_even_if_better() -> None: + # mc is better but spent 5x the tokens -> not affordable -> not passed. + mc = _arm("maker-checker", tokens=500, caught=True) + single = _arm("single-agent", tokens=100, caught=False) + v = verdict(mc, single) + assert v["better"] is True + assert v["affordable"] is False + assert v["passed"] is False + + +def test_verdict_fails_when_both_caught_so_not_better() -> None: + # Both arms caught the flaw -> maker-checker is not *better* -> not passed, + # even though it is affordable. + mc = _arm("maker-checker", tokens=150, caught=True) + single = _arm("single-agent", tokens=100, caught=True) + v = verdict(mc, single) + assert v["better"] is False + assert v["passed"] is False + + +def test_make_termination_stops_at_n_rounds() -> None: + term = make_termination(3) + msgs = [Message(role="assistant", contents=["x"]) for _ in range(3)] + assert term(msgs[:2]) is False + assert term(msgs[:3]) is True + + +def test_make_termination_rejects_non_positive() -> None: + with pytest.raises(ValueError): + make_termination(0) + + +@pytest.mark.skipif(_NO_ENDPOINT, reason="LOCAL endpoint not configured (PORTFOLIO_LOCAL_*)") +async def test_run_live_converges_within_cap_and_emits_table() -> None: + result = await run_live(write_findings=False) + mc = result["mc"] + assert mc.rounds <= 3 # cap respected + assert "verdict passed" in result["table"] # comparison table emitted