ctan/pf1e-simulator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(phase2): llm report writer (P5.3)

Browse Source 
- New module src/pf1e_simulator/report.py with ReportContext, build_report_context,
  LLMReportPolicy, ReportBudget, format_llm_report, build_report_prompt
- Prompt pack fixture: tests/fixtures/prompt_pack_fs/report.md (FR narrator)
- CLI integration: --llm-report + --output flags, mutually exclusive with --transcript
- Budget enforcement: ReportBudget integrates with CostBudget, allows max_retries+1 calls
- Tests: test_report_context, test_report_prompt_pack, test_llm_report_policy,
  test_format_llm_report, test_cli_llm_report, test_report_budget (27 new tests)
This commit is contained in:
Thien An
2026-09-26 10:40:58 +02:00
parent c454a67b84
commit 98985ee6fc
9 changed files with 1024 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pf1e_simulator/cli.py
+119 -12
View File
@@ -19,11 +19,24 @@ if TYPE_CHECKING:
from pf1e_simulator.runner import EncounterReport, EncounterSpec
from pf1e_simulator.config import ENV_PROMPT_PACK_DIR, env_path
from pf1e_simulator.loaders import CombatantLoadError, SheetLoadError, load_combatant
from pf1e_simulator.map import MapValidationError, load_map
from pf1e_simulator.runner import EncounterSpec, Side, run_encounter
from pf1e_simulator.prompt import PromptPack
from pf1e_simulator.prompt_builder import PromptBuilder
from pf1e_simulator.report import (
LLMReportPolicy,
build_report_context,
format_llm_report,
)
from pf1e_simulator.runner import EncounterSpec, RunOutcome, Side, run_encounter
from pf1e_simulator.spells import SPELLS_DIR, load_spell_registry
try:
from pf1e_simulator.provider import OpenAILLMClient
except ImportError: # pragma: no cover - optional dependency
OpenAILLMClient = None # type: ignore[assignment,misc]
class CliError(Exception):
"""User-facing CLI error (bad arguments or unusable files)."""
@@ -129,7 +142,7 @@ def _build_spec(args: argparse.Namespace) -> EncounterSpec:
raise CliError(msg)
unknown = sorted(provided_names - set(map_spec.deployment))
if unknown:
msg = f"--side names not in map deployment: {', '.join(unknown)}"
msg = f"--side names not in map deployment: {', '.join(sorted(unknown))}"
raise CliError(msg)
sides: list[Side] = []
for raw in provided:
@@ -172,6 +185,56 @@ def format_report(report: EncounterReport, spec: EncounterSpec) -> str:
return "\n".join(lines)
def _run_llm_report(
spec: EncounterSpec,
args: argparse.Namespace,
registry: dict[str, object] | None,
) -> str:
"""Generate the LLM-powered report."""
if OpenAILLMClient is None:
print("pf1e-sim: LLM provider not available", file=sys.stderr)
raise SystemExit(2)
pack_dir = env_path(ENV_PROMPT_PACK_DIR)
if pack_dir is None:
print(f"pf1e-sim: {ENV_PROMPT_PACK_DIR} environment variable not set", file=sys.stderr)
raise SystemExit(2)
pack = PromptPack.load(pack_dir)
prompt_version = pack.prompt_version("report")
builder = PromptBuilder(pack)
client = OpenAILLMClient(
prompt_version=prompt_version,
model=args.llm_model,
api_key=args.llm_api_key,
base_url=args.llm_base_url,
)
report: EncounterReport
outcomes: tuple[RunOutcome, ...]
report, outcomes = run_encounter( # type: ignore[misc]
spec,
runs=args.runs,
seed=args.seed,
spell_registry=registry, # type: ignore[arg-type]
capture_first_transcript=args.transcript,
monsters_die_at_zero=args.simple_monster_death,
collect_runs=True,
)
ctx = build_report_context(
encounter=report, # type: ignore[arg-type]
spec_name=spec.map.name,
outcomes=outcomes, # type: ignore[arg-type]
party="players",
)
policy = LLMReportPolicy(client=client, builder=builder, prompt_name="report")
narrative = policy(ctx)
return format_llm_report(ctx, narrative)
def main(argv: list[str] | None = None) -> int:
"""CLI entry point; returns the process exit code."""
parser = argparse.ArgumentParser(
@@ -207,9 +270,39 @@ def main(argv: list[str] | None = None) -> int:
"(player characters are never affected)"
),
)
parser.add_argument(
"--llm-report",
action="store_true",
help="generate an LLM-powered French narrative report (requires PF1E_LLM_API_KEY)",
)
parser.add_argument(
"--output",
type=Path,
help="write the full report (factual + LLM narrative) to this file",
)
parser.add_argument(
"--llm-model",
type=str,
help="override the LLM model (default: PF1E_LLM_MODEL env or gpt-4o-mini)",
)
parser.add_argument(
"--llm-base-url",
type=str,
help="override the LLM base URL (default: PF1E_LLM_BASE_URL env or https://api.openai.com/v1)",
)
parser.add_argument(
"--llm-api-key",
type=str,
help="override the LLM API key (default: PF1E_LLM_API_KEY env)",
)
args = parser.parse_args(argv)
if not args.side:
parser.error("at least one --side is required")
if args.llm_report and args.transcript:
parser.error("--llm-report and --transcript are mutually exclusive")
if args.output and not args.llm_report:
parser.error("--output requires --llm-report")
try:
spec = _build_spec(args)
except CliError as exc:
@@ -218,22 +311,36 @@ def main(argv: list[str] | None = None) -> int:
except (MapValidationError, CombatantLoadError, SheetLoadError, OSError) as exc:
print(f"pf1e-sim: {exc}", file=sys.stderr)
return 2
try:
registry = load_spell_registry(SPELLS_DIR)
report = run_encounter(
spec,
runs=args.runs,
seed=args.seed,
spell_registry=registry,
capture_first_transcript=args.transcript,
monsters_die_at_zero=args.simple_monster_death,
)
report = None
if args.llm_report:
full_report = _run_llm_report(spec, args, registry) # type: ignore[arg-type]
else:
report = run_encounter(
spec,
runs=args.runs,
seed=args.seed,
spell_registry=registry,
capture_first_transcript=args.transcript,
monsters_die_at_zero=args.simple_monster_death,
)
full_report = format_report(report, spec)
except ValueError as exc:
print(f"pf1e-sim: {exc}", file=sys.stderr)
return 2
if args.transcript and report.transcript is not None:
if args.transcript and report is not None and report.transcript is not None:
print("=== Déroulé du premier combat ===")
print(render_transcript_table(report.transcript))
print()
print(format_report(report, spec))
print(full_report)
if args.output:
args.output.write_text(full_report, encoding="utf-8")
return 0
src/pf1e_simulator/report.py
+297
View File
@@ -0,0 +1,297 @@
"""LLM report writer (Phase 3) — aggregator + French narrative generator.
This module provides the data structures and pure functions to build a
report context from Monte Carlo outcomes and LLM observability counters,
then format the final French report combining the factual balance report
with an LLM-generated narrative section.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
from pf1e_simulator.client import BudgetExceededError, CostBudget
from pf1e_simulator.objectives import danger_metrics
if TYPE_CHECKING:
from collections.abc import Sequence
from pf1e_simulator.client import ClientStats, LLMClient
from pf1e_simulator.nl_policy import NLPolicyStats
from pf1e_simulator.objectives import DangerMetrics
from pf1e_simulator.prompt_builder import PromptBuilder
from pf1e_simulator.robustness import (
DangerCell,
MatchupBenchmark,
SensitivityEntry,
)
from pf1e_simulator.runner import EncounterReport, RunOutcome
@dataclass(frozen=True)
class ReportContext:
"""Complete context for the LLM report writer.
Fields marked as optional are filled when the corresponding analysis
was executed (e.g. `collect_runs=True` for DangerMetrics, robustness
calls for MatchupBenchmark/SensitivityEntry/DangerCell).
"""
encounter: EncounterReport
spec_name: str
danger: DangerMetrics | None = None
nl_stats: NLPolicyStats | None = None
client_stats: ClientStats | None = None
matchup: MatchupBenchmark | None = None
danger_map: tuple[DangerCell, ...] | None = None
frontier_value: float | None = None
sensitivity: tuple[SensitivityEntry, ...] | None = None
def build_report_context( # noqa: PLR0913 - intentional builder with many optional params
*,
encounter: EncounterReport,
spec_name: str,
outcomes: Sequence[RunOutcome] | None = None,
party: str = "players",
nl_stats: NLPolicyStats | None = None,
client_stats: ClientStats | None = None,
matchup: MatchupBenchmark | None = None,
danger_map: tuple[DangerCell, ...] | None = None,
frontier_value: float | None = None,
sensitivity: tuple[SensitivityEntry, ...] | None = None,
) -> ReportContext:
"""Build a `ReportContext` from raw Monte Carlo outputs."""
danger: DangerMetrics | None = None
if outcomes is not None:
danger = danger_metrics(outcomes, party)
return ReportContext(
encounter=encounter,
spec_name=spec_name,
danger=danger,
nl_stats=nl_stats,
client_stats=client_stats,
matchup=matchup,
danger_map=danger_map,
frontier_value=frontier_value,
sensitivity=sensitivity,
)
def _render_encounter(ctx: ReportContext) -> str:
"""Render the EncounterReport in the same format as cli.format_report."""
r = ctx.encounter
lines = ["=== Rapport d'équilibrage ===", f"Carte : {ctx.spec_name}"]
lines.extend(f"{side_name} : {r.wins.get(side_name, 0)} combats" for side_name in r.wins)
lines.append(f"{r.runs} combats simulés (seed {r.seed}, plafond ? rounds)")
lines.append("")
for side_name in r.wins:
wins = r.wins.get(side_name, 0)
rate = r.win_rate(side_name)
low, high = r.win_rate_band(side_name)
lines.append(
f"Victoires {side_name} : {rate:.1%} ({wins}) — "
f"bande 3σ : [{low:.1%}, {high:.1%}]"
)
lines.append(f"Nuls : {r.draws / r.runs:.1%} ({r.draws})")
lines.append(f"Rounds moyens : {r.avg_rounds:.1f}")
lines.append("")
lines.append("Attrition moyenne par combatant :")
for cid, attr in r.attrition.items():
lines.append(
f" {cid} : {attr.hits:.2f} touches · {attr.crits:.2f} critiques · "
f"{attr.damage_dealt:.2f} dégâts infligés · {attr.damage_taken:.2f} subis"
f" · {attr.healing_done:.2f} soins prodigués"
)
return "\n".join(lines)
def _render_danger(danger: DangerMetrics) -> str:
"""Render DangerMetrics section."""
lines = ["=== Métriques de danger ==="]
lines.append(f"runs: {danger.runs}")
lines.append(f"p_survives: {danger.p_survives:.2f}")
lines.append(f"p_tpk: {danger.p_tpk:.2f}")
lines.append(f"p_any_death: {danger.p_any_death:.2f}")
lines.append(f"expected_hp_lost: {danger.expected_hp_lost:.1f}")
lines.append(f"expected_rounds: {danger.expected_rounds:.1f}")
lines.append(f"worst_decile_party_hp: {danger.worst_decile_party_hp:.1f}")
return "\n".join(lines)
def _render_nl_stats(stats: NLPolicyStats) -> str:
lines = ["=== Observabilité LLM ==="]
lines.append(f"calls: {stats.calls}")
lines.append(f"repairs_attempted: {stats.repairs_attempted}")
lines.append(f"repairs_succeeded: {stats.repairs_succeeded}")
lines.append(f"fallbacks: {stats.fallbacks}")
return "\n".join(lines)
def _render_client_stats(stats: ClientStats) -> str:
lines = ["=== Coût LLM ==="]
lines.append(f"prompt_tokens: {stats.prompt_tokens}")
lines.append(f"completion_tokens: {stats.completion_tokens}")
lines.append(f"cost_usd: {stats.cost_usd:.3f}")
lines.append(f"cache_hits: {stats.cache_hits}")
lines.append(f"cache_misses: {stats.cache_misses}")
return "\n".join(lines)
def _render_matchup(matchup: MatchupBenchmark) -> str:
lines = ["=== Robustesse (matchup) ==="]
lines.append(f"party: {matchup.party}")
lines.append(f"adversary: {matchup.adversary}")
lines.append(f"runs: {matchup.runs}")
lines.append(f"seed: {matchup.seed}")
lines.append(f"party_danger.p_survives: {matchup.party_danger.p_survives:.2f}")
lines.append(f"party_danger.p_tpk: {matchup.party_danger.p_tpk:.2f}")
lines.append(f"adversary_danger.p_survives: {matchup.adversary_danger.p_survives:.2f}")
lines.append(f"adversary_danger.p_tpk: {matchup.adversary_danger.p_tpk:.2f}")
return "\n".join(lines)
def build_report_prompt(ctx: ReportContext, template: str) -> str:
"""Build the complete LLM prompt for the report writer."""
sections: list[str] = [template, ""]
sections.append(_render_encounter(ctx))
sections.append("")
if ctx.danger is not None:
sections.append(_render_danger(ctx.danger))
sections.append("")
if ctx.matchup is not None:
sections.append(_render_matchup(ctx.matchup))
sections.append("")
if ctx.nl_stats is not None:
sections.append(_render_nl_stats(ctx.nl_stats))
sections.append("")
if ctx.client_stats is not None:
sections.append(_render_client_stats(ctx.client_stats))
sections.append("")
if sections[-1] == "":
sections.pop()
return "\n".join(sections)
@dataclass(frozen=True)
class ReportBudget:
"""Budget limits for a single report generation (integrates with CostBudget).
Mirrors the ceilings from the architecture doc (section 13):
small caps for showcase/vitrine mode — max 1 call per report.
"""
max_calls: int = 1
max_total_tokens: int | None = None
max_cost_usd: float | None = None
def to_cost_budget(self) -> CostBudget:
"""Convert to the client's CostBudget type."""
return CostBudget(
max_calls=self.max_calls,
max_total_tokens=self.max_total_tokens,
max_cost_usd=self.max_cost_usd,
)
def for_retries(self, max_retries: int) -> ReportBudget:
"""Return a budget allowing for retries (max_calls = max_retries + 1)."""
if self.max_calls == 1:
return ReportBudget(
max_calls=max_retries + 1,
max_total_tokens=self.max_total_tokens,
max_cost_usd=self.max_cost_usd,
)
return self
@dataclass
class ReportStats:
"""Mutable counters for the LLM report writer."""
calls: int = 0
retries: int = 0
fallbacks: int = 0
@dataclass(frozen=True)
class LLMReportPolicy:
"""Generate a French narrative report via LLM completion.
Distinct from NLPolicy: produces free-text narrative (not JSON actions),
uses simple retry loop (no repair trail), and tracks separate stats.
"""
client: LLMClient
builder: PromptBuilder
budget: ReportBudget = field(default_factory=ReportBudget)
prompt_name: str = "report"
max_retries: int = 1
stats: ReportStats = field(default_factory=ReportStats)
def __post_init__(self) -> None:
if self.max_retries < 0:
msg = f"max_retries must be >= 0, got {self.max_retries}"
raise ValueError(msg)
expected = self.builder.prompt_version(self.prompt_name)
if self.client.prompt_version != expected:
msg = (
f"client prompt_version {self.client.prompt_version!r} does not "
f"match pack {expected!r} for {self.prompt_name!r}"
)
raise ValueError(msg)
# Apply the report budget (adjusted for retries) to the client
if self.client._budget is None: # noqa: SLF001 - private access required for budget injection # pyright: ignore[reportPrivateUsage]
adjusted_budget = self.budget.for_retries(self.max_retries)
self.client._budget = adjusted_budget.to_cost_budget() # noqa: SLF001 - private access required for budget injection # pyright: ignore[reportPrivateUsage]
def __call__(self, ctx: ReportContext) -> str:
prompt = build_report_prompt(ctx, self.builder.pack.get(self.prompt_name))
retries = 0
while True:
self.stats.calls += 1
try:
return self.client.complete(prompt)
except BudgetExceededError:
# Budget exceeded during retry - treat as fallback
self.stats.fallbacks += 1
return ""
except Exception: # noqa: BLE001 - client may raise various exceptions
if retries >= self.max_retries:
self.stats.fallbacks += 1
return ""
retries += 1
self.stats.retries += 1
def format_llm_report(ctx: ReportContext, llm_text: str) -> str:
"""Combine the factual balance report with the LLM narrative section.
The result is the complete French report printed to stdout and/or written
to a file via `--output`.
Parameters
----------
ctx: The populated `ReportContext` with all available data.
llm_text: The narrative text returned by `LLMReportPolicy` (may be empty).
Returns
-------
Complete French report as a single string.
"""
factual = _render_encounter(ctx)
if not llm_text.strip():
return factual
return f"{factual}\n\n=== Récit d'équilibrage (LLM) ===\n{llm_text.strip()}"