Spaces:
Running
Running
| """Writer agent — assembles the brief. Computes nothing. | |
| The claim table is minted deterministically from tool-computed metrics *before* | |
| the model is asked for anything, and the model's returned claim list is | |
| overwritten with that canonical table. The model therefore controls only: | |
| * which verified claim each sentence cites, and | |
| * the prose around it. | |
| It cannot alter a value, and the schema rejects prose containing a bare numeral. | |
| Whatever it returns is then recomputed from raw price bars by the verifier. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| from typing import Any | |
| from pydantic import ValidationError | |
| from app.core.budget import BudgetExceededError | |
| from app.core.claude import AgentRole, LLMRequest, PromptHint | |
| from app.core.events import EventKind | |
| from app.core.security import harden_untrusted_text | |
| from app.graph.compose import compose_brief | |
| from app.graph.context import RunContext, current_context | |
| from app.graph.llm import call_model | |
| from app.graph.prompts import WRITER_SYSTEM, build_emit_brief_tool | |
| from app.graph.state import RunState | |
| from app.models.brief import Brief | |
| from app.models.run import RunError, RunMode, RunStatus, TokenSpend | |
| logger = logging.getLogger(__name__) | |
| #: How far the injected-fault demo mode shifts a claim, in the claim's own units. | |
| DEMO_MISMATCH_DELTA = 7.77 | |
| def data_gaps_from_state(state: RunState) -> list[str]: | |
| """Human-readable gaps, derived from recorded run errors.""" | |
| gaps: list[str] = [] | |
| seen: set[str] = set() | |
| for error in state.get("errors", []): | |
| if error.severity != "error": | |
| continue | |
| label = f"{error.ticker or 'run'}: {error.message}" | |
| if label in seen: | |
| continue | |
| seen.add(label) | |
| gaps.append(label) | |
| return gaps[:10] | |
| def build_canonical_brief(state: RunState) -> Brief: | |
| """Mint the verified claim table and a complete deterministic brief.""" | |
| return compose_brief( | |
| session_date=str(state.get("session_date", "")), | |
| tickers=list(state.get("tickers", [])), | |
| metrics=state.get("metrics", {}), | |
| fundamentals=state.get("fundamentals", {}), | |
| sentiment=state.get("sentiment", {}), | |
| news=state.get("news", {}), | |
| risk_events=list(state.get("risk_events", [])), | |
| data_gaps=data_gaps_from_state(state), | |
| ) | |
| def _writer_context_payload(state: RunState, canonical: Brief) -> dict[str, Any]: | |
| """Everything the model may cite, already verified.""" | |
| headlines: dict[str, list[str]] = {} | |
| for ticker, feed in state.get("news", {}).items(): | |
| headlines[ticker] = [harden_untrusted_text(item.title) for item in feed.items[:6]] | |
| return { | |
| "session_date": state.get("session_date"), | |
| "watchlist": list(state.get("tickers", [])), | |
| "claims": [claim.model_dump() for claim in canonical.claims], | |
| "snapshot_tickers": [row.ticker for row in canonical.snapshot], | |
| "companies": {row.ticker: row.company for row in canonical.snapshot if row.company}, | |
| "sentiment_labels": { | |
| ticker: value.label for ticker, value in state.get("sentiment", {}).items() | |
| }, | |
| "risk_events": [event.model_dump() for event in state.get("risk_events", [])], | |
| "headlines": headlines, | |
| "data_gaps": canonical.data_gaps, | |
| "partial": canonical.partial, | |
| } | |
| async def _ask_model_for_brief( | |
| ctx: RunContext, | |
| state: RunState, | |
| canonical: Brief, | |
| mismatch_feedback: list[str], | |
| ) -> tuple[Brief | None, TokenSpend | None, list[RunError]]: | |
| """Ask Claude to write the brief; fall back to the canonical one on failure.""" | |
| payload = _writer_context_payload(state, canonical) | |
| schema = Brief.model_json_schema() | |
| tool = build_emit_brief_tool(schema) | |
| instruction = ( | |
| "Write the morning brief using ONLY the verified claims below. Reference a " | |
| "figure as {{claim_id}} — for example {{c1}}. Do not write any numeral in " | |
| "narrative prose. Copy quoted headlines character-for-character.\n\n" | |
| f"VERIFIED STATE:\n{json.dumps(payload, indent=2, default=str)[:24000]}" | |
| ) | |
| if mismatch_feedback: | |
| instruction += ( | |
| "\n\nYour previous attempt failed deterministic verification for these " | |
| "reasons. Fix them exactly — do not restate a figure differently, cite the " | |
| "correct claim id instead:\n- " + "\n- ".join(mismatch_feedback[:10]) | |
| ) | |
| errors: list[RunError] = [] | |
| outcome = await call_model( | |
| ctx, | |
| LLMRequest( | |
| role=AgentRole.WRITER, | |
| hint=PromptHint.WRITER_COMPOSE, | |
| system=WRITER_SYSTEM, | |
| messages=[{"role": "user", "content": instruction}], | |
| tools=[tool], | |
| forced_tool=tool.name, | |
| max_tokens=ctx.settings.llm_max_tokens, | |
| context={ | |
| "session_date": state.get("session_date"), | |
| "tickers": list(state.get("tickers", [])), | |
| "metrics": state.get("metrics", {}), | |
| "fundamentals": state.get("fundamentals", {}), | |
| "sentiment": state.get("sentiment", {}), | |
| "news": state.get("news", {}), | |
| "risk_events": list(state.get("risk_events", [])), | |
| "data_gaps": canonical.data_gaps, | |
| }, | |
| ), | |
| ) | |
| raw = outcome.result.first_tool(tool.name) | |
| if not raw: | |
| errors.append( | |
| RunError( | |
| stage="writer", | |
| message="writer returned no emit_brief call; using the deterministic assembly", | |
| severity="warning", | |
| ) | |
| ) | |
| return None, outcome.spend, errors | |
| # The model never controls values: the canonical claim table always wins. | |
| raw["claims"] = [claim.model_dump() for claim in canonical.claims] | |
| raw.setdefault("generated_for", canonical.generated_for) | |
| raw.setdefault("watchlist", canonical.watchlist) | |
| raw["partial"] = canonical.partial | |
| try: | |
| return Brief.model_validate(raw), outcome.spend, errors | |
| except ValidationError as exc: | |
| detail = "; ".join( | |
| f"{'.'.join(str(p) for p in err['loc'])}: {err['msg']}" for err in exc.errors()[:5] | |
| ) | |
| logger.warning("writer output failed schema validation: %s", detail) | |
| errors.append( | |
| RunError( | |
| stage="writer", | |
| message=f"writer output rejected by schema ({detail}); " | |
| "fell back to the deterministic assembly", | |
| severity="warning", | |
| ) | |
| ) | |
| return None, outcome.spend, errors | |
| def _inject_demo_mismatch(brief: Brief) -> Brief: | |
| """Corrupt one claim so the verification screen visibly goes red.""" | |
| if not brief.claims: | |
| return brief | |
| claims = [claim.model_copy() for claim in brief.claims] | |
| target = claims[0] | |
| claims[0] = target.model_copy(update={"value": target.value + DEMO_MISMATCH_DELTA}) | |
| return brief.model_copy(update={"claims": claims}) | |
| async def writer_node(state: RunState) -> dict[str, Any]: | |
| """Graph node: produce the Pydantic-enforced brief.""" | |
| ctx = current_context() | |
| previous_report = state.get("verification") | |
| is_regeneration = previous_report is not None | |
| regenerations = int(state.get("regenerations", 0)) + (1 if is_regeneration else 0) | |
| await ctx.emit( | |
| EventKind.WRITER_STARTED, | |
| "Regenerating the brief after failed verification" | |
| if is_regeneration | |
| else "Assembling the brief from verified state", | |
| {"regeneration": is_regeneration}, | |
| ) | |
| canonical = build_canonical_brief(state) | |
| errors: list[RunError] = [] | |
| spend: TokenSpend | None = None | |
| brief = canonical | |
| if ctx.engine.name == "anthropic": | |
| feedback: list[str] = [] | |
| if isinstance(previous_report, dict): | |
| checks = previous_report.get("claim_checks", []) | |
| feedback = [ | |
| f"{c.get('claim_id')} ({c.get('ticker')}.{c.get('metric')}): {c.get('detail')}" | |
| for c in checks | |
| if isinstance(c, dict) and c.get("status") != "match" | |
| ] | |
| feedback += list(previous_report.get("structural_issues", [])) | |
| try: | |
| with ctx.tracer.step("writer", run_id=ctx.run_id) as span: | |
| model_brief, spend, model_errors = await _ask_model_for_brief( | |
| ctx, state, canonical, feedback | |
| ) | |
| span.update(output={"used_model_output": model_brief is not None}) | |
| errors.extend(model_errors) | |
| if model_brief is not None: | |
| brief = model_brief | |
| except BudgetExceededError as exc: | |
| await ctx.emit(EventKind.RUN_FAILED, "Budget exhausted while writing", {}) | |
| return { | |
| "status": RunStatus.BUDGET_ABORT, | |
| "abort_reason": str(exc), | |
| "errors": [RunError(stage="writer", message=str(exc))], | |
| "regenerations": regenerations, | |
| } | |
| if state.get("mode") == RunMode.DEMO_MISMATCH: | |
| brief = _inject_demo_mismatch(brief) | |
| errors.append( | |
| RunError( | |
| stage="writer", | |
| message="demo_mismatch mode: a claim value was deliberately corrupted " | |
| "to exercise the verifier", | |
| severity="warning", | |
| ) | |
| ) | |
| await ctx.emit( | |
| EventKind.WRITER_COMPLETED, | |
| f"Brief assembled — {len(brief.claims)} claims, {len(brief.snapshot)} snapshot rows", | |
| { | |
| "claims": len(brief.claims), | |
| "snapshot_rows": len(brief.snapshot), | |
| "partial": brief.partial, | |
| "headline": brief.headline, | |
| }, | |
| ) | |
| update: dict[str, Any] = { | |
| "brief": brief, | |
| "regenerations": regenerations, | |
| "errors": errors, | |
| } | |
| if spend is not None: | |
| update["token_spend"] = spend | |
| return update | |