diff --git a/sales-agent/.env.example b/sales-agent/.env.example index a97ad98..71db156 100644 --- a/sales-agent/.env.example +++ b/sales-agent/.env.example @@ -12,15 +12,17 @@ OPENAI_API_KEY= OPENAI_MODEL=gpt-4o-mini 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: GEMINI_API_KEY= GEMINI_MODEL=gemini-2.5-flash ANTHROPIC_API_KEY= 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) PRIORITY_THRESHOLD=0.72 STANDARD_THRESHOLD=0.45 @@ -36,6 +38,12 @@ EXA_MONTHLY_REQUEST_CAP=1000 EXA_PER_RUN_REQUEST_CAP=25 EXA_RESULTS_PER_QUERY=5 EXA_LOOKBACK_DAYS=45 -# Scope news to each account's own domain (1, default) to cut ticker/analyst -# noise, or search the broad web (0). Override per run: loop --no-domain-filter -EXA_DOMAIN_FILTER=1 +# Exa content category. "news" keeps out careers/marketing landing pages — the +# biggest signal-quality lever. "" disables category filtering. +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= diff --git a/sales-agent/README.md b/sales-agent/README.md index 2b87892..bc76e8a 100644 --- a/sales-agent/README.md +++ b/sales-agent/README.md @@ -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. | | 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`. | -| 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 @@ -75,9 +75,16 @@ LLM_MODEL=claude-sonnet-4-6 ``` 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 -briefs so the full pipeline still runs offline. All providers use plain `urllib` — -no SDKs. LLM calls are batched (one relevance call + one synthesis call per account). +compatible preferred). All providers use plain `urllib` — 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 @@ -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 ``` -### Domain scoping (default on) +### Signal quality: news category & domain scoping -By default, news search is **scoped to each account's own `domain`** (via Exa -`includeDomains`) — newsroom, press releases, investor-relations — which cuts out -stock-ticker and analyst noise from a bare company-name search. Toggle it: +The single biggest lever on signal quality is **`EXA_CATEGORY=news`** (default): it +restricts results to news articles and keeps out evergreen pages (careers, marketing, +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 -python3 run.py loop --no-domain-filter # widen to broad third-party web coverage -# or set EXA_DOMAIN_FILTER=0 in .env to default to broad +python3 run.py loop --domain-filter # account's own domain (newsroom/press only) +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`, -`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 diff --git a/sales-agent/run.py b/sales-agent/run.py index 4ada49e..9563c97 100644 --- a/sales-agent/run.py +++ b/sales-agent/run.py @@ -23,6 +23,8 @@ from src.memory import store, feedback def cmd_loop(args): if args.no_domain_filter: 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) print("\n--- digest preview ---") 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("--tier", choices=["strategic", "enterprise", "growth"]) 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_dig = sub.add_parser("digest", help="Print the daily intelligence digest") diff --git a/sales-agent/src/agents/relevance.py b/sales-agent/src/agents/relevance.py index 55ea46a..3c8073d 100644 --- a/sales-agent/src/agents/relevance.py +++ b/sales-agent/src/agents/relevance.py @@ -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: lines = [] 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) -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]]: - """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] = {} for _, cands in prepared: 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())), signals_block=_signals_block(prepared), ) - try: - 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) + data = llm.call_json(SYSTEM, user, max_tokens=2200, require_keys={"signals"}) by_index = {} for item in data.get("signals", []): @@ -161,8 +122,7 @@ def filter_for_relevance( 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)) + out.extend(_evaluate_batch(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 11aa6bd..6430c5b 100644 --- a/sales-agent/src/agents/synthesizer.py +++ b/sales-agent/src/agents/synthesizer.py @@ -65,40 +65,9 @@ def _target_contacts(account: dict, ideal: dict) -> list[dict]: 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]: - """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.""" + """One LLM call producing a brief body per capability group. Raises on LLM error + (no mock fallback). Capabilities the model omits are simply not synthesized.""" blocks = [] for cap_id, cap, items in groups: 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", capability_blocks="\n\n".join(blocks), ) - try: - 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 {} + data = llm.call_json(SYSTEM, user, max_tokens=4096, require_keys={"briefs"}) return { b["capabilityId"]: b for b in data.get("briefs", []) @@ -144,12 +109,14 @@ def synthesize( if not groups: return [] - # One LLM call for all of this account's briefs (mock builds them locally). - bodies = {} if llm.is_mock() else _batched_bodies(groups, account) + # One LLM call for all of this account's briefs (raises on failure -> caller skips). + bodies = _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) + 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() ideal = items[0][1].ideal_contact diff --git a/sales-agent/src/config.py b/sales-agent/src/config.py index ec29b27..1b25484 100644 --- a/sales-agent/src/config.py +++ b/sales-agent/src/config.py @@ -28,12 +28,13 @@ def _load_dotenv() -> None: _load_dotenv() # --- LLM --- -# Provider abstraction: openai | gemini | anthropic | mock. If LLM_PROVIDER is -# unset we auto-detect from whichever key is present (OpenAI-compatible first), -# falling back to a deterministic mock reasoner so the loop still runs offline. +# Provider abstraction: openai | gemini | anthropic. If LLM_PROVIDER is unset we +# auto-detect from whichever key is present (OpenAI-compatible first). There is NO +# 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, -# 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_API_KEY # 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_PRESENCE_PENALTY = float(os.environ.get("OPENAI_PRESENCE_PENALTY", "0.0")) # 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")) +# 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() LLM_MODEL = os.environ.get("LLM_MODEL", "claude-sonnet-4-6") @@ -68,24 +74,22 @@ if not _provider: elif ANTHROPIC_API_KEY: _provider = "anthropic" else: - _provider = "mock" -# A forced USE_MOCK_LLM=1, or selecting a provider whose key is missing, => mock. -if os.environ.get("USE_MOCK_LLM", "").strip() == "1": - _provider = "mock" -elif _provider == "openai" and not OPENAI_API_KEY: - _provider = "mock" + _provider = "none" +# Selecting a provider whose key is missing => no usable provider. +if _provider == "openai" and not OPENAI_API_KEY: + _provider = "none" elif _provider == "gemini" and not GEMINI_API_KEY: - _provider = "mock" + _provider = "none" elif _provider == "anthropic" and not ANTHROPIC_API_KEY: - _provider = "mock" + _provider = "none" LLM_PROVIDER = _provider -USE_MOCK_LLM = LLM_PROVIDER == "mock" +LLM_CONFIGURED = LLM_PROVIDER != "none" def llm_label() -> str: - if LLM_PROVIDER == "mock": - return "MOCK reasoner (no LLM key)" + if LLM_PROVIDER == "none": + return "NONE (no LLM provider configured)" if LLM_PROVIDER == "openai": host = OPENAI_BASE_URL.split("//")[-1].split("/")[0] 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_RESULTS_PER_QUERY = int(os.environ.get("EXA_RESULTS_PER_QUERY", "5")) 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 -# third-party ticker/analyst noise. Toggle off in .env or per-run with -# `loop --no-domain-filter` to include broad third-party coverage. -EXA_DOMAIN_FILTER = os.environ.get("EXA_DOMAIN_FILTER", "1").strip() == "1" +# Exa content category. "news" restricts results to news articles (press releases, +# coverage) and keeps out evergreen pages like careers/marketing landing pages — +# the single biggest signal-quality lever. Set to "" to disable category filtering. +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 --- SEMANTIC_DEDUP_DAYS = 90 diff --git a/sales-agent/src/connectors/exa.py b/sales-agent/src/connectors/exa.py index 5f9e0c1..c4ab6e8 100644 --- a/sales-agent/src/connectors/exa.py +++ b/sales-agent/src/connectors/exa.py @@ -84,18 +84,18 @@ def _count_request() -> None: 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"] - if domain_scoped: - # Domain is already restricting results, so keep the query topical/broad. - return ( - f"{name} announcements: funding, acquisition, merger, leadership change, " - 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" + events = ( + "acquisition, merger, earnings results, product launch, partnership, " + "expansion, restructuring, layoffs, leadership change, executive appointment, " + "data center, cloud, modernization, regulatory, strategy" ) + if domain_scoped: + return f"{name} news and announcements: {events}" + # No domain restriction -> anchor on the exact company name to limit drift. + return f'"{name}" news: {events}' def harvest_news(account: dict) -> list[RawSignal] | None: @@ -117,8 +117,12 @@ def harvest_news(account: dict) -> list[RawSignal] | None: "startPublishedDate": start, "contents": {"text": {"maxCharacters": 1000}}, } + if config.EXA_CATEGORY: + payload["category"] = config.EXA_CATEGORY if domain_scoped: payload["includeDomains"] = [domain] + if config.EXA_EXCLUDE_DOMAINS: + payload["excludeDomains"] = config.EXA_EXCLUDE_DOMAINS req = urllib.request.Request( _ENDPOINT, data=json.dumps(payload).encode("utf-8"), @@ -162,7 +166,8 @@ def harvest_news(account: dict) -> list[RawSignal] | None: ) ) remaining = usage_status()["monthly_remaining"] + cat = config.EXA_CATEGORY or "any" 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})") return signals diff --git a/sales-agent/src/llm.py b/sales-agent/src/llm.py index fd702ed..7d64510 100644 --- a/sales-agent/src/llm.py +++ b/sales-agent/src/llm.py @@ -1,8 +1,10 @@ """LLM access layer (provider-agnostic). -Dispatches to a configured provider — Gemini or Anthropic — via plain urllib -(no SDKs). When no key is configured it falls back to a deterministic mock -reasoner so the full loop runs offline. All paths return parsed JSON dicts. +Dispatches to a configured provider — OpenAI-compatible, Gemini, or Anthropic — +via plain urllib (no SDKs). There is NO mock fallback: a missing provider raises, +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 @@ -14,14 +16,15 @@ 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 +class LLMNotConfigured(LLMError): + """No LLM provider is configured — the loop must not run.""" + + class LLMRetryable(LLMError): """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]}") +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: 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( url, data=data, 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")) 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)})") + # Rate limited: WAIT (Retry-After if given, else configured interval) + # and retry. No mock fallback — we'd rather grind slowly than degrade. + if e.code == 429 and attempt < config.LLM_RATELIMIT_MAX_RETRIES: + wait = _retry_after_seconds(e) or config.LLM_RATELIMIT_WAIT_SECONDS + 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) 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") + 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: @@ -163,10 +186,13 @@ def call_json( 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 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: - raise _MockSignal() # callers catch this and run their mock path + if not config.LLM_CONFIGURED: + 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) for i in range(attempts): try: @@ -183,11 +209,3 @@ def call_json( f"({i + 1}/{attempts - 1}): {str(e)[:100]}") continue raise - - -class _MockSignal(Exception): - """Internal sentinel: tells the caller to use its deterministic mock branch.""" - - -def is_mock() -> bool: - return config.USE_MOCK_LLM diff --git a/sales-agent/src/pipeline.py b/sales-agent/src/pipeline.py index 9a6cfc3..fe2d07f 100644 --- a/sales-agent/src/pipeline.py +++ b/sales-agent/src/pipeline.py @@ -1,7 +1,7 @@ """The account cycle — the closed loop, run per account (design section 7.3).""" from __future__ import annotations -from . import config +from . import config, llm from .agents import harvester, normalizer, relevance, synthesizer, scorer from .delivery import render from .connectors import crm, exa @@ -27,18 +27,25 @@ def run_account_cycle(account: dict, verbose: bool = True) -> list[dict]: store.save_memory(mem) return [] - # 3. Relevance filter (LLM reasoning vs. capability library) - relevant = relevance.filter_for_relevance(signals, account) - if verbose: - 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: - store.save_memory(mem) + # 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) + if verbose: + print(f" {len(relevant)} relevant signal(s)") + if not relevant: + store.record_signals(mem, signals) # evaluated, found irrelevant -> mark seen + store.save_memory(mem) + return [] + 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 [] - # 4. Synthesize into opportunity briefs - briefs = synthesizer.synthesize(relevant, account) + # Signals fully processed by the LLM -> mark seen so we don't re-evaluate them. + store.record_signals(mem, signals) # 5. Score + prioritize 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]: 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) if not accounts: print("No matching accounts.")