Improve signal quality (Exa news category) and remove mock fallback

Exa harvester:
- EXA_CATEGORY=news (default) restricts to news articles, excluding evergreen
  careers/marketing landing pages that were producing irrelevant signals.
- Event-focused query (acquisition/earnings/launch/leadership), dropped
  hiring/career terms that pulled recruiting pages.
- Domain scoping now OFF by default (broad-web news about the account beats
  first-party careers pages); add EXA_EXCLUDE_DOMAINS and loop --domain-filter.

LLM layer:
- Remove the mock reasoner entirely (no provider => loop refuses to run).
- On HTTP 429, WAIT (LLM_RATELIMIT_WAIT_SECONDS, default 900s; honors
  Retry-After) and retry up to LLM_RATELIMIT_MAX_RETRIES (24) instead of
  degrading to mock output.
- If the LLM is ultimately unavailable for an account, skip it and leave its
  signals unconsumed (retried next run) — never fabricate briefs.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-24 10:41:41 -04:00
parent 2374ef137d
commit 9131e83c08
9 changed files with 169 additions and 173 deletions

View File

@@ -12,15 +12,17 @@ OPENAI_API_KEY=
OPENAI_MODEL=gpt-4o-mini OPENAI_MODEL=gpt-4o-mini
OPENAI_JSON_MODE=1 # set 0 if the endpoint rejects response_format=json_object OPENAI_JSON_MODE=1 # set 0 if the endpoint rejects response_format=json_object
# There is NO mock fallback. On HTTP 429 the loop WAITS this long and retries
# (honors Retry-After if present), so it grinds through tight rate limits.
LLM_RATELIMIT_WAIT_SECONDS=900
LLM_RATELIMIT_MAX_RETRIES=24
# Other providers: # Other providers:
GEMINI_API_KEY= GEMINI_API_KEY=
GEMINI_MODEL=gemini-2.5-flash GEMINI_MODEL=gemini-2.5-flash
ANTHROPIC_API_KEY= ANTHROPIC_API_KEY=
LLM_MODEL=claude-sonnet-4-6 LLM_MODEL=claude-sonnet-4-6
# Force the offline mock reasoner even if a key is set:
# USE_MOCK_LLM=1
# Scoring thresholds (design Phase 5) # Scoring thresholds (design Phase 5)
PRIORITY_THRESHOLD=0.72 PRIORITY_THRESHOLD=0.72
STANDARD_THRESHOLD=0.45 STANDARD_THRESHOLD=0.45
@@ -36,6 +38,12 @@ EXA_MONTHLY_REQUEST_CAP=1000
EXA_PER_RUN_REQUEST_CAP=25 EXA_PER_RUN_REQUEST_CAP=25
EXA_RESULTS_PER_QUERY=5 EXA_RESULTS_PER_QUERY=5
EXA_LOOKBACK_DAYS=45 EXA_LOOKBACK_DAYS=45
# Scope news to each account's own domain (1, default) to cut ticker/analyst # Exa content category. "news" keeps out careers/marketing landing pages — the
# noise, or search the broad web (0). Override per run: loop --no-domain-filter # biggest signal-quality lever. "" disables category filtering.
EXA_DOMAIN_FILTER=1 EXA_CATEGORY=news
# Scope to the account's own domain (1) or broad-web news about it (0, default).
# With category=news, broad web finds actual coverage; domain scoping finds the
# company's own newsroom. Override per run: loop --no-domain-filter / --domain-filter
EXA_DOMAIN_FILTER=0
# Hosts to always exclude (comma-separated), e.g. talent.example.com,careers.example.com
EXA_EXCLUDE_DOMAINS=

View File

