Batch LLM calls per account; add 5 pilot accounts

- Relevance filter: evaluate all of an account's signals in one LLM call
  (keyed array, chunked at 8) instead of one call per signal.
- Synthesizer: produce all of an account's briefs in one call (one object
  per capability) instead of one call per capability. ~2 calls/account.
- llm: wrap JSONDecodeError in LLMError so callers fall back to mock instead
  of crashing; disable Gemini thinking (thinkingBudget=0) to stop thinking
  tokens truncating JSON; add 429 backoff.
- accounts: add Booz Allen, Dominion Energy, Hope Gas, Markel, GW Medical
  Faculty Associates (verified domains; CRM tier captured in crmTier).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-23 18:16:08 -04:00
parent 39a338990f
commit b4e3ed9d9b
4 changed files with 270 additions and 113 deletions

View File

@@ -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": [ "accounts": [
{ {
"accountId": "acct_lowes", "accountId": "acct_lowes",
@@ -77,6 +77,66 @@
"currentProducts": [], "currentProducts": [],
"targetContacts": [], "targetContacts": [],
"lastRunAt": null "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
} }
] ]
} }

View File

@@ -1,55 +1,79 @@
"""Phase 3: Relevance filtering against the Capability Library (LLM reasoning). """Phase 3: Relevance filtering against the Capability Library (LLM reasoning).
Each signal is pre-filtered to candidate capabilities (keyword ANN stand-in), Signals are pre-filtered to candidate capabilities (keyword ANN stand-in), then
then evaluated by the LLM. A deterministic mock path runs when no API key is set. 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 __future__ import annotations
from .. import capabilities, llm from .. import capabilities, llm
from ..types import NormalizedSignal, RelevanceResult from ..types import NormalizedSignal, RelevanceResult
SYSTEM = """You are an account intelligence analyst for an enterprise software company. # Max signals per batched LLM call (Exa returns ~5/account; this bounds prompt size).
Your job: evaluate whether a business signal indicates a genuine cross-sell opportunity BATCH_SIZE = 8
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."""
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} Company: {account_name}
Current Products: {current_products} Current Products: {current_products}
Account Tier: {account_tier} Account Tier: {account_tier}
SIGNAL
{content}
Source: {source} | Type: {signal_type} | Date: {published_at}
CANDIDATE CAPABILITIES CANDIDATE CAPABILITIES
{capability_summary} {capability_summary}
INSTRUCTIONS SIGNALS (evaluate EACH one independently, by its index)
1. Determine if this signal indicates a business need one or more candidate capabilities directly addresses. {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). 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. 3. If the account ALREADY has that capability (see Current Products), set "isExpansion": true.
4. Assign signalStrength: strong | moderate | weak. 4. Assign signalStrength: strong | moderate | weak.
5. Identify the ideal contact to approach (title + department). 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: Return ONLY this JSON (one object per signal index, all indices included):
{{"relevant": bool, "capabilityIds": [str], "rationale": str, "signalStrength": "strong|moderate|weak", {{"signals": [
"isExpansion": bool, "idealContact": {{"title": str, "department": str}}, "suggestedAngle": str}}""" {{"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: def _mock_evaluate(signal: NormalizedSignal, candidates: list[dict], account: dict) -> dict:
"""Deterministic stand-in for the LLM: keyword overlap + seniority heuristics.""" """Deterministic stand-in for the LLM: keyword overlap + seniority heuristics."""
if not candidates: if not candidates:
return {"relevant": False, "capabilityIds": [], "rationale": "No candidate capability matched.", return {"relevant": False}
"signalStrength": "weak", "isExpansion": False,
"idealContact": {"title": "", "department": ""}, "suggestedAngle": ""}
top = candidates[0] top = candidates[0]
seniority = (signal.author or {}).get("seniority", "") seniority = (signal.author or {}).get("seniority", "")
strength = "strong" if (seniority == "C-suite" or "[" in signal.content[:3]) else "moderate" strength = "strong" if (seniority == "C-suite" or "[" in signal.content[:3]) else "moderate"
if signal.source == "jobs" and signal.signal_type == "hiring_surge": if signal.source == "jobs" and signal.signal_type == "hiring_surge":
strength = "strong" strength = "strong"
is_expansion = top["id"] in account.get("currentProducts", [])
return { return {
"relevant": True, "relevant": True,
"capabilityIds": [top["id"]], "capabilityIds": [top["id"]],
@@ -59,7 +83,7 @@ def _mock_evaluate(signal: NormalizedSignal, candidates: list[dict], account: di
f"author seniority={seniority or 'n/a'}." f"author seniority={seniority or 'n/a'}."
), ),
"signalStrength": strength, "signalStrength": strength,
"isExpansion": is_expansion, "isExpansion": top["id"] in account.get("currentProducts", []),
"idealContact": { "idealContact": {
"title": top["targetBuyer"]["titles"][0], "title": top["targetBuyer"]["titles"][0],
"department": top["targetBuyer"]["departments"][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: def _signals_block(prepared: list[tuple[NormalizedSignal, list[dict]]]) -> str:
candidates = capabilities.prefilter(signal.content) lines = []
if not candidates: for i, (s, cands) in enumerate(prepared):
return None 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}"
)
return "\n".join(lines)
if llm.is_mock():
data = _mock_evaluate(signal, candidates, account) def _mock_batch(prepared, account) -> list[tuple[NormalizedSignal, RelevanceResult]]:
else: out = []
user = USER_TMPL.format( 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"], account_name=account["accountName"],
current_products=", ".join(account.get("currentProducts", [])) or "none", current_products=", ".join(account.get("currentProducts", [])) or "none",
account_tier=account["accountTier"], account_tier=account["accountTier"],
content=signal.content, capability_summary=capabilities.library_summary_for_prompt(list(cap_by_id.values())),
source=signal.source, signals_block=_signals_block(prepared),
signal_type=signal.signal_type,
published_at=signal.published_at,
capability_summary=capabilities.library_summary_for_prompt(candidates),
) )
try: try:
data = llm.call_json(SYSTEM, user) data = llm.call_json(SYSTEM, user, max_tokens=2200)
except llm.LLMError as e: except llm.LLMError as e:
print(f" [relevance] LLM error, falling back to mock: {e}") print(f" [relevance] batched LLM error, falling back to mock: {e}")
data = _mock_evaluate(signal, candidates, account) return _mock_batch(prepared, account)
if not data.get("relevant"): by_index = {}
return None for item in data.get("signals", []):
return RelevanceResult( if isinstance(item, dict) and "index" in item:
signal_id=signal.id, by_index[item["index"]] = item
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", ""),
)
def filter_for_relevance(signals: list[NormalizedSignal], account: dict) -> list[tuple[NormalizedSignal, RelevanceResult]]:
out = [] out = []
for s in signals: for i, (s, _cands) in enumerate(prepared):
res = evaluate_signal(s, account) r = _to_result(s, by_index.get(i, {}))
if res: if r:
out.append((s, res)) 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 return out

View File

@@ -14,26 +14,30 @@ from ..types import (
now_utc, iso, fingerprint, now_utc, iso, fingerprint,
) )
SYSTEM = """You are synthesizing validated account-intelligence signals into a single, SYSTEM = """You are synthesizing validated account-intelligence signals into actionable
actionable sales opportunity brief. Stay grounded — never over-promise beyond what the sales opportunity briefs — one brief per target capability. Stay grounded — never
signals say. Lead with the most compelling signal. Return JSON only.""" 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} {account_name} (tier: {account_tier}); current products: {current_products}
TARGET CAPABILITY Produce ONE brief for EACH target capability below, using only that capability's
{capability_name}: {capability_desc} validated signals.
VALIDATED SIGNALS {capability_blocks}
{signals_block}
Create a brief. Return ONLY this JSON: Return ONLY this JSON (one object per capabilityId):
{{"headline": str (one sentence), {{"briefs": [
{{"capabilityId": str,
"headline": str (one sentence),
"executiveSummary": str (2-3 sentences for the rep), "executiveSummary": str (2-3 sentences for the rep),
"talkingPoints": [str, str, str], "talkingPoints": [str, str, str],
"outreachHook": str (one opener for email/LinkedIn DM), "outreachHook": str (one opener for email/LinkedIn DM),
"urgencyLevel": "high|medium|low", "urgencyLevel": "high|medium|low",
"confidenceScore": float 0..1}}""" "confidenceScore": float 0..1}}
]}}"""
def _signals_block(items: list[tuple[NormalizedSignal, RelevanceResult]]) -> str: 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( def synthesize(
relevant: list[tuple[NormalizedSignal, RelevanceResult]], account: dict relevant: list[tuple[NormalizedSignal, RelevanceResult]], account: dict
) -> list[OpportunityBrief]: ) -> list[OpportunityBrief]:
@@ -101,31 +133,23 @@ def synthesize(
for cap_id in r.capability_ids: for cap_id in r.capability_ids:
by_cap.setdefault(cap_id, []).append((s, r)) 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(): for cap_id, items in by_cap.items():
cap = capabilities.capability_by_id(cap_id) cap = capabilities.capability_by_id(cap_id)
if not cap: if not cap:
continue continue
# Strongest signal first. items.sort(key=lambda x: order.get(x[1].signal_strength, 3)) # strongest first
order = {"strong": 0, "moderate": 1, "weak": 2} groups.append((cap_id, cap, items))
items.sort(key=lambda x: order.get(x[1].signal_strength, 3)) if not groups:
return []
if llm.is_mock(): # One LLM call for all of this account's briefs (mock builds them locally).
body = _mock_brief_body(cap, items, account) bodies = {} if llm.is_mock() else _batched_bodies(groups, account)
else:
user = USER_TMPL.format( briefs: list[OpportunityBrief] = []
account_name=account["accountName"], for cap_id, cap, items in groups:
account_tier=account["accountTier"], body = bodies.get(cap_id) or _mock_brief_body(cap, items, account)
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)
created = now_utc() created = now_utc()
ideal = items[0][1].ideal_contact ideal = items[0][1].ideal_contact

View File

@@ -8,11 +8,15 @@ from __future__ import annotations
import json import json
import re import re
import time
import urllib.request import urllib.request
import urllib.error import urllib.error
from . import config 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): class LLMError(RuntimeError):
pass pass
@@ -29,24 +33,37 @@ def _extract_json(text: str) -> dict:
end = text.rfind("}") end = text.rfind("}")
if start == -1 or end == -1: if start == -1 or end == -1:
raise LLMError(f"No JSON object found in LLM output: {text[:200]}") raise LLMError(f"No JSON object found in LLM output: {text[:200]}")
try:
return json.loads(text[start : end + 1]) 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: def _post(url: str, payload: dict, headers: dict, provider: str) -> dict:
data = json.dumps(payload).encode("utf-8")
for attempt in range(len(_RETRY_WAITS) + 1):
req = urllib.request.Request( req = urllib.request.Request(
url, url, data=data,
data=json.dumps(payload).encode("utf-8"), headers={"content-type": "application/json", **headers}, method="POST",
headers={"content-type": "application/json", **headers},
method="POST",
) )
try: try:
with urllib.request.urlopen(req, timeout=60) as resp: with urllib.request.urlopen(req, timeout=60) as resp:
return json.loads(resp.read().decode("utf-8")) return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e: except urllib.error.HTTPError as e:
detail = e.read().decode("utf-8")[:400] if hasattr(e, "read") else "" 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}") raise LLMError(f"{provider} API error {e.code}: {detail}")
except urllib.error.URLError as e: except urllib.error.URLError as e:
raise LLMError(f"Network error calling {provider} API: {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: 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, "maxOutputTokens": max_tokens,
"temperature": 0.2, "temperature": 0.2,
"responseMimeType": "application/json", "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") body = _post(url, payload, {"x-goog-api-key": config.GEMINI_API_KEY}, "Gemini")