feat(fase3): run_portfolio sequential orchestrator + PortfolioResult aggregate
This commit is contained in:
parent
c7c42eeb75
commit
52f6f65b7d
4 changed files with 239 additions and 3 deletions
|
|
@ -1,7 +1,13 @@
|
|||
"""portfolio-optimiser — generic MAF framework for per-project cost-savings optimization."""
|
||||
|
||||
from portfolio_optimiser.run import RunResult, run_project
|
||||
from portfolio_optimiser.run import PortfolioResult, RunResult, run_portfolio, run_project
|
||||
|
||||
__version__ = "0.1.0"
|
||||
|
||||
__all__ = ["RunResult", "run_project", "__version__"]
|
||||
__all__ = [
|
||||
"PortfolioResult",
|
||||
"RunResult",
|
||||
"run_portfolio",
|
||||
"run_project",
|
||||
"__version__",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ durable learned verdict captured out-of-band in the VerdictStore (D7-portable).
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from collections.abc import Callable, Sequence
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
|
|
@ -68,6 +68,25 @@ class RunResult:
|
|||
debate_output: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PortfolioResult:
|
||||
"""The outcome of a sequential fan-out over N projects (SC2).
|
||||
|
||||
``runs`` is one ``RunResult`` per project in input order; ``store`` is the ONE shared
|
||||
``VerdictStore`` threaded across every run (the cross-project ExpeL learning loop).
|
||||
The remaining fields are a thin aggregate over ``runs``: ``validated_count`` /
|
||||
``rejected_count`` partition the outcomes; ``sum_claimed_saving_nok`` totals the claimed
|
||||
saving of the validated proposals only; ``sum_token_usage`` totals every run's
|
||||
provenance token usage."""
|
||||
|
||||
runs: tuple[RunResult, ...]
|
||||
store: VerdictStore
|
||||
validated_count: int
|
||||
rejected_count: int
|
||||
sum_claimed_saving_nok: float
|
||||
sum_token_usage: int
|
||||
|
||||
|
||||
# The GroupChat orchestrator emits this terminal notice as a participant output; it is NOT the
|
||||
# debate's substantive result, so the F1 extractor filters it out (verified against MAF 1.9.0).
|
||||
_ORCH_NOTICE = "reached the maximum number of rounds"
|
||||
|
|
@ -209,6 +228,63 @@ async def run_project(
|
|||
)
|
||||
|
||||
|
||||
def _aggregate(runs: tuple[RunResult, ...], store: VerdictStore) -> PortfolioResult:
|
||||
"""Thin aggregate over the per-project runs (SC2): partition validated/rejected, total the
|
||||
validated claimed saving, and total every run's provenance token usage."""
|
||||
validated = [r for r in runs if isinstance(r.outcome, ValidatedProposal)]
|
||||
rejected = [r for r in runs if isinstance(r.outcome, Rejection)]
|
||||
return PortfolioResult(
|
||||
runs=runs,
|
||||
store=store,
|
||||
validated_count=len(validated),
|
||||
rejected_count=len(rejected),
|
||||
sum_claimed_saving_nok=sum(r.outcome.proposal.claimed_saving_nok for r in validated),
|
||||
sum_token_usage=sum(r.provenance.token_usage for r in runs),
|
||||
)
|
||||
|
||||
|
||||
async def run_portfolio(
|
||||
project_ids: Sequence[str] | None = None,
|
||||
profile: Profile | str = Profile.LOCAL,
|
||||
*,
|
||||
store: VerdictStore | None = None,
|
||||
client_factory: Callable[[str], BaseChatClient] | None = None,
|
||||
max_rounds: int = 3,
|
||||
max_tokens: int = 100_000,
|
||||
top_k: int = 3,
|
||||
meter_factory: Callable[[], TokenMeter] | None = None,
|
||||
) -> PortfolioResult:
|
||||
"""Fan out over a portfolio of independent projects SEQUENTIALLY, composing ``run_project``
|
||||
as-is (every project's execution state — meter, debate, retrieval context — is built fresh
|
||||
per call, so sequential reuse is inherently isolated). ONE ``VerdictStore`` is threaded
|
||||
across every run, so a verdict on project k informs the ExpeL retrieval of project k+1 (the
|
||||
cross-project learning loop). ``project_ids`` defaults to every loaded project; an unknown id
|
||||
raises ``ValueError``. ``meter_factory`` (test seam) supplies a per-project meter — inject a
|
||||
shared meter to make the isolation guard go red (SC3)."""
|
||||
projects = {p.id: p for p in load_reference_projects()}
|
||||
ids = list(project_ids) if project_ids is not None else list(projects)
|
||||
store = store if store is not None else VerdictStore(verdicts=[])
|
||||
runs: list[RunResult] = []
|
||||
for pid in ids:
|
||||
if pid not in projects:
|
||||
raise ValueError(f"unknown project_id: {pid!r}")
|
||||
project = projects[pid]
|
||||
result = await run_project(
|
||||
pid,
|
||||
profile,
|
||||
docs_dir=project.docs_dir,
|
||||
verdict_input=project.verdict_input,
|
||||
store=store,
|
||||
client_factory=client_factory,
|
||||
max_rounds=max_rounds,
|
||||
max_tokens=max_tokens,
|
||||
top_k=top_k,
|
||||
meter=meter_factory() if meter_factory is not None else None,
|
||||
)
|
||||
runs.append(result)
|
||||
return _aggregate(tuple(runs), store)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
"""Single-command console entry: run the slice for one project against a docs folder."""
|
||||
import argparse
|
||||
|
|
|
|||
|
|
@ -93,6 +93,76 @@ def make_client_factory() -> Callable[..., Callable[[str], BaseChatClient]]:
|
|||
return _make
|
||||
|
||||
|
||||
# A generic VALID SavingsProposal reply for any project not present in a portfolio reply map:
|
||||
# affected total = 1 x 100_000 = 100_000, P90 = 0.30 x 100_000 = 30_000, claimed 20_000 <= both
|
||||
# (Pydantic affected-total invariant and the validator P90 gate) -> always validates.
|
||||
_PORTFOLIO_DEFAULT_REPLY = (
|
||||
'{"measure":"Reduce scope","affected_items":'
|
||||
'[{"code":"01.1","quantity":1,"unit_cost":100000}],"claimed_saving_nok":20000}'
|
||||
)
|
||||
|
||||
|
||||
class _ProjectAwareUsageChatClient(SyntheticUsageChatClient):
|
||||
"""A ``SyntheticUsageChatClient`` that selects its reply by scanning the incoming prompt for
|
||||
a known ``project_id`` substring (the prompt embeds ``project.id`` at run.py:162 and
|
||||
generate.py:48), falling back to a default valid proposal. This keeps ``run_portfolio``'s
|
||||
single ``client_factory`` production-shaped while letting tests vary the proposal per
|
||||
project."""
|
||||
|
||||
def __init__(
|
||||
self, replies: dict[str, str], *, default_reply: str, tokens_per_reply: int = 8
|
||||
) -> None:
|
||||
super().__init__(default_reply=default_reply, tokens_per_reply=tokens_per_reply)
|
||||
self._replies = dict(replies)
|
||||
|
||||
def _inner_get_response(
|
||||
self, *, messages: Sequence[Message], stream: bool, options: Any, **kwargs: Any
|
||||
) -> Any:
|
||||
blob = " ".join(getattr(m, "text", "") or "" for m in messages)
|
||||
reply = next((r for pid, r in self._replies.items() if pid in blob), self._default)
|
||||
self.call_count += 1
|
||||
usage = UsageDetails(total_token_count=self._tokens)
|
||||
if stream:
|
||||
|
||||
async def _agen() -> Any:
|
||||
yield ChatResponseUpdate(
|
||||
role="assistant", contents=[{"type": "text", "text": reply}]
|
||||
)
|
||||
|
||||
return self._build_response_stream(_agen())
|
||||
|
||||
async def _coro() -> ChatResponse:
|
||||
return ChatResponse(
|
||||
messages=[Message(role="assistant", contents=[reply])],
|
||||
response_id="synthetic",
|
||||
usage_details=usage,
|
||||
)
|
||||
|
||||
return _coro()
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def make_portfolio_client_factory() -> Callable[..., Callable[[str], BaseChatClient]]:
|
||||
"""Return a maker that builds a single project-aware client factory: every client it
|
||||
produces picks its reply from ``replies`` by scanning the prompt for the project id, so one
|
||||
factory serves the whole portfolio (matching ``run_portfolio``'s single-factory seam)."""
|
||||
|
||||
def _make(
|
||||
replies: dict[str, str],
|
||||
*,
|
||||
default_reply: str = _PORTFOLIO_DEFAULT_REPLY,
|
||||
tokens: int = 8,
|
||||
) -> Callable[[str], BaseChatClient]:
|
||||
def factory(role: str) -> BaseChatClient:
|
||||
return _ProjectAwareUsageChatClient(
|
||||
replies, default_reply=default_reply, tokens_per_reply=tokens
|
||||
)
|
||||
|
||||
return factory
|
||||
|
||||
return _make
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def fresh_store() -> VerdictStore:
|
||||
return VerdictStore(verdicts=[])
|
||||
|
|
|
|||
84
tests/test_portfolio.py
Normal file
84
tests/test_portfolio.py
Normal file
|
|
@ -0,0 +1,84 @@
|
|||
"""Fase 3 — sequential portfolio fan-out over N independent projects.
|
||||
|
||||
Covers the success criteria: SC2 (fan-out + aggregate sums), SC4 (shared store accumulates +
|
||||
load-bearing cross-project retrieval), SC3 (load-bearing meter-isolation detach), SC7 (both
|
||||
profiles offline + resolve_model teeth), SC1 (new project via config only + no-hardcoded-id src
|
||||
guard), SC6 (extension doc exists + references the seams).
|
||||
|
||||
Pattern: tests/test_vertical_slice_e2e.py:28 (run_project call shape). The project-aware
|
||||
synthetic client (conftest ``make_portfolio_client_factory``) selects each project's reply by
|
||||
scanning the prompt for its id, so one production-shaped factory serves the whole portfolio.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from portfolio_optimiser.run import PortfolioResult, RunResult, run_portfolio
|
||||
|
||||
# The synthetic reply IS the proposal: generate._parse_ir builds affected_items (each with its
|
||||
# own quantity/unit_cost) straight from this JSON, and the validator's P90 = 0.30 x Σ(qty·unit_cost)
|
||||
# ONLY when ``assumptions`` is empty (degenerate Monte Carlo, validator.py:108-113). All three
|
||||
# replies therefore OMIT ``assumptions`` and carry explicit magnitudes so each
|
||||
# ``claimed_saving_nok`` <= P90. Verified against validator + ir:
|
||||
# FV42-GSV-E1 Σ=1,482,500 P90=444,750 claimed 200,000 -> validates
|
||||
# RV13-RAS-TP Σ= 756,000 P90=226,800 claimed 130,000 -> validates (decoy)
|
||||
# BRU-LAKS-REHAB Σ=2,580,500 P90=774,150 claimed 210,000 -> validates
|
||||
# ``measure`` is byte-identical "Reduce scope" for FV42+BRU (measure-match is exact string
|
||||
# equality, verdicts.py:68) and "Material substitution" for the decoy, so the BRU<->FV42 pair
|
||||
# overlaps (shared code 05.2 + measure + magnitude bucket) while the decoy does not.
|
||||
REPLIES = {
|
||||
"FV42-GSV-E1": (
|
||||
'{"measure":"Reduce scope","affected_items":['
|
||||
'{"code":"05.2","quantity":4300,"unit_cost":215},'
|
||||
'{"code":"03.1","quantity":1800,"unit_cost":310}],"claimed_saving_nok":200000}'
|
||||
),
|
||||
"RV13-RAS-TP": (
|
||||
'{"measure":"Material substitution","affected_items":['
|
||||
'{"code":"88.2","quantity":180,"unit_cost":4200}],"claimed_saving_nok":130000}'
|
||||
),
|
||||
"BRU-LAKS-REHAB": (
|
||||
'{"measure":"Reduce scope","affected_items":['
|
||||
'{"code":"05.2","quantity":4300,"unit_cost":215},'
|
||||
'{"code":"07.4","quantity":2400,"unit_cost":690}],"claimed_saving_nok":210000}'
|
||||
),
|
||||
}
|
||||
|
||||
_PORTFOLIO_IDS = ["FV42-GSV-E1", "RV13-RAS-TP", "BRU-LAKS-REHAB"]
|
||||
|
||||
|
||||
async def test_a_fanout_returns_one_runresult_per_project(
|
||||
make_portfolio_client_factory, fresh_store
|
||||
) -> None:
|
||||
"""SC2: run_portfolio fans out sequentially, one RunResult per project, and aggregates the
|
||||
validated/rejected counts + claimed-saving + token sums."""
|
||||
result = await run_portfolio(
|
||||
_PORTFOLIO_IDS,
|
||||
"local",
|
||||
store=fresh_store,
|
||||
client_factory=make_portfolio_client_factory(REPLIES),
|
||||
)
|
||||
assert isinstance(result, PortfolioResult)
|
||||
assert len(result.runs) == 3
|
||||
assert all(isinstance(r, RunResult) for r in result.runs)
|
||||
# Aggregate pinned to the fixture constants above (all three validate).
|
||||
assert result.validated_count == 3
|
||||
assert result.rejected_count == 0
|
||||
assert result.sum_claimed_saving_nok == 540000
|
||||
# Explicit element-wise wiring check (not the field's own sum() definition).
|
||||
expected_tokens = (
|
||||
result.runs[0].provenance.token_usage
|
||||
+ result.runs[1].provenance.token_usage
|
||||
+ result.runs[2].provenance.token_usage
|
||||
)
|
||||
assert result.sum_token_usage == expected_tokens
|
||||
assert all(r.provenance.token_usage > 0 for r in result.runs)
|
||||
|
||||
|
||||
async def test_a2_unknown_project_id_raises(make_portfolio_client_factory, fresh_store) -> None:
|
||||
"""The unknown-id error path: an id absent from the loaded portfolio raises ValueError."""
|
||||
with pytest.raises(ValueError):
|
||||
await run_portfolio(
|
||||
["NOPE-DOES-NOT-EXIST"],
|
||||
"local",
|
||||
store=fresh_store,
|
||||
client_factory=make_portfolio_client_factory(REPLIES),
|
||||
)
|
||||
Loading…
Add table
Add a link
Reference in a new issue