@@ -20,7 +20,7 @@ and iterate. Deliberate v1 simplifications (we'll revisit each tomorrow):
| Postgres + Redis + pgvector | **JSON files** under `output/state/` | Hash dedup works; semantic dedup is a documented v2 hook. | | Postgres + Redis + pgvector | **JSON files** under `output/state/` | Hash dedup works; semantic dedup is a documented v2 hook. |
| Vector ANN capability pre-filter | **Keyword-overlap pre-filter** | Same interface, no embedding service. | | Vector ANN capability pre-filter | **Keyword-overlap pre-filter** | Same interface, no embedding service. |
| Slack Bolt + SendGrid | **Console + markdown files** | Briefs render Slack-style; digest writes to `output/digest.md`. | | Slack Bolt + SendGrid | **Console + markdown files** | Briefs render Slack-style; digest writes to `output/digest.md`. |
| LLM relevance + synthesis | **Gemini or Anthropic when a key is set, else a deterministic mock reasoner** | Provider-agnostic (`LLM_PROVIDER`); loop runs offline without a key. | | LLM relevance + synthesis | **OpenAI-compatible / Gemini / Anthropic** (required) | Provider-agnostic (`LLM_PROVIDER`). **No mock fallback** — on a rate limit it waits and retries; without a key it refuses to run. |
## Run it ## Run it
@@ -75,9 +75,16 @@ LLM_MODEL=claude-sonnet-4-6
``` ```
If `LLM_PROVIDER` is blank it auto-detects from whichever key is present (OpenAI- If `LLM_PROVIDER` is blank it auto-detects from whichever key is present (OpenAI-
compatible preferred). With no key, a deterministic mock reasoner produces grounded compatible preferred). All providers use plain `urllib` — no SDKs. LLM calls are
briefs so the full pipeline still runs offline. All providers use plain `urllib` batched (one relevance call + one synthesis call per account).
no SDKs. LLM calls are batched (one relevance call + one synthesis call per account).
**No mock fallback.** Quality is the priority: the loop refuses to run without a
configured provider, and on HTTP 429 it **waits and retries** (`LLM_RATELIMIT_WAIT_
SECONDS`, default 900s; honors `Retry-After`) up to `LLM_RATELIMIT_MAX_RETRIES`
(default 24) rather than emitting low-quality output. If the LLM is ultimately
unavailable for an account, that account is skipped and its signals are left
unconsumed (retried next run) — never fabricated. Degenerate/unparseable responses
(e.g. a model repetition loop) are re-sampled `LLM_CONTENT_RETRIES` times.
## Exa.ai news connector & free-plan protection ## Exa.ai news connector & free-plan protection
@@ -100,23 +107,26 @@ request per loop run (so the default 3 sample accounts = 3 requests/run).
python3 run.py exa-usage # show remaining monthly budget python3 run.py exa-usage # show remaining monthly budget
``` ```
### Domain scoping (default on) ### Signal quality: news category & domain scoping
By default, news search is **scoped to each account's own `domain`** (via Exa The single biggest lever on signal quality is **`EXA_CATEGORY=news`** (default): it
`includeDomains`) — newsroom, press releases, investor-relations — which cuts out restricts results to news articles and keeps out evergreen pages (careers, marketing,
stock-ticker and analyst noise from a bare company-name search. Toggle it: generic corporate). The query is also event-focused (acquisition, earnings, product
launch, leadership change, ...) rather than pulling recruiting pages.
Domain scoping is **off by default** — with `category=news`, broad-web search finds
actual coverage *about* the account (PRNewswire, CNBC, trade press). Scope to the
company's own newsroom only if you want first-party press releases:
```bash ```bash
python3 run.py loop --no-domain-filter # widen to broad third-party web coverage python3 run.py loop --domain-filter # account's own domain (newsroom/press only)
# or set EXA_DOMAIN_FILTER=0 in .env to default to broad python3 run.py loop --no-domain-filter # broad-web news about the account (default)
# or set EXA_DOMAIN_FILTER / EXA_CATEGORY / EXA_EXCLUDE_DOMAINS in .env
``` ```
Note: domain scoping only returns results when the account's real domain has indexed
content. The fictional sample accounts return 0 domain-scoped articles by design —
real target-account domains will return their first-party news.
Tune all limits in `.env` (`EXA_MONTHLY_REQUEST_CAP`, `EXA_PER_RUN_REQUEST_CAP`, Tune all limits in `.env` (`EXA_MONTHLY_REQUEST_CAP`, `EXA_PER_RUN_REQUEST_CAP`,
`EXA_RESULTS_PER_QUERY`, `EXA_LOOKBACK_DAYS`). Set `EXA_ENABLED=0` to force fixtures. `EXA_RESULTS_PER_QUERY`, `EXA_LOOKBACK_DAYS`, `EXA_CATEGORY`). Set `EXA_ENABLED=0` to
force fixtures.
## How the loop maps to the design ## How the loop maps to the design

View File

