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 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 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]] 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 _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} 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, "aggregate_reliability_meaningful": False, }, "language": state.get("language") or "English", "company_profile": 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 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 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) 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.content if isinstance(chunk.content, str) else "" 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")) 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 {} brief_dict["display_policy"] = { "event_returns_aligned": ( market.get("event_aligned") is True and market.get("event_comparison_allowed") is True ), "market_expectations_aligned": ( market.get("period_aligned") is True and market.get("comparison_allowed") is True ), "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 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("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", "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, "language": language, "brief": None, "brief_markdown": None, "synthesis_error": None, "coverage": None, "verification_report": None, }) return final.get("brief")