Spaces:
Sleeping
Sleeping
| import json | |
| import re | |
| from concurrent.futures import ThreadPoolExecutor | |
| from datetime import datetime, timezone | |
| from typing import TypedDict, Annotated, Optional | |
| from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, ToolMessage | |
| from langgraph.graph import StateGraph, END | |
| from langgraph.graph.message import add_messages | |
| from agent.tools import ( | |
| get_financial_metrics, | |
| search_filing, | |
| search_transcript, | |
| search_news, | |
| get_analyst_expectations, | |
| ) | |
| from agent.prompts import SYSTEM_PROMPT, SYNTHESIS_STRUCTURED_PROMPT, language_directive | |
| from agent.schemas import BriefOutput | |
| from agent.post_synthesis import apply_reliability, attach_edge_signals, apply_continuity_threads | |
| from agent.evidence import ( | |
| NO_VERIFIED_SYNTHESIS_MESSAGE, | |
| evidence_records_from, | |
| evidence_envelope, | |
| is_usable_evidence_payload, | |
| parse_evidence_envelope, | |
| ) | |
| from agent.llm import RunConfig, default_config, make_chat_model, build_system_message, ANTHROPIC_DEFAULT_MODEL | |
| from analytics.consensus import aligned_consensus | |
| TOOLS = [ | |
| get_financial_metrics, | |
| search_filing, | |
| search_transcript, | |
| search_news, | |
| get_analyst_expectations, | |
| ] | |
| MAX_TOOL_ROUNDS = 10 | |
| MODEL = ANTHROPIC_DEFAULT_MODEL # legacy alias — prefer agent.llm.RunConfig for new code | |
| TOOL_BY_NAME = {t.name: t for t in TOOLS} | |
| _EXPECTED_TOOL_SOURCES = { | |
| "get_financial_metrics": {"metrics"}, | |
| "search_filing": {"10-K", "10-Q"}, | |
| "search_transcript": {"transcript"}, | |
| "search_news": {"news"}, | |
| "get_analyst_expectations": {"analyst"}, | |
| } | |
| def _invoke_one(tc: dict) -> ToolMessage: | |
| """Invoke a single tool call and return a ToolMessage.""" | |
| name = tc["name"] | |
| args = tc.get("args") or {} | |
| tool_call_id = tc.get("id", "") | |
| try: | |
| fn = TOOL_BY_NAME.get(name) | |
| if fn is None: | |
| raise ValueError(f"Unknown tool: {name!r}") | |
| result = fn.invoke(args) | |
| content = result if isinstance(result, str) else str(result) | |
| except Exception as exc: | |
| content = evidence_envelope( | |
| tool=name, | |
| query=args, | |
| status="ERROR", | |
| message=str(exc), | |
| error_code="TOOL_INVOCATION_ERROR", | |
| ) | |
| return ToolMessage(tool_call_id=tool_call_id, name=name, content=content) | |
| # Tool names that the agent must touch before synthesis can proceed. | |
| # Without filing AND transcript evidence, the structured brief schema | |
| # (mda_summary, risks_categorized, management_commentary, etc.) cannot | |
| # be filled without fabrication. | |
| REQUIRED_TOOL_NAMES = {"search_filing", "search_transcript"} | |
| class AgentState(TypedDict): | |
| ticker: str | |
| messages: Annotated[list[BaseMessage], add_messages] | |
| tool_round_count: int | |
| nudge_fired: bool | |
| edge_signals: Optional[list[dict]] # precomputed deterministic signals | |
| profile_payloads: Optional[list[str]] | |
| open_threads: Optional[list[dict]] # currently-open continuity threads for this ticker | |
| language: Optional[str] # e.g. "French" — prose fields in brief will use this language | |
| brief: Optional[dict] | |
| brief_markdown: Optional[str] | |
| synthesis_error: Optional[str] | |
| coverage: Optional[dict] | |
| verification_report: Optional[dict] | |
| def _called_tool_names(messages: list[BaseMessage]) -> set[str]: | |
| return {m.name for m in messages if isinstance(m, ToolMessage) and getattr(m, "name", None)} | |
| def _filing_call_count(messages: list[BaseMessage]) -> int: | |
| return sum( | |
| 1 for m in messages | |
| if isinstance(m, ToolMessage) | |
| and getattr(m, "name", None) == "search_filing" | |
| and is_usable_evidence_payload(m) | |
| ) | |
| def _tool_envelopes(messages: list[BaseMessage], name: str) -> list[dict]: | |
| envelopes = [] | |
| for message in messages: | |
| if not isinstance(message, ToolMessage) or getattr(message, "name", None) != name: | |
| continue | |
| envelope = parse_evidence_envelope(message) | |
| if envelope and envelope.get("tool") == name: | |
| envelopes.append(envelope) | |
| return envelopes | |
| def _records_for_tool(envelope: dict, name: str): | |
| allowed = _EXPECTED_TOOL_SOURCES.get(name, set()) | |
| return [ | |
| record for record in evidence_records_from(envelope) | |
| if record.ref.source in allowed | |
| ] | |
| def _coverage_report(messages: list[BaseMessage]) -> dict: | |
| groups = { | |
| "metrics": "get_financial_metrics", | |
| "filings": "search_filing", | |
| "transcripts": "search_transcript", | |
| "analyst": "get_analyst_expectations", | |
| "news": "search_news", | |
| } | |
| report: dict = {} | |
| for label, name in groups.items(): | |
| envelopes = _tool_envelopes(messages, name) | |
| valid_counts = {id(e): len(_records_for_tool(e, name)) for e in envelopes} | |
| ok = [ | |
| e for e in envelopes | |
| if e.get("status") == "OK" and valid_counts[id(e)] > 0 | |
| ] | |
| invalid = [ | |
| e for e in envelopes | |
| if e.get("status") == "OK" and valid_counts[id(e)] == 0 | |
| ] | |
| if ok: | |
| status = "OK" | |
| elif any(e.get("status") == "ERROR" for e in envelopes): | |
| status = "ERROR" | |
| elif invalid: | |
| status = "INVALID" | |
| elif any(e.get("status") == "EMPTY" for e in envelopes): | |
| status = "EMPTY" | |
| else: | |
| status = "NOT_CALLED" | |
| report[label] = { | |
| "status": status, | |
| "successful_calls": len(ok), | |
| "evidence_count": sum(valid_counts[id(e)] for e in ok), | |
| "errors": [ | |
| (e.get("error") or {}).get("message", "tool error") | |
| for e in envelopes if e.get("status") == "ERROR" | |
| ] + (["invalid evidence record"] if invalid else []), | |
| } | |
| gaps = [] | |
| if report["metrics"]["evidence_count"] < 1: | |
| gaps.append("financial metrics unavailable") | |
| if report["filings"]["evidence_count"] < 1: | |
| gaps.append("filing evidence unavailable") | |
| elif report["filings"]["successful_calls"] < 2: | |
| gaps.append("only one filing search completed") | |
| if report["transcripts"]["evidence_count"] < 1: | |
| gaps.append("transcript evidence unavailable") | |
| report["gaps"] = gaps | |
| report["status"] = "COMPLETE" if not gaps else "PARTIAL" | |
| return report | |
| def _can_synthesize(messages: list[BaseMessage]) -> bool: | |
| coverage = _coverage_report(messages) | |
| return ( | |
| coverage["metrics"]["evidence_count"] >= 1 | |
| and coverage["filings"]["evidence_count"] >= 1 | |
| ) | |
| def _coverage_gaps(messages: list[BaseMessage]) -> list[str]: | |
| """Return retryable gaps; EMPTY is confirmed absence, not retryable.""" | |
| gaps = [] | |
| metrics = _tool_envelopes(messages, "get_financial_metrics") | |
| filings = _tool_envelopes(messages, "search_filing") | |
| transcripts = _tool_envelopes(messages, "search_transcript") | |
| if not any(e.get("status") == "OK" and _records_for_tool(e, "get_financial_metrics") for e in metrics): | |
| gaps.append("get_financial_metrics") | |
| filing_ok = sum( | |
| e.get("status") == "OK" and bool(_records_for_tool(e, "search_filing")) | |
| for e in filings | |
| ) | |
| if filing_ok < 2 and not any(e.get("status") == "EMPTY" for e in filings): | |
| gaps.append(f"search_filing (need 2 successful calls; {filing_ok} so far)") | |
| if not any( | |
| e.get("status") == "EMPTY" | |
| or (e.get("status") == "OK" and _records_for_tool(e, "search_transcript")) | |
| for e in transcripts | |
| ): | |
| gaps.append("search_transcript") | |
| return gaps | |
| def should_continue(state: AgentState) -> str: | |
| last = state["messages"][-1] | |
| under_cap = state["tool_round_count"] < MAX_TOOL_ROUNDS | |
| if getattr(last, "tool_calls", None) and under_cap: | |
| return "tools" | |
| # Agent stopped emitting tool calls. Check coverage floor before | |
| # routing to synthesis: require ≥2 filing calls + ≥1 transcript call. | |
| if not getattr(last, "tool_calls", None) and under_cap and not state.get("nudge_fired"): | |
| if _coverage_gaps(state["messages"]): | |
| return "nudge" | |
| return "synthesis" if _can_synthesize(state["messages"]) else "partial" | |
| def _chunk_text(content) -> str: | |
| """Text carried by one streamed chunk, whatever shape the provider used. | |
| A plain string when the model streams text only. Models that think — | |
| Claude Opus 5 runs adaptive thinking by default — stream a list of typed | |
| blocks instead. Take the text blocks and drop the rest: thinking is | |
| commentary, not the brief JSON we are parsing. | |
| """ | |
| if isinstance(content, str): | |
| return content | |
| if not isinstance(content, list): | |
| return "" | |
| parts = [] | |
| for block in content: | |
| if isinstance(block, str): | |
| parts.append(block) | |
| elif isinstance(block, dict) and block.get("type") == "text": | |
| parts.append(block.get("text") or "") | |
| return "".join(parts) | |
| def _extract_json(raw: str) -> str: | |
| """Extract the first complete JSON object from LLM output. | |
| Uses raw_decode so trailing content — extra commentary, a duplicate object | |
| (e.g. when the model was asked about multiple tickers), or a stray code | |
| fence — is ignored instead of triggering a json.loads 'Extra data' error. | |
| """ | |
| start = raw.find("{") | |
| if start != -1: | |
| try: | |
| _obj, end = json.JSONDecoder().raw_decode(raw, start) | |
| return raw[start:end] | |
| except json.JSONDecodeError: | |
| pass | |
| # Last-resort fallback: original first-{ to last-} heuristic | |
| end = raw.rfind("}") | |
| if start != -1 and end != -1 and end > start: | |
| return raw[start:end + 1] | |
| return raw | |
| def nudge_node(state: AgentState) -> dict: | |
| """One-shot prod when the agent tries to terminate without minimum coverage.""" | |
| gaps = _coverage_gaps(state["messages"]) | |
| gaps_str = " and ".join(f"`{g}`" for g in gaps) | |
| nudge = HumanMessage(content=( | |
| f"Before synthesizing, you must satisfy these coverage requirements: {gaps_str}. " | |
| "Filings (10-Q/10-K) are the primary source — `what_changed`, `bull_points`, " | |
| "`bear_points`, and `risks_categorized` must be predominantly filing-sourced. " | |
| "Issue the missing search call(s) now (batched where independent)." | |
| )) | |
| return {"messages": [nudge], "nudge_fired": True} | |
| # Checked in this order, and reported in it: revenue is the metric expected | |
| # to align (street revenue estimates track reported GAAP revenue), while eps | |
| # is expected to be refused as basis_mismatch, because yfinance publishes | |
| # adjusted estimates against this repo's GAAP reported figure. See | |
| # analytics/consensus.py's module docstring. | |
| _CONSENSUS_METRICS = ("revenue", "eps") | |
| def _market_expectations_aligned_metrics( | |
| ticker: str, period: Optional[str] | |
| ) -> list[str]: | |
| """The metrics whose consensus passes the alignment gate, e.g. ["revenue"]. | |
| THIS, not the boolean below, is what a consumer rendering an | |
| actual-versus-consensus line must check, and it must check it for the | |
| metric it is about to render. Normally only revenue aligns: eps is | |
| expected to be refused as basis_mismatch, and that refusal is the correct | |
| answer rather than a defect. A consumer that read the boolean and drew an | |
| EPS comparison would be drawing a comparison the gate explicitly refused. | |
| Order is stable and mirrors _CONSENSUS_METRICS. | |
| """ | |
| if not period: | |
| return [] | |
| aligned: list[str] = [] | |
| try: | |
| for metric in _CONSENSUS_METRICS: | |
| snapshot, _status = aligned_consensus(ticker, period, metric) | |
| if snapshot is not None: | |
| aligned.append(metric) | |
| except Exception: | |
| # An unreadable archive must never fail a brief. Fail closed: report | |
| # nothing aligned rather than a partial list of unknown provenance. | |
| return [] | |
| return aligned | |
| def _market_expectations_aligned(ticker: str, period: Optional[str]) -> bool: | |
| """True when AT LEAST ONE metric's consensus passes the alignment gate. | |
| Kept for backward compatibility, and deliberately coarse. It says nothing | |
| about WHICH metric aligned, so it is not sufficient authority to render | |
| any particular comparison — use | |
| ``_market_expectations_aligned_metrics`` (and the | |
| ``market_expectations_aligned_metrics`` display-policy key it feeds) for | |
| that. The gate itself is unchanged and still refuses by default; what | |
| changed is that the archive can now supply the pre-release estimate | |
| timestamp it always required. | |
| """ | |
| return bool(_market_expectations_aligned_metrics(ticker, period)) | |
| def _partial_brief(state: AgentState, reason: str) -> dict: | |
| coverage = _coverage_report(state.get("messages", [])) | |
| company_name = state["ticker"].upper() | |
| filing_date = "" | |
| for envelope in _tool_envelopes(state.get("messages", []), "get_financial_metrics"): | |
| for record in envelope.get("records") or []: | |
| metadata = record.get("metadata") or {} | |
| company_name = metadata.get("company_name") or company_name | |
| filing_date = metadata.get("filing_date") or filing_date | |
| if filing_date: | |
| break | |
| if not filing_date or company_name == state["ticker"].upper(): | |
| try: | |
| from storage.metrics_db import get_metrics | |
| metrics = get_metrics(state["ticker"]) | |
| if isinstance(metrics, dict): | |
| if company_name == state["ticker"].upper(): | |
| company_name = metrics.get("company_name") or company_name | |
| if not filing_date: | |
| filing_date = metrics.get("filing_date") or filing_date | |
| except Exception: | |
| pass | |
| generated_at = datetime.now(timezone.utc).isoformat() | |
| return { | |
| "ticker": state["ticker"].upper(), | |
| "company_name": company_name, | |
| "filing_date": filing_date, | |
| "status": "PARTIAL", | |
| "schema_version": "brief.v2", | |
| "generated_at": generated_at, | |
| "data_as_of": filing_date, | |
| "what_matters_most": NO_VERIFIED_SYNTHESIS_MESSAGE, | |
| "non_obvious_takeaway": "", | |
| "analytical_tensions": [], | |
| "between_the_lines": [], | |
| "earnings_quality_signals": [], | |
| "standout_number": None, | |
| "what_changed": [], | |
| "bull_points": [], | |
| "bear_points": [], | |
| "what_to_watch": [], | |
| "trends": [], | |
| "evidence_notes": [reason] + list(coverage.get("gaps") or []), | |
| "evidence_coverage": { | |
| "status": "INCOMPLETE", "verified": 0, "unverified": 0, | |
| "failed": 0, "total": 0, | |
| }, | |
| "verification_report": { | |
| "verified": 0, "unverified": 0, "failed": 0, "removed": 0, | |
| }, | |
| "coverage": coverage, | |
| "mda_summary": { | |
| "drivers": [], "headwinds": [], "language_shift": "", "key_quote": None, | |
| }, | |
| "risks_categorized": [], | |
| "management_commentary": [], | |
| "guidance_history": [], | |
| "sentiment": None, | |
| "market_expectations": None, | |
| "quarter_deltas": state.get("edge_signals") or [], | |
| "display_policy": { | |
| "event_returns_aligned": False, | |
| "market_expectations_aligned": False, | |
| # A partial brief has no verified market data, so no metric may | |
| # be shown against consensus. Present-and-empty, never absent: a | |
| # consumer must not have to tell "nothing aligned" from "this | |
| # brief predates the key". | |
| "market_expectations_aligned_metrics": [], | |
| "aggregate_reliability_meaningful": False, | |
| }, | |
| "language": state.get("language") or "English", | |
| "company_profile": None, | |
| "narrative": None, | |
| "narrative_usage": None, | |
| } | |
| def _format_signals_message(signals: list[dict]) -> str: | |
| """Format precomputed edge signals as a compact labelled block for the agent.""" | |
| lines = ["== PRECOMPUTED EDGE SIGNALS (deterministic, no LLM) ==\n"] | |
| kind_labels = { | |
| "risk_added": "NEW RISK", | |
| "risk_removed": "REMOVED RISK", | |
| "risk_reworded": "REWORDED RISK", | |
| "guidance_language_shift": "GUIDANCE LANGUAGE SHIFT", | |
| "term_frequency": "TERM FREQUENCY SHIFT", | |
| "kpi_dropped": "DROPPED KPI", | |
| "tone_trend": "MANAGEMENT TONE TREND", | |
| "topic_arc": "TRANSCRIPT TOPIC ARC", | |
| "recurring_evasion": "RECURRING Q&A EVASION", | |
| "topic_fade": "PREPARED-REMARKS TOPIC FADE", | |
| } | |
| for i, s in enumerate(signals, 1): | |
| kind = s.get("kind", "") | |
| label = kind_labels.get(kind, kind.upper()) | |
| sig = s.get("significance", "MEDIUM") | |
| term = s.get("term", "") | |
| term_str = f" — {term}" if term else "" | |
| lines.append(f"[SIG-{i}] {label}{term_str} [{sig}]") | |
| if s.get("before_text"): | |
| lines.append(f" BEFORE ({s.get('period_from','')}): \"{s['before_text']}\"") | |
| if s.get("after_text"): | |
| lines.append(f" AFTER ({s.get('period_to','')}): \"{s['after_text']}\"") | |
| if s.get("computed_metric"): | |
| lines.append(f" METRIC: {s['computed_metric']}") | |
| lines.append("") | |
| lines.append("== END PRECOMPUTED EDGE SIGNALS ==") | |
| return "\n".join(lines) | |
| MAX_EDGE_SIGNALS = 12 | |
| MAX_FILING_SIGNALS = 8 | |
| MAX_TRANSCRIPT_SIGNALS = 6 | |
| # Output budget for the single-call synthesis (brief + company profile + | |
| # one full evidence_ref per fact). 16384 was inherited from the pre-merge | |
| # two-call split and truncated real AAPL briefs mid-JSON (stop_reason= | |
| # max_tokens at exactly 16384 output tokens). 64000 is the claude-haiku-4-5 | |
| # output ceiling; billing only covers tokens actually generated. | |
| SYNTHESIS_MAX_TOKENS = 64000 | |
| # Output budget for one agent-loop turn (tool calls, no prose). LangChain | |
| # defaults to 1024, which is ample for a tool call but not for a model that | |
| # thinks first — thinking tokens count against max_tokens, and Claude Opus 5 | |
| # runs adaptive thinking by default. Unused headroom is not billed. | |
| AGENT_MAX_TOKENS = 8192 | |
| class SynthesisTruncatedError(RuntimeError): | |
| """The synthesis stream was cut off by the max_tokens limit.""" | |
| def _cap_signals(signals: list[dict], max_total: int = MAX_EDGE_SIGNALS) -> list[dict]: | |
| """Bound the edge-signal block: per-source caps, then a global cap. | |
| Sorts HIGH→MEDIUM→LOW, keeps at most MAX_FILING_SIGNALS filing-sourced and | |
| MAX_TRANSCRIPT_SIGNALS transcript-sourced signals, then truncates to | |
| max_total so neither module can flood the prompt. | |
| """ | |
| order = {"HIGH": 0, "MEDIUM": 1, "LOW": 2} | |
| signals = sorted(signals, key=lambda s: order.get(s.get("significance", "MEDIUM"), 2)) | |
| capped: list[dict] = [] | |
| filing_n = transcript_n = 0 | |
| for s in signals: | |
| if s.get("source") == "transcript": | |
| if transcript_n >= MAX_TRANSCRIPT_SIGNALS: | |
| continue | |
| transcript_n += 1 | |
| else: | |
| if filing_n >= MAX_FILING_SIGNALS: | |
| continue | |
| filing_n += 1 | |
| capped.append(s) | |
| return capped[:max_total] | |
| def signals_node(state: AgentState) -> dict: | |
| """Run deterministic analysis modules and inject signals into conversation.""" | |
| ticker = state["ticker"] | |
| raw_signals = [] | |
| try: | |
| from analysis.textdiff import compute as compute_text_deltas | |
| raw_signals.extend(compute_text_deltas(ticker)) | |
| except Exception as exc: | |
| import sys | |
| print(f"[signals_node] textdiff error: {exc}", file=sys.stderr) | |
| try: | |
| from analysis.tone_drift import compute as compute_tone_deltas | |
| raw_signals.extend(compute_tone_deltas(ticker)) | |
| except Exception as exc: | |
| import sys | |
| print(f"[signals_node] tone_drift error: {exc}", file=sys.stderr) | |
| signals = _cap_signals([s.model_dump() for s in raw_signals]) | |
| if signals: | |
| msg = HumanMessage(content=_format_signals_message(signals)) | |
| return {"edge_signals": signals, "messages": [msg]} | |
| return {"edge_signals": [], "messages": []} | |
| def _format_open_threads_message(threads: list[dict]) -> str: | |
| """Format currently-open continuity threads as a compact labelled block.""" | |
| lines = ["== OPEN THREADS FROM PRIOR QUARTERS ==\n"] | |
| for t in threads: | |
| last_note = (t.get("history") or [{}])[-1].get("note", "") | |
| lines.append( | |
| f"[{t['thread_id']}] {t['topic']} ({t['category']}, {t['status']} " | |
| f"since {t.get('last_updated_period', '?')}): {last_note}" | |
| ) | |
| return "\n".join(lines) | |
| def continuity_node(state: AgentState) -> dict: | |
| """Load open continuity threads for this ticker and inject them as context.""" | |
| from storage import threads_db | |
| try: | |
| open_threads = threads_db.get_open_threads(state["ticker"]) | |
| except Exception as exc: | |
| import sys | |
| print(f"[continuity_node] failed to load open threads: {exc}", file=sys.stderr) | |
| return {"open_threads": [], "messages": []} | |
| if not open_threads: | |
| return {"open_threads": [], "messages": []} | |
| msg = HumanMessage(content=_format_open_threads_message(open_threads)) | |
| return {"open_threads": open_threads, "messages": [msg]} | |
| def profile_evidence_node(state: AgentState) -> dict: | |
| """Inject deterministic profile evidence as parseable human messages.""" | |
| from agent.company_profile import collect_profile_evidence | |
| payloads = collect_profile_evidence(state["ticker"], include_metrics=False) | |
| if not payloads: | |
| return {"profile_payloads": [], "messages": []} | |
| messages = [ | |
| HumanMessage( | |
| content=( | |
| "== COMPANY PROFILE EVIDENCE " | |
| "(deterministic retrieval, evidence.v1 envelopes follow) ==" | |
| ) | |
| ) | |
| ] | |
| messages.extend(HumanMessage(content=payload) for payload in payloads) | |
| return {"profile_payloads": payloads, "messages": messages} | |
| def _pop_company_profile(data: dict) -> dict | None: | |
| """Remove the separately validated profile section from synthesis output.""" | |
| return data.pop("company_profile", None) | |
| def _finalize_synthesis_profile( | |
| state: AgentState, | |
| profile_section: dict, | |
| model: str | None, | |
| ) -> dict | None: | |
| """Finalize and persist a synthesized profile without risking the brief.""" | |
| try: | |
| from agent.company_profile import finalize_profile_from_synthesis | |
| from storage import company_profiles | |
| all_payloads = list(state.get("messages") or []) + list( | |
| state.get("profile_payloads") or [] | |
| ) | |
| profile = finalize_profile_from_synthesis( | |
| state["ticker"], profile_section, all_payloads, model | |
| ) | |
| company_profiles.save_profile(state["ticker"], profile) | |
| return profile | |
| except Exception as exc: | |
| import sys | |
| print( | |
| f"[synthesis] company profile finalization failed: {exc}", | |
| file=sys.stderr, | |
| ) | |
| return None | |
| def create_graph(config: Optional[RunConfig] = None): | |
| cfg = config or default_config() | |
| llm = make_chat_model(cfg, max_tokens=AGENT_MAX_TOKENS) | |
| llm_with_tools = llm.bind_tools(TOOLS) | |
| def agent_node(state: AgentState) -> dict: | |
| system_block = build_system_message(cfg, SYSTEM_PROMPT) | |
| response = llm_with_tools.invoke([system_block] + state["messages"]) | |
| return {"messages": [response]} | |
| def tool_node(state: AgentState) -> dict: | |
| last = state["messages"][-1] | |
| tool_calls = getattr(last, "tool_calls", None) or [] | |
| if not tool_calls: | |
| return {"messages": [], "tool_round_count": state["tool_round_count"] + 1} | |
| with ThreadPoolExecutor(max_workers=min(8, len(tool_calls))) as ex: | |
| results = list(ex.map(_invoke_one, tool_calls)) | |
| return {"messages": results, "tool_round_count": state["tool_round_count"] + 1} | |
| def partial_node(state: AgentState) -> dict: | |
| brief = _partial_brief(state, "Required evidence was unavailable or invalid.") | |
| return { | |
| "brief": brief, | |
| "brief_markdown": None, | |
| "synthesis_error": None, | |
| "coverage": brief["coverage"], | |
| "verification_report": brief["verification_report"], | |
| } | |
| def synthesis_node(state: AgentState) -> dict: | |
| try: | |
| llm_plain = make_chat_model(cfg, max_tokens=SYNTHESIS_MAX_TOKENS) | |
| # Main prompt — cached (ephemeral) on Anthropic. Keep this block stable | |
| # so the cache hit rate is preserved regardless of the chosen language. | |
| lang = state.get("language") or "English" | |
| extra_texts = [language_directive(lang)] if lang != "English" else None | |
| system_block = build_system_message(cfg, SYNTHESIS_STRUCTURED_PROMPT, extra_texts) | |
| synthesis_messages = [system_block] + state["messages"] | |
| # If cap was hit mid-round the last AIMessage may still carry tool_calls. | |
| # Anthropic rejects conversations where tool_use blocks have no matching | |
| # tool_result — insert stubs so the message history is valid. | |
| last_msg = synthesis_messages[-1] | |
| if getattr(last_msg, "tool_calls", None): | |
| stubs = [ | |
| ToolMessage( | |
| tool_call_id=tc["id"], | |
| name=tc["name"], | |
| content=evidence_envelope( | |
| tool=tc["name"], query=tc.get("args") or {}, status="ERROR", | |
| message="Tool call interrupted at round cap.", | |
| error_code="ROUND_CAP_INTERRUPTED", | |
| ), | |
| ) | |
| for tc in last_msg.tool_calls | |
| ] | |
| synthesis_messages = synthesis_messages + stubs | |
| if not isinstance(synthesis_messages[-1], HumanMessage): | |
| synthesis_messages = synthesis_messages + [ | |
| HumanMessage(content="Now produce the structured research brief as a JSON object.") | |
| ] | |
| chunks = [] | |
| stop_reason = None | |
| for chunk in llm_plain.stream(synthesis_messages): | |
| metadata = getattr(chunk, "response_metadata", None) or {} | |
| stop_reason = ( | |
| metadata.get("stop_reason") | |
| or metadata.get("finish_reason") | |
| or stop_reason | |
| ) | |
| text = _chunk_text(chunk.content) | |
| if text: | |
| chunks.append(text) | |
| if stop_reason in ("max_tokens", "length"): | |
| raise SynthesisTruncatedError( | |
| "Synthesis output was truncated by the token limit " | |
| f"(max_tokens={SYNTHESIS_MAX_TOKENS}); the brief JSON was incomplete." | |
| ) | |
| raw = "".join(chunks) | |
| clean = _extract_json(raw) | |
| data = json.loads(clean) | |
| profile_section = _pop_company_profile(data) | |
| brief = BriefOutput.model_validate(data) | |
| brief_dict = apply_reliability( | |
| brief.model_dump(), evidence_payloads=state.get("messages", []) | |
| ) | |
| brief_dict = attach_edge_signals(brief_dict, state.get("edge_signals")) | |
| try: | |
| brief_dict["continuity_threads"] = apply_continuity_threads( | |
| state["ticker"], brief_dict.get("filing_date") or "", brief_dict.get("continuity_threads") or [] | |
| ) | |
| except Exception as exc: | |
| import sys | |
| # Leave brief_dict["continuity_threads"] as the LLM's raw, unpersisted | |
| # output rather than letting a threads_db failure (e.g. "database is | |
| # locked" on the shared HF Space) propagate up to the blanket | |
| # `except Exception` below and discard an otherwise fully-synthesized, | |
| # good brief. | |
| print(f"[synthesis] continuity thread persistence failed: {exc}", file=sys.stderr) | |
| brief_dict["language"] = lang | |
| coverage = _coverage_report(state.get("messages", [])) | |
| evidence_coverage = brief_dict.get("evidence_coverage") or {} | |
| if evidence_coverage.get("verified", 0) < 1: | |
| note = ( | |
| "Synthesis contained no claim that passed deterministic " | |
| "evidence verification." | |
| ) | |
| brief_dict["evidence_notes"] = ( | |
| list(brief_dict.get("evidence_notes") or []) + [note] | |
| )[:6] | |
| generated_at = datetime.now(timezone.utc).isoformat() | |
| evidence_dates = [] | |
| for message in state.get("messages", []): | |
| envelope = parse_evidence_envelope(message) | |
| if not envelope or envelope.get("status") != "OK": | |
| continue | |
| evidence_dates.extend( | |
| record.get("ref", {}).get("as_of") | |
| for record in envelope.get("records") or [] | |
| if record.get("ref", {}).get("as_of") | |
| ) | |
| brief_dict["schema_version"] = "brief.v2" | |
| brief_dict["generated_at"] = generated_at | |
| brief_dict["model"] = cfg.model | |
| brief_dict["data_as_of"] = max(evidence_dates) if evidence_dates else brief_dict.get("filing_date", "") | |
| brief_dict["coverage"] = coverage | |
| if coverage.get("status") != "COMPLETE" or evidence_coverage.get("status") != "VERIFIED": | |
| brief_dict["status"] = "PARTIAL" | |
| market = brief_dict.get("market_expectations") or {} | |
| aligned_metrics = _market_expectations_aligned_metrics( | |
| state["ticker"], market.get("target_period") | |
| ) | |
| brief_dict["display_policy"] = { | |
| "event_returns_aligned": ( | |
| market.get("event_aligned") is True | |
| and market.get("event_comparison_allowed") is True | |
| ), | |
| # Coarse: "at least one metric aligned". The provider-flag | |
| # branch is payload-wide and names no metric at all, so it | |
| # can set this True while the per-metric list below is empty. | |
| "market_expectations_aligned": ( | |
| ( | |
| market.get("period_aligned") is True | |
| and market.get("comparison_allowed") is True | |
| ) | |
| or bool(aligned_metrics) | |
| ), | |
| # Per-metric, from the archive gate. Anything rendering an | |
| # actual-versus-consensus figure for a specific metric must | |
| # check membership here, not the boolean above. | |
| "market_expectations_aligned_metrics": aligned_metrics, | |
| "aggregate_reliability_meaningful": False, | |
| } | |
| if isinstance(profile_section, dict) and profile_section: | |
| brief_dict["company_profile"] = _finalize_synthesis_profile( | |
| state, profile_section, cfg.model | |
| ) | |
| else: | |
| brief_dict["company_profile"] = None | |
| # The press-style narrative is composed in a second call that runs | |
| # AFTER verification and sees only the facts that passed it, so it | |
| # cannot cite unverified material. It is strictly additive: any | |
| # failure here leaves narrative=None and the brief untouched. | |
| try: | |
| from agent.narrative import compose_narrative | |
| # Product and region results come from the filing's XBRL | |
| # segment dimensions, not from anything the agent retrieved, | |
| # so they are attached here rather than synthesised. | |
| try: | |
| from analytics.segments import digest_lines | |
| brief_dict["segment_lines"] = digest_lines(state["ticker"]) | |
| except Exception as exc: | |
| import sys | |
| print(f"[narrative] segment lines unavailable: {exc}", file=sys.stderr) | |
| article, narrative_report, narrative_usage = compose_narrative( | |
| brief_dict, cfg, lang | |
| ) | |
| brief_dict["narrative"] = article | |
| # Kept whether or not the article survived: the tokens were | |
| # spent either way, and the run's logged cost must say so. | |
| brief_dict["narrative_usage"] = narrative_usage | |
| if article is None and narrative_report: | |
| import sys | |
| print( | |
| "[narrative] no article: " + "; ".join(narrative_report), | |
| file=sys.stderr, | |
| ) | |
| except Exception as exc: | |
| import sys | |
| brief_dict["narrative"] = None | |
| print(f"[narrative] skipped: {exc}", file=sys.stderr) | |
| return { | |
| "brief": brief_dict, | |
| "brief_markdown": None, | |
| "synthesis_error": None, | |
| "coverage": coverage, | |
| "verification_report": brief_dict.get("verification_report"), | |
| } | |
| except Exception as exc: | |
| import sys | |
| print(f"[synthesis error] {exc}", file=sys.stderr) | |
| if isinstance(exc, SynthesisTruncatedError): | |
| reason = str(exc) | |
| else: | |
| reason = f"Synthesis failed validation: {exc}" | |
| partial = _partial_brief(state, reason) | |
| return { | |
| "brief": partial, | |
| "brief_markdown": None, | |
| "synthesis_error": str(exc), | |
| "coverage": partial["coverage"], | |
| "verification_report": partial["verification_report"], | |
| } | |
| builder = StateGraph(AgentState) | |
| builder.add_node("signals", signals_node) | |
| builder.add_node("profile_evidence", profile_evidence_node) | |
| builder.add_node("continuity", continuity_node) | |
| builder.add_node("agent", agent_node) | |
| builder.add_node("tools", tool_node) | |
| builder.add_node("nudge", nudge_node) | |
| builder.add_node("synthesis", synthesis_node) | |
| builder.add_node("partial", partial_node) | |
| builder.set_entry_point("signals") | |
| builder.add_edge("signals", "profile_evidence") | |
| builder.add_edge("profile_evidence", "continuity") | |
| builder.add_edge("continuity", "agent") | |
| builder.add_conditional_edges( | |
| "agent", | |
| should_continue, | |
| {"tools": "tools", "nudge": "nudge", "synthesis": "synthesis", "partial": "partial"}, | |
| ) | |
| builder.add_edge("tools", "agent") | |
| builder.add_edge("nudge", "agent") | |
| builder.add_edge("synthesis", END) | |
| builder.add_edge("partial", END) | |
| return builder.compile() | |
| def run_brief(ticker: str, language: str = "English", config: Optional[RunConfig] = None) -> Optional[dict]: | |
| graph = create_graph(config) | |
| final = graph.invoke({ | |
| "ticker": ticker.upper(), | |
| "messages": [HumanMessage(content=f"Generate a research brief for {ticker.upper()}.")], | |
| "tool_round_count": 0, | |
| "nudge_fired": False, | |
| "edge_signals": None, | |
| "profile_payloads": None, | |
| "open_threads": None, | |
| "language": language, | |
| "brief": None, | |
| "brief_markdown": None, | |
| "synthesis_error": None, | |
| "coverage": None, | |
| "verification_report": None, | |
| }) | |
| return final.get("brief") | |