diff --git a/sales-agent/data/accounts.json b/sales-agent/data/accounts.json index bc67b37..ff7c2ab 100644 --- a/sales-agent/data/accounts.json +++ b/sales-agent/data/accounts.json @@ -1,5 +1,5 @@ { - "_note": "Pilot target list (7 accounts; ~200 total to come). REAL fields: accountName, domain, industry. PLACEHOLDER fields to confirm from CRM/install-base: accountTier, assignedRep, currentProducts (drives anti-signal suppression — empty here means nothing is suppressed yet), targetContacts (empty; the LLM/signal authors identify the ideal contact, CRM enrichment fills the rest).", + "_note": "Pilot target list (~200 total to come). REAL fields: accountName, domain, industry, crmTier (verbatim CRM tier where known). crmTier->accountTier mapping used by the 3-band scorer: Tier 1->strategic, Tier 2->enterprise, Tier 3->growth, Tier 4->growth (NOTE: Tier 3 & 4 currently collapse to the same scoring weight — extend scorer to 4 bands to distinguish). PLACEHOLDER fields to confirm from CRM/install-base: assignedRep, currentProducts (drives anti-signal suppression — empty means nothing suppressed yet), targetContacts (empty; LLM/signal authors identify the ideal contact, CRM enrichment fills the rest).", "accounts": [ { "accountId": "acct_lowes", @@ -77,6 +77,66 @@ "currentProducts": [], "targetContacts": [], "lastRunAt": null + }, + { + "accountId": "acct_booz_allen", + "accountName": "Booz Allen Hamilton", + "domain": "boozallen.com", + "industry": "Consulting & Government Services", + "crmTier": "Tier 3", + "accountTier": "growth", + "assignedRep": { "id": "rep_priya", "name": "Priya R.", "email": "priya.r@yourco.com" }, + "currentProducts": [], + "targetContacts": [], + "lastRunAt": null + }, + { + "accountId": "acct_dominion", + "accountName": "Dominion Energy", + "domain": "dominionenergy.com", + "industry": "Energy & Utilities", + "crmTier": "Tier 1", + "accountTier": "strategic", + "assignedRep": { "id": "rep_priya", "name": "Priya R.", "email": "priya.r@yourco.com" }, + "currentProducts": [], + "targetContacts": [], + "lastRunAt": null + }, + { + "accountId": "acct_hope_gas", + "accountName": "Hope Gas Holdings", + "domain": "hopegas.com", + "industry": "Energy & Utilities (Natural Gas)", + "crmTier": "Tier 3", + "accountTier": "growth", + "assignedRep": { "id": "rep_priya", "name": "Priya R.", "email": "priya.r@yourco.com" }, + "currentProducts": [], + "targetContacts": [], + "lastRunAt": null + }, + { + "accountId": "acct_markel", + "accountName": "Markel", + "domain": "markel.com", + "industry": "Insurance", + "crmTier": "Tier 3", + "accountTier": "growth", + "assignedRep": { "id": "rep_marcus", "name": "Marcus L.", "email": "marcus.l@yourco.com" }, + "currentProducts": [], + "targetContacts": [], + "lastRunAt": null + }, + { + "accountId": "acct_gw_mfa", + "accountName": "GW Medical Faculty Associates", + "domain": "gwdocs.com", + "industry": "Healthcare", + "crmTier": "Tier 4", + "accountTier": "growth", + "assignedRep": { "id": "rep_marcus", "name": "Marcus L.", "email": "marcus.l@yourco.com" }, + "currentProducts": [], + "targetContacts": [], + "lastRunAt": null } ] } diff --git a/sales-agent/src/agents/relevance.py b/sales-agent/src/agents/relevance.py index 87a5947..2b73958 100644 --- a/sales-agent/src/agents/relevance.py +++ b/sales-agent/src/agents/relevance.py @@ -1,55 +1,79 @@ """Phase 3: Relevance filtering against the Capability Library (LLM reasoning). -Each signal is pre-filtered to candidate capabilities (keyword ANN stand-in), -then evaluated by the LLM. A deterministic mock path runs when no API key is set. +Signals are pre-filtered to candidate capabilities (keyword ANN stand-in), then +evaluated by the LLM. To stay within provider rate limits, ALL of an account's +signals are evaluated in a single batched LLM call (chunked for safety) rather +than one call per signal. A deterministic mock path runs when no key is set or +the call errors. """ from __future__ import annotations from .. import capabilities, llm from ..types import NormalizedSignal, RelevanceResult -SYSTEM = """You are an account intelligence analyst for an enterprise software company. -Your job: evaluate whether a business signal indicates a genuine cross-sell opportunity -for one of our product capabilities. Be skeptical — only flag signals grounded in a real, -specific business need that a listed capability directly addresses. Return JSON only.""" +# Max signals per batched LLM call (Exa returns ~5/account; this bounds prompt size). +BATCH_SIZE = 8 -USER_TMPL = """ACCOUNT CONTEXT +SYSTEM = """You are an account intelligence analyst for an enterprise software company. +Your job: evaluate whether each business signal indicates a genuine cross-sell +opportunity for one of our product capabilities. Be skeptical — only flag signals +grounded in a real, specific business need that a listed capability directly +addresses. Return JSON only.""" + +BATCH_USER_TMPL = """ACCOUNT CONTEXT Company: {account_name} Current Products: {current_products} Account Tier: {account_tier} -SIGNAL -{content} -Source: {source} | Type: {signal_type} | Date: {published_at} - CANDIDATE CAPABILITIES {capability_summary} -INSTRUCTIONS -1. Determine if this signal indicates a business need one or more candidate capabilities directly addresses. +SIGNALS (evaluate EACH one independently, by its index) +{signals_block} + +For EACH signal: +1. Decide if it indicates a business need one or more candidate capabilities directly addresses. 2. If yes, list the specific capability id(s) and explain WHY (cite the signal). 3. If the account ALREADY has that capability (see Current Products), set "isExpansion": true. 4. Assign signalStrength: strong | moderate | weak. 5. Identify the ideal contact to approach (title + department). -6. If no genuine relevance exists, set "relevant": false. +6. If no genuine relevance exists, set "relevant": false for that signal. -Return ONLY this JSON: -{{"relevant": bool, "capabilityIds": [str], "rationale": str, "signalStrength": "strong|moderate|weak", -"isExpansion": bool, "idealContact": {{"title": str, "department": str}}, "suggestedAngle": str}}""" +Return ONLY this JSON (one object per signal index, all indices included): +{{"signals": [ + {{"index": int, "relevant": bool, "capabilityIds": [str], "rationale": str, + "signalStrength": "strong|moderate|weak", "isExpansion": bool, + "idealContact": {{"title": str, "department": str}}, "suggestedAngle": str}} +]}}""" + + +def _to_result(signal: NormalizedSignal, data: dict) -> RelevanceResult | None: + if not data or not data.get("relevant"): + return None + cap_ids = data.get("capabilityIds", []) or [] + if not cap_ids: + return None + return RelevanceResult( + signal_id=signal.id, + relevant=True, + capability_ids=cap_ids, + rationale=data.get("rationale", ""), + signal_strength=data.get("signalStrength", "weak"), + is_expansion=bool(data.get("isExpansion", False)), + ideal_contact=data.get("idealContact", {}), + suggested_angle=data.get("suggestedAngle", ""), + ) def _mock_evaluate(signal: NormalizedSignal, candidates: list[dict], account: dict) -> dict: """Deterministic stand-in for the LLM: keyword overlap + seniority heuristics.""" if not candidates: - return {"relevant": False, "capabilityIds": [], "rationale": "No candidate capability matched.", - "signalStrength": "weak", "isExpansion": False, - "idealContact": {"title": "", "department": ""}, "suggestedAngle": ""} + return {"relevant": False} top = candidates[0] seniority = (signal.author or {}).get("seniority", "") strength = "strong" if (seniority == "C-suite" or "[" in signal.content[:3]) else "moderate" if signal.source == "jobs" and signal.signal_type == "hiring_surge": strength = "strong" - is_expansion = top["id"] in account.get("currentProducts", []) return { "relevant": True, "capabilityIds": [top["id"]], @@ -59,7 +83,7 @@ def _mock_evaluate(signal: NormalizedSignal, candidates: list[dict], account: di f"author seniority={seniority or 'n/a'}." ), "signalStrength": strength, - "isExpansion": is_expansion, + "isExpansion": top["id"] in account.get("currentProducts", []), "idealContact": { "title": top["targetBuyer"]["titles"][0], "department": top["targetBuyer"]["departments"][0], @@ -68,48 +92,77 @@ def _mock_evaluate(signal: NormalizedSignal, candidates: list[dict], account: di } -def evaluate_signal(signal: NormalizedSignal, account: dict) -> RelevanceResult | None: - candidates = capabilities.prefilter(signal.content) - if not candidates: - return None - - if llm.is_mock(): - data = _mock_evaluate(signal, candidates, account) - else: - user = USER_TMPL.format( - account_name=account["accountName"], - current_products=", ".join(account.get("currentProducts", [])) or "none", - account_tier=account["accountTier"], - content=signal.content, - source=signal.source, - signal_type=signal.signal_type, - published_at=signal.published_at, - capability_summary=capabilities.library_summary_for_prompt(candidates), +def _signals_block(prepared: list[tuple[NormalizedSignal, list[dict]]]) -> str: + lines = [] + for i, (s, cands) in enumerate(prepared): + author = "" + if s.author: + author = f" — {s.author.get('name')} ({s.author.get('title')})" + cand_ids = ", ".join(c["id"] for c in cands) + lines.append( + f"[{i}] {s.source}/{s.signal_type}{author} ({s.published_at[:10]})\n" + f' "{s.content[:600]}"\n' + f" candidate capabilities: {cand_ids}" ) - try: - data = llm.call_json(SYSTEM, user) - except llm.LLMError as e: - print(f" [relevance] LLM error, falling back to mock: {e}") - data = _mock_evaluate(signal, candidates, account) - - if not data.get("relevant"): - return None - return RelevanceResult( - signal_id=signal.id, - relevant=True, - capability_ids=data.get("capabilityIds", []), - rationale=data.get("rationale", ""), - signal_strength=data.get("signalStrength", "weak"), - is_expansion=bool(data.get("isExpansion", False)), - ideal_contact=data.get("idealContact", {}), - suggested_angle=data.get("suggestedAngle", ""), - ) + return "\n".join(lines) -def filter_for_relevance(signals: list[NormalizedSignal], account: dict) -> list[tuple[NormalizedSignal, RelevanceResult]]: +def _mock_batch(prepared, account) -> list[tuple[NormalizedSignal, RelevanceResult]]: out = [] - for s in signals: - res = evaluate_signal(s, account) - if res: - out.append((s, res)) + for s, cands in prepared: + r = _to_result(s, _mock_evaluate(s, cands, account)) + if r: + out.append((s, r)) + return out + + +def _evaluate_batch(prepared, account) -> list[tuple[NormalizedSignal, RelevanceResult]]: + """One LLM call for a chunk of (signal, candidates) pairs.""" + cap_by_id: dict[str, dict] = {} + for _, cands in prepared: + for c in cands: + cap_by_id[c["id"]] = c + + user = BATCH_USER_TMPL.format( + account_name=account["accountName"], + current_products=", ".join(account.get("currentProducts", [])) or "none", + account_tier=account["accountTier"], + capability_summary=capabilities.library_summary_for_prompt(list(cap_by_id.values())), + signals_block=_signals_block(prepared), + ) + try: + data = llm.call_json(SYSTEM, user, max_tokens=2200) + except llm.LLMError as e: + print(f" [relevance] batched LLM error, falling back to mock: {e}") + return _mock_batch(prepared, account) + + by_index = {} + for item in data.get("signals", []): + if isinstance(item, dict) and "index" in item: + by_index[item["index"]] = item + + out = [] + for i, (s, _cands) in enumerate(prepared): + r = _to_result(s, by_index.get(i, {})) + if r: + out.append((s, r)) + return out + + +def filter_for_relevance( + signals: list[NormalizedSignal], account: dict +) -> list[tuple[NormalizedSignal, RelevanceResult]]: + # Pre-filter each signal to candidate capabilities (local, no API cost). + prepared: list[tuple[NormalizedSignal, list[dict]]] = [] + for s in signals: + cands = capabilities.prefilter(s.content) + if cands: + prepared.append((s, cands)) + if not prepared: + return [] + + runner = _mock_batch if llm.is_mock() else _evaluate_batch + out: list[tuple[NormalizedSignal, RelevanceResult]] = [] + for start in range(0, len(prepared), BATCH_SIZE): + out.extend(runner(prepared[start : start + BATCH_SIZE], account)) return out diff --git a/sales-agent/src/agents/synthesizer.py b/sales-agent/src/agents/synthesizer.py index 9036048..1593b9a 100644 --- a/sales-agent/src/agents/synthesizer.py +++ b/sales-agent/src/agents/synthesizer.py @@ -14,26 +14,30 @@ from ..types import ( now_utc, iso, fingerprint, ) -SYSTEM = """You are synthesizing validated account-intelligence signals into a single, -actionable sales opportunity brief. Stay grounded — never over-promise beyond what the -signals say. Lead with the most compelling signal. Return JSON only.""" +SYSTEM = """You are synthesizing validated account-intelligence signals into actionable +sales opportunity briefs — one brief per target capability. Stay grounded — never +over-promise beyond what the signals say. Lead each brief with its most compelling +signal. Return JSON only.""" -USER_TMPL = """ACCOUNT +# One call per account: every capability opportunity is synthesized together. +BATCH_USER_TMPL = """ACCOUNT {account_name} (tier: {account_tier}); current products: {current_products} -TARGET CAPABILITY -{capability_name}: {capability_desc} +Produce ONE brief for EACH target capability below, using only that capability's +validated signals. -VALIDATED SIGNALS -{signals_block} +{capability_blocks} -Create a brief. Return ONLY this JSON: -{{"headline": str (one sentence), -"executiveSummary": str (2-3 sentences for the rep), -"talkingPoints": [str, str, str], -"outreachHook": str (one opener for email/LinkedIn DM), -"urgencyLevel": "high|medium|low", -"confidenceScore": float 0..1}}""" +Return ONLY this JSON (one object per capabilityId): +{{"briefs": [ + {{"capabilityId": str, + "headline": str (one sentence), + "executiveSummary": str (2-3 sentences for the rep), + "talkingPoints": [str, str, str], + "outreachHook": str (one opener for email/LinkedIn DM), + "urgencyLevel": "high|medium|low", + "confidenceScore": float 0..1}} +]}}""" def _signals_block(items: list[tuple[NormalizedSignal, RelevanceResult]]) -> str: @@ -92,6 +96,34 @@ def _mock_brief_body(cap: dict, items, account) -> dict: } +def _batched_bodies(groups: list[tuple[str, dict, list]], account: dict) -> dict[str, dict]: + """One LLM call producing a brief body per capability group. Empty dict on error + so the caller can fall back to the mock body per group.""" + blocks = [] + for cap_id, cap, items in groups: + blocks.append( + f"=== capabilityId: {cap_id} | {cap['name']} ===\n" + f"{cap['description']}\n" + f"VALIDATED SIGNALS:\n{_signals_block(items)}" + ) + user = BATCH_USER_TMPL.format( + account_name=account["accountName"], + account_tier=account["accountTier"], + current_products=", ".join(account.get("currentProducts", [])) or "none", + capability_blocks="\n\n".join(blocks), + ) + try: + data = llm.call_json(SYSTEM, user, max_tokens=4096) + except llm.LLMError as e: + print(f" [synthesize] batched LLM error, falling back to mock: {e}") + return {} + return { + b["capabilityId"]: b + for b in data.get("briefs", []) + if isinstance(b, dict) and b.get("capabilityId") + } + + def synthesize( relevant: list[tuple[NormalizedSignal, RelevanceResult]], account: dict ) -> list[OpportunityBrief]: @@ -101,31 +133,23 @@ def synthesize( for cap_id in r.capability_ids: by_cap.setdefault(cap_id, []).append((s, r)) - briefs: list[OpportunityBrief] = [] + order = {"strong": 0, "moderate": 1, "weak": 2} + groups: list[tuple[str, dict, list]] = [] for cap_id, items in by_cap.items(): cap = capabilities.capability_by_id(cap_id) if not cap: continue - # Strongest signal first. - order = {"strong": 0, "moderate": 1, "weak": 2} - items.sort(key=lambda x: order.get(x[1].signal_strength, 3)) + items.sort(key=lambda x: order.get(x[1].signal_strength, 3)) # strongest first + groups.append((cap_id, cap, items)) + if not groups: + return [] - if llm.is_mock(): - body = _mock_brief_body(cap, items, account) - else: - user = USER_TMPL.format( - account_name=account["accountName"], - account_tier=account["accountTier"], - current_products=", ".join(account.get("currentProducts", [])) or "none", - capability_name=cap["name"], - capability_desc=cap["description"], - signals_block=_signals_block(items), - ) - try: - body = llm.call_json(SYSTEM, user, max_tokens=1200) - except llm.LLMError as e: - print(f" [synthesize] LLM error, falling back to mock: {e}") - body = _mock_brief_body(cap, items, account) + # One LLM call for all of this account's briefs (mock builds them locally). + bodies = {} if llm.is_mock() else _batched_bodies(groups, account) + + briefs: list[OpportunityBrief] = [] + for cap_id, cap, items in groups: + body = bodies.get(cap_id) or _mock_brief_body(cap, items, account) created = now_utc() ideal = items[0][1].ideal_contact diff --git a/sales-agent/src/llm.py b/sales-agent/src/llm.py index 3352885..9fe9905 100644 --- a/sales-agent/src/llm.py +++ b/sales-agent/src/llm.py @@ -8,11 +8,15 @@ from __future__ import annotations import json import re +import time import urllib.request import urllib.error from . import config +# Backoff on HTTP 429 (rate limit) before giving up and letting callers fall to mock. +_RETRY_WAITS = [4, 10, 20] + class LLMError(RuntimeError): pass @@ -29,24 +33,37 @@ def _extract_json(text: str) -> dict: end = text.rfind("}") if start == -1 or end == -1: raise LLMError(f"No JSON object found in LLM output: {text[:200]}") - return json.loads(text[start : end + 1]) + try: + return json.loads(text[start : end + 1]) + except json.JSONDecodeError as e: + # Wrap so callers' `except LLMError` can fall back to mock gracefully + # (e.g. on a truncated/malformed response). + raise LLMError(f"Malformed JSON from LLM ({e}): ...{text[max(0, end-120):end + 1]}") def _post(url: str, payload: dict, headers: dict, provider: str) -> dict: - req = urllib.request.Request( - url, - data=json.dumps(payload).encode("utf-8"), - headers={"content-type": "application/json", **headers}, - method="POST", - ) - try: - with urllib.request.urlopen(req, timeout=60) as resp: - return json.loads(resp.read().decode("utf-8")) - except urllib.error.HTTPError as e: - detail = e.read().decode("utf-8")[:400] if hasattr(e, "read") else "" - raise LLMError(f"{provider} API error {e.code}: {detail}") - except urllib.error.URLError as e: - raise LLMError(f"Network error calling {provider} API: {e}") + data = json.dumps(payload).encode("utf-8") + for attempt in range(len(_RETRY_WAITS) + 1): + req = urllib.request.Request( + url, data=data, + headers={"content-type": "application/json", **headers}, method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=60) as resp: + return json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as e: + detail = e.read().decode("utf-8")[:400] if hasattr(e, "read") else "" + # Rate limited: wait and retry before surfacing the error. + if e.code == 429 and attempt < len(_RETRY_WAITS): + wait = _RETRY_WAITS[attempt] + print(f" [llm] {provider} rate-limited (429); backing off {wait}s " + f"(retry {attempt + 1}/{len(_RETRY_WAITS)})") + time.sleep(wait) + continue + raise LLMError(f"{provider} API error {e.code}: {detail}") + except urllib.error.URLError as e: + raise LLMError(f"Network error calling {provider} API: {e}") + raise LLMError(f"{provider} API still rate-limited after retries") def _call_gemini(system: str, user: str, max_tokens: int) -> dict: @@ -61,6 +78,9 @@ def _call_gemini(system: str, user: str, max_tokens: int) -> dict: "maxOutputTokens": max_tokens, "temperature": 0.2, "responseMimeType": "application/json", + # gemini-2.5-* are thinking models; thinking tokens otherwise consume + # the output budget and truncate the JSON. Disable for structured calls. + "thinkingConfig": {"thinkingBudget": 0}, }, } body = _post(url, payload, {"x-goog-api-key": config.GEMINI_API_KEY}, "Gemini")