@@ -23,6 +23,8 @@ from src.memory import store, feedback
def cmd_loop(args): def cmd_loop(args):
if args.no_domain_filter: if args.no_domain_filter:
config.EXA_DOMAIN_FILTER = False # per-run override config.EXA_DOMAIN_FILTER = False # per-run override
if args.domain_filter:
config.EXA_DOMAIN_FILTER = True
pipeline.run_all(tier=args.tier, account_id=args.account) pipeline.run_all(tier=args.tier, account_id=args.account)
print("\n--- digest preview ---") print("\n--- digest preview ---")
print(digest_mod.write_digest()) print(digest_mod.write_digest())
@@ -85,7 +87,9 @@ def main(argv=None):
p_loop.add_argument("--account", help="single account id") p_loop.add_argument("--account", help="single account id")
p_loop.add_argument("--tier", choices=["strategic", "enterprise", "growth"]) p_loop.add_argument("--tier", choices=["strategic", "enterprise", "growth"])
p_loop.add_argument("--no-domain-filter", action="store_true", p_loop.add_argument("--no-domain-filter", action="store_true",
help="include broad third-party news instead of scoping Exa to the account domain") help="broad-web news about the account (default behavior)")
p_loop.add_argument("--domain-filter", action="store_true",
help="scope Exa to the account's own domain (newsroom/press only)")
p_loop.set_defaults(func=cmd_loop) p_loop.set_defaults(func=cmd_loop)
p_dig = sub.add_parser("digest", help="Print the daily intelligence digest") p_dig = sub.add_parser("digest", help="Print the daily intelligence digest")

View File

@@ -65,33 +65,6 @@ def _to_result(signal: NormalizedSignal, data: dict) -> RelevanceResult | None:
) )
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}
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"
return {
"relevant": True,
"capabilityIds": [top["id"]],
"rationale": (
f"Signal references themes central to {top['name']} "
f"(e.g. {', '.join(top['keywords'][:3])}). Source={signal.source}, "
f"author seniority={seniority or 'n/a'}."
),
"signalStrength": strength,
"isExpansion": top["id"] in account.get("currentProducts", []),
"idealContact": {
"title": top["targetBuyer"]["titles"][0],
"department": top["targetBuyer"]["departments"][0],
},
"suggestedAngle": top["businessProblems"][0],
}
def _signals_block(prepared: list[tuple[NormalizedSignal, list[dict]]]) -> str: def _signals_block(prepared: list[tuple[NormalizedSignal, list[dict]]]) -> str:
lines = [] lines = []
for i, (s, cands) in enumerate(prepared): for i, (s, cands) in enumerate(prepared):
@@ -107,17 +80,9 @@ def _signals_block(prepared: list[tuple[NormalizedSignal, list[dict]]]) -> str:
return "\n".join(lines) return "\n".join(lines)
def _mock_batch(prepared, account) -> list[tuple[NormalizedSignal, RelevanceResult]]:
out = []
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]]: def _evaluate_batch(prepared, account) -> list[tuple[NormalizedSignal, RelevanceResult]]:
"""One LLM call for a chunk of (signal, candidates) pairs.""" """One LLM call for a chunk of (signal, candidates) pairs. Raises on LLM error
(no mock fallback) so the pipeline can skip the account rather than fabricate."""
cap_by_id: dict[str, dict] = {} cap_by_id: dict[str, dict] = {}
for _, cands in prepared: for _, cands in prepared:
for c in cands: for c in cands:
@@ -130,11 +95,7 @@ def _evaluate_batch(prepared, account) -> list[tuple[NormalizedSignal, Relevance
capability_summary=capabilities.library_summary_for_prompt(list(cap_by_id.values())), capability_summary=capabilities.library_summary_for_prompt(list(cap_by_id.values())),
signals_block=_signals_block(prepared), signals_block=_signals_block(prepared),
) )
try:
data = llm.call_json(SYSTEM, user, max_tokens=2200, require_keys={"signals"}) data = llm.call_json(SYSTEM, user, max_tokens=2200, require_keys={"signals"})
except llm.LLMError as e:
print(f" [relevance] batched LLM error, falling back to mock: {e}")
return _mock_batch(prepared, account)
by_index = {} by_index = {}
for item in data.get("signals", []): for item in data.get("signals", []):
@@ -161,8 +122,7 @@ def filter_for_relevance(
if not prepared: if not prepared:
return [] return []
runner = _mock_batch if llm.is_mock() else _evaluate_batch
out: list[tuple[NormalizedSignal, RelevanceResult]] = [] out: list[tuple[NormalizedSignal, RelevanceResult]] = []
for start in range(0, len(prepared), BATCH_SIZE): for start in range(0, len(prepared), BATCH_SIZE):
out.extend(runner(prepared[start : start + BATCH_SIZE], account)) out.extend(_evaluate_batch(prepared[start : start + BATCH_SIZE], account))
return out return out

View File

@@ -65,40 +65,9 @@ def _target_contacts(account: dict, ideal: dict) -> list[dict]:
return contacts return contacts
def _mock_brief_body(cap: dict, items, account) -> dict:
strong = [r for _, r in items if r.signal_strength == "strong"]
urgency = "high" if strong else ("medium" if len(items) > 1 else "low")
confidence = min(0.95, 0.45 + 0.18 * len(strong) + 0.08 * len(items))
lead = items[0][0]
tps = [
f"{cap['name']} directly addresses: {cap['businessProblems'][0]}.",
f"Signal evidence: {items[0][1].rationale}",
]
if len(items) > 1:
tps.append(f"Reinforced by {len(items)-1} additional signal(s) across "
f"{', '.join(sorted({s.source for s, _ in items}))}.")
hook_who = (lead.author or {}).get("name", "there")
return {
"headline": f"{account['accountName']}: {len(items)} signal(s) point to {cap['name']}.",
"executiveSummary": (
f"{account['accountName']} is showing intent around {cap['name'].lower()}. "
f"Lead signal: {lead.signal_type} via {lead.source}. "
f"{len(strong)} strong / {len(items)} total contributing signals."
),
"talkingPoints": tps,
"outreachHook": (
f"{hook_who} — saw the recent signal around "
f"{cap['keywords'][0]}. We've helped similar {account.get('industry','')} "
f"teams here. Worth 20 minutes?"
),
"urgencyLevel": urgency,
"confidenceScore": round(confidence, 2),
}
def _batched_bodies(groups: list[tuple[str, dict, list]], account: dict) -> dict[str, 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 """One LLM call producing a brief body per capability group. Raises on LLM error
so the caller can fall back to the mock body per group.""" (no mock fallback). Capabilities the model omits are simply not synthesized."""
blocks = [] blocks = []
for cap_id, cap, items in groups: for cap_id, cap, items in groups:
blocks.append( blocks.append(
@@ -112,11 +81,7 @@ def _batched_bodies(groups: list[tuple[str, dict, list]], account: dict) -> dict
current_products=", ".join(account.get("currentProducts", [])) or "none", current_products=", ".join(account.get("currentProducts", [])) or "none",
capability_blocks="\n\n".join(blocks), capability_blocks="\n\n".join(blocks),
) )
try:
data = llm.call_json(SYSTEM, user, max_tokens=4096, require_keys={"briefs"}) data = llm.call_json(SYSTEM, user, max_tokens=4096, require_keys={"briefs"})
except llm.LLMError as e:
print(f" [synthesize] batched LLM error, falling back to mock: {e}")
return {}
return { return {
b["capabilityId"]: b b["capabilityId"]: b
for b in data.get("briefs", []) for b in data.get("briefs", [])
@@ -144,12 +109,14 @@ def synthesize(
if not groups: if not groups:
return [] return []
# One LLM call for all of this account's briefs (mock builds them locally). # One LLM call for all of this account's briefs (raises on failure -> caller skips).
bodies = {} if llm.is_mock() else _batched_bodies(groups, account) bodies = _batched_bodies(groups, account)
briefs: list[OpportunityBrief] = [] briefs: list[OpportunityBrief] = []
for cap_id, cap, items in groups: for cap_id, cap, items in groups:
body = bodies.get(cap_id) or _mock_brief_body(cap, items, account) body = bodies.get(cap_id)
if not body:
continue # model didn't return a brief for this capability — skip, don't fabricate
created = now_utc() created = now_utc()
ideal = items[0][1].ideal_contact ideal = items[0][1].ideal_contact

View File

@@ -28,12 +28,13 @@ def _load_dotenv() -> None:
_load_dotenv() _load_dotenv()
# --- LLM --- # --- LLM ---
# Provider abstraction: openai | gemini | anthropic | mock. If LLM_PROVIDER is # Provider abstraction: openai | gemini | anthropic. If LLM_PROVIDER is unset we
# unset we auto-detect from whichever key is present (OpenAI-compatible first), # auto-detect from whichever key is present (OpenAI-compatible first). There is NO
# falling back to a deterministic mock reasoner so the loop still runs offline. # mock fallback: the loop refuses to run without a configured provider, and on
# rate limits it WAITS and retries rather than emitting low-quality output.
# #
# "openai" is any OpenAI-compatible /chat/completions endpoint — OpenAI, OpenRouter, # "openai" is any OpenAI-compatible /chat/completions endpoint — OpenAI, OpenRouter,
# Together, Groq, vLLM, Ollama, LM Studio, etc. — configured by three env vars: # Together, Groq, NVIDIA, vLLM, Ollama, LM Studio, etc. — configured by:
# OPENAI_BASE_URL (e.g. https://api.openai.com/v1) # OPENAI_BASE_URL (e.g. https://api.openai.com/v1)
# OPENAI_API_KEY # OPENAI_API_KEY
# OPENAI_MODEL (e.g. gpt-4o-mini) # OPENAI_MODEL (e.g. gpt-4o-mini)
@@ -50,8 +51,13 @@ OPENAI_TEMPERATURE = float(os.environ.get("OPENAI_TEMPERATURE", "0.5"))
OPENAI_FREQUENCY_PENALTY = float(os.environ.get("OPENAI_FREQUENCY_PENALTY", "0.4")) OPENAI_FREQUENCY_PENALTY = float(os.environ.get("OPENAI_FREQUENCY_PENALTY", "0.4"))
OPENAI_PRESENCE_PENALTY = float(os.environ.get("OPENAI_PRESENCE_PENALTY", "0.0")) OPENAI_PRESENCE_PENALTY = float(os.environ.get("OPENAI_PRESENCE_PENALTY", "0.0"))
# Retries when a model returns degenerate/unparseable content (stochastic — a # Retries when a model returns degenerate/unparseable content (stochastic — a
# retry usually succeeds). Distinct from the network/429 backoff. # retry usually succeeds).
LLM_CONTENT_RETRIES = int(os.environ.get("LLM_CONTENT_RETRIES", "3")) LLM_CONTENT_RETRIES = int(os.environ.get("LLM_CONTENT_RETRIES", "3"))
# On HTTP 429 (rate limit) we WAIT this long and retry (no mock fallback). Honors
# a Retry-After response header when present. Lets the loop grind slowly but
# correctly through a tight rate limit instead of degrading quality.
LLM_RATELIMIT_WAIT_SECONDS = int(os.environ.get("LLM_RATELIMIT_WAIT_SECONDS", "900"))
LLM_RATELIMIT_MAX_RETRIES = int(os.environ.get("LLM_RATELIMIT_MAX_RETRIES", "24"))
ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY", "").strip() ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY", "").strip()
LLM_MODEL = os.environ.get("LLM_MODEL", "claude-sonnet-4-6") LLM_MODEL = os.environ.get("LLM_MODEL", "claude-sonnet-4-6")
@@ -68,24 +74,22 @@ if not _provider:
elif ANTHROPIC_API_KEY: elif ANTHROPIC_API_KEY:
_provider = "anthropic" _provider = "anthropic"
else: else:
_provider = "mock" _provider = "none"
# A forced USE_MOCK_LLM=1, or selecting a provider whose key is missing, => mock. # Selecting a provider whose key is missing => no usable provider.
if os.environ.get("USE_MOCK_LLM", "").strip() == "1": if _provider == "openai" and not OPENAI_API_KEY:
_provider = "mock" _provider = "none"
elif _provider == "openai" and not OPENAI_API_KEY:
_provider = "mock"
elif _provider == "gemini" and not GEMINI_API_KEY: elif _provider == "gemini" and not GEMINI_API_KEY:
_provider = "mock" _provider = "none"
elif _provider == "anthropic" and not ANTHROPIC_API_KEY: elif _provider == "anthropic" and not ANTHROPIC_API_KEY:
_provider = "mock" _provider = "none"
LLM_PROVIDER = _provider LLM_PROVIDER = _provider
USE_MOCK_LLM = LLM_PROVIDER == "mock" LLM_CONFIGURED = LLM_PROVIDER != "none"
def llm_label() -> str: def llm_label() -> str:
if LLM_PROVIDER == "mock": if LLM_PROVIDER == "none":
return "MOCK reasoner (no LLM key)" return "NONE (no LLM provider configured)"
if LLM_PROVIDER == "openai": if LLM_PROVIDER == "openai":
host = OPENAI_BASE_URL.split("//")[-1].split("/")[0] host = OPENAI_BASE_URL.split("//")[-1].split("/")[0]
return f"OpenAI-compatible ({OPENAI_MODEL} @ {host})" return f"OpenAI-compatible ({OPENAI_MODEL} @ {host})"
@@ -111,10 +115,18 @@ EXA_MONTHLY_REQUEST_CAP = int(os.environ.get("EXA_MONTHLY_REQUEST_CAP", "1000"))
EXA_PER_RUN_REQUEST_CAP = int(os.environ.get("EXA_PER_RUN_REQUEST_CAP", "25")) EXA_PER_RUN_REQUEST_CAP = int(os.environ.get("EXA_PER_RUN_REQUEST_CAP", "25"))
EXA_RESULTS_PER_QUERY = int(os.environ.get("EXA_RESULTS_PER_QUERY", "5")) EXA_RESULTS_PER_QUERY = int(os.environ.get("EXA_RESULTS_PER_QUERY", "5"))
EXA_LOOKBACK_DAYS = int(os.environ.get("EXA_LOOKBACK_DAYS", "45")) EXA_LOOKBACK_DAYS = int(os.environ.get("EXA_LOOKBACK_DAYS", "45"))
# Scope news search to each account's own domain (newsroom/press/IR) to cut # Exa content category. "news" restricts results to news articles (press releases,
# third-party ticker/analyst noise. Toggle off in .env or per-run with # coverage) and keeps out evergreen pages like careers/marketing landing pages —
# `loop --no-domain-filter` to include broad third-party coverage. # the single biggest signal-quality lever. Set to "" to disable category filtering.
EXA_DOMAIN_FILTER = os.environ.get("EXA_DOMAIN_FILTER", "1").strip() == "1" EXA_CATEGORY = os.environ.get("EXA_CATEGORY", "news").strip()
# Domain scoping is OFF by default now: with category=news, broad-web search finds
# actual news ABOUT the account from outlets, whereas scoping to the company's own
# domain surfaces careers/marketing pages. Turn on for first-party newsroom only.
EXA_DOMAIN_FILTER = os.environ.get("EXA_DOMAIN_FILTER", "0").strip() == "1"
# Hosts to always exclude (comma-separated) — e.g. careers/talent subdomains.
EXA_EXCLUDE_DOMAINS = [
d.strip() for d in os.environ.get("EXA_EXCLUDE_DOMAINS", "").split(",") if d.strip()
]
# --- Dedup / suppression windows --- # --- Dedup / suppression windows ---
SEMANTIC_DEDUP_DAYS = 90 SEMANTIC_DEDUP_DAYS = 90

View File

@@ -84,18 +84,18 @@ def _count_request() -> None:
def _build_query(account: dict, domain_scoped: bool) -> str: def _build_query(account: dict, domain_scoped: bool) -> str:
# Event-focused: target datable business events, NOT evergreen pages. We avoid
# generic words like "hiring"/"careers" that pull recruiting landing pages.
name = account["accountName"] name = account["accountName"]
events = (
"acquisition, merger, earnings results, product launch, partnership, "
"expansion, restructuring, layoffs, leadership change, executive appointment, "
"data center, cloud, modernization, regulatory, strategy"
)
if domain_scoped: if domain_scoped:
# Domain is already restricting results, so keep the query topical/broad. return f"{name} news and announcements: {events}"
return ( # No domain restriction -> anchor on the exact company name to limit drift.
f"{name} announcements: funding, acquisition, merger, leadership change, " return f'"{name}" news: {events}'
f"strategy, restructuring, product launch, hiring, earnings, partnership"
)
# No domain restriction -> anchor harder on the company name to limit drift.
return (
f'"{name}" company news: funding, acquisition, merger, leadership change, '
f"strategy, restructuring, product launch, hiring, earnings, partnership"
)
def harvest_news(account: dict) -> list[RawSignal] | None: def harvest_news(account: dict) -> list[RawSignal] | None:
@@ -117,8 +117,12 @@ def harvest_news(account: dict) -> list[RawSignal] | None:
"startPublishedDate": start, "startPublishedDate": start,
"contents": {"text": {"maxCharacters": 1000}}, "contents": {"text": {"maxCharacters": 1000}},
} }
if config.EXA_CATEGORY:
payload["category"] = config.EXA_CATEGORY
if domain_scoped: if domain_scoped:
payload["includeDomains"] = [domain] payload["includeDomains"] = [domain]
if config.EXA_EXCLUDE_DOMAINS:
payload["excludeDomains"] = config.EXA_EXCLUDE_DOMAINS
req = urllib.request.Request( req = urllib.request.Request(
_ENDPOINT, _ENDPOINT,
data=json.dumps(payload).encode("utf-8"), data=json.dumps(payload).encode("utf-8"),
@@ -162,7 +166,8 @@ def harvest_news(account: dict) -> list[RawSignal] | None:
) )
) )
remaining = usage_status()["monthly_remaining"] remaining = usage_status()["monthly_remaining"]
cat = config.EXA_CATEGORY or "any"
scope = f"domain={domain}" if domain_scoped else "broad web" scope = f"domain={domain}" if domain_scoped else "broad web"
print(f" [exa] {len(signals)} article(s) for {account['accountName']} ({scope}) " print(f" [exa] {len(signals)} {cat} article(s) for {account['accountName']} ({scope}) "
f"(monthly budget left: {remaining}/{config.EXA_MONTHLY_REQUEST_CAP})") f"(monthly budget left: {remaining}/{config.EXA_MONTHLY_REQUEST_CAP})")
return signals return signals

View File

@@ -1,8 +1,10 @@
"""LLM access layer (provider-agnostic). """LLM access layer (provider-agnostic).
Dispatches to a configured provider — Gemini or Anthropic — via plain urllib Dispatches to a configured provider — OpenAI-compatible, Gemini, or Anthropic —
(no SDKs). When no key is configured it falls back to a deterministic mock via plain urllib (no SDKs). There is NO mock fallback: a missing provider raises,
reasoner so the full loop runs offline. All paths return parsed JSON dicts. and on HTTP 429 the call WAITS (honoring Retry-After, else a configured interval)
and retries, so the loop grinds through a rate limit rather than degrading output.
All paths return parsed JSON dicts.
""" """
from __future__ import annotations from __future__ import annotations
@@ -14,14 +16,15 @@ 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
class LLMNotConfigured(LLMError):
"""No LLM provider is configured — the loop must not run."""
class LLMRetryable(LLMError): class LLMRetryable(LLMError):
"""A failure worth retrying with a fresh sample (degenerate/unparseable output).""" """A failure worth retrying with a fresh sample (degenerate/unparseable output)."""
@@ -44,9 +47,23 @@ def _extract_json(text: str) -> dict:
raise LLMRetryable(f"Malformed JSON from LLM ({e}): ...{text[max(0, end-120):end + 1]}") raise LLMRetryable(f"Malformed JSON from LLM ({e}): ...{text[max(0, end-120):end + 1]}")
def _fmt_secs(s: int) -> str:
return f"{s // 60}m{s % 60:02d}s" if s >= 60 else f"{s}s"
def _retry_after_seconds(e: "urllib.error.HTTPError") -> int | None:
"""Parse a Retry-After header (seconds form) if the server provides one."""
try:
val = e.headers.get("Retry-After") if e.headers else None
return int(val) if val and val.strip().isdigit() else None
except Exception:
return None
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") data = json.dumps(payload).encode("utf-8")
for attempt in range(len(_RETRY_WAITS) + 1): # +1 for the initial attempt before any rate-limit wait.
for attempt in range(config.LLM_RATELIMIT_MAX_RETRIES + 1):
req = urllib.request.Request( req = urllib.request.Request(
url, data=data, url, data=data,
headers={"content-type": "application/json", **headers}, method="POST", headers={"content-type": "application/json", **headers}, method="POST",
@@ -56,17 +73,23 @@ def _post(url: str, payload: dict, headers: dict, provider: str) -> dict:
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. # Rate limited: WAIT (Retry-After if given, else configured interval)
if e.code == 429 and attempt < len(_RETRY_WAITS): # and retry. No mock fallback — we'd rather grind slowly than degrade.
wait = _RETRY_WAITS[attempt] if e.code == 429 and attempt < config.LLM_RATELIMIT_MAX_RETRIES:
print(f" [llm] {provider} rate-limited (429); backing off {wait}s " wait = _retry_after_seconds(e) or config.LLM_RATELIMIT_WAIT_SECONDS
f"(retry {attempt + 1}/{len(_RETRY_WAITS)})") from .types import iso, now_utc
from datetime import timedelta
resume = iso(now_utc() + timedelta(seconds=wait))
print(f" [llm] {provider} rate-limited (429). Waiting {_fmt_secs(wait)} "
f"then retrying (attempt {attempt + 1}/{config.LLM_RATELIMIT_MAX_RETRIES}; "
f"resume ~{resume}).")
time.sleep(wait) time.sleep(wait)
continue 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") raise LLMError(f"{provider} API still rate-limited after "
f"{config.LLM_RATELIMIT_MAX_RETRIES} waits")
def _call_openai(system: str, user: str, max_tokens: int) -> dict: def _call_openai(system: str, user: str, max_tokens: int) -> dict:
@@ -163,10 +186,13 @@ def call_json(
network/HTTP errors and 429s are handled in _post. If `require_keys` is given, network/HTTP errors and 429s are handled in _post. If `require_keys` is given,
a response missing all of those top-level keys is treated as degenerate and a response missing all of those top-level keys is treated as degenerate and
retried (catches valid-but-empty junk like `{}` from a repetition loop). retried (catches valid-but-empty junk like `{}` from a repetition loop).
After retries are exhausted the error propagates and callers fall back to mock. After content retries are exhausted the error propagates (no mock fallback).
""" """
if config.USE_MOCK_LLM: if not config.LLM_CONFIGURED:
raise _MockSignal() # callers catch this and run their mock path raise LLMNotConfigured(
"No LLM provider configured. Set OPENAI_API_KEY (and OPENAI_BASE_URL/"
"OPENAI_MODEL), or GEMINI_API_KEY, or ANTHROPIC_API_KEY in .env."
)
attempts = max(1, config.LLM_CONTENT_RETRIES + 1) attempts = max(1, config.LLM_CONTENT_RETRIES + 1)
for i in range(attempts): for i in range(attempts):
try: try:
@@ -183,11 +209,3 @@ def call_json(
f"({i + 1}/{attempts - 1}): {str(e)[:100]}") f"({i + 1}/{attempts - 1}): {str(e)[:100]}")
continue continue
raise raise
class _MockSignal(Exception):
"""Internal sentinel: tells the caller to use its deterministic mock branch."""
def is_mock() -> bool:
return config.USE_MOCK_LLM

View File

@@ -1,7 +1,7 @@
"""The account cycle — the closed loop, run per account (design section 7.3).""" """The account cycle — the closed loop, run per account (design section 7.3)."""
from __future__ import annotations from __future__ import annotations
from . import config from . import config, llm
from .agents import harvester, normalizer, relevance, synthesizer, scorer from .agents import harvester, normalizer, relevance, synthesizer, scorer
from .delivery import render from .delivery import render
from .connectors import crm, exa from .connectors import crm, exa
@@ -27,18 +27,25 @@ def run_account_cycle(account: dict, verbose: bool = True) -> list[dict]:
store.save_memory(mem) store.save_memory(mem)
return [] return []
# 3. Relevance filter (LLM reasoning vs. capability library) # 3 + 4. Relevance filter + synthesis (LLM). On LLM failure we skip the account
# and do NOT consume the signals, so they are retried on the next run (rather
# than emitting low-quality output). 429s are waited out inside the LLM layer.
try:
relevant = relevance.filter_for_relevance(signals, account) relevant = relevance.filter_for_relevance(signals, account)
if verbose: if verbose:
print(f" {len(relevant)} relevant signal(s)") print(f" {len(relevant)} relevant signal(s)")
# Record all processed signals so we don't re-evaluate them next run.
store.record_signals(mem, signals)
if not relevant: if not relevant:
store.record_signals(mem, signals) # evaluated, found irrelevant -> mark seen
store.save_memory(mem) store.save_memory(mem)
return [] return []
# 4. Synthesize into opportunity briefs
briefs = synthesizer.synthesize(relevant, account) briefs = synthesizer.synthesize(relevant, account)
except llm.LLMError as e:
print(f" ! LLM unavailable for {account['accountName']}: {str(e)[:160]}")
print(" skipping account — signals NOT consumed, will retry next run.")
return []
# Signals fully processed by the LLM -> mark seen so we don't re-evaluate them.
store.record_signals(mem, signals)
# 5. Score + prioritize # 5. Score + prioritize
briefs = scorer.score_and_prioritize(briefs, account) briefs = scorer.score_and_prioritize(briefs, account)
@@ -75,6 +82,11 @@ def run_account_cycle(account: dict, verbose: bool = True) -> list[dict]:
def run_all(tier: str | None = None, account_id: str | None = None) -> list[dict]: def run_all(tier: str | None = None, account_id: str | None = None) -> list[dict]:
config.ensure_dirs() config.ensure_dirs()
if not config.LLM_CONFIGURED:
print("ERROR: no LLM provider configured. Set OPENAI_API_KEY (+ OPENAI_BASE_URL/"
"OPENAI_MODEL), GEMINI_API_KEY, or ANTHROPIC_API_KEY in sales-agent/.env.")
print("Refusing to run without a reasoning model (no mock fallback).")
return []
accounts = crm.load_accounts(tier=tier, account_id=account_id) accounts = crm.load_accounts(tier=tier, account_id=account_id)
if not accounts: if not accounts:
print("No matching accounts.") print("No matching accounts.")