Spaces:
Running
Running
| """ | |
| check_grok.py — does grounded discovery actually work, and is it actually grounded? | |
| The whole point of this provider is that the model must LOOK THINGS UP instead of | |
| answering from memory. So this script does not just check that a reply comes | |
| back; it counts tool calls and re-checks the websites the agent returned. | |
| uv run python scripts/check_grok.py | |
| uv run python scripts/check_grok.py --provider groq | |
| uv run python scripts/check_grok.py --list-models | |
| uv run python scripts/check_grok.py --icp kitchens --max 3 | |
| uv run python scripts/check_grok.py --icp "your own free-text ICP here" | |
| Costs real tokens on whichever provider runs. No database is touched. | |
| A run that returns companies but made ZERO tool calls is a FAILURE, not a pass — | |
| that is the old ungrounded behaviour wearing a new coat. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import asyncio | |
| import sys | |
| from pathlib import Path | |
| sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) | |
| import requests # noqa: E402 | |
| from app.services import providers # noqa: E402 | |
| from app.services.providers import tools as provider_tools # noqa: E402 | |
| # Named ICPs to test against. Pass a name (--icp golf) or free text. | |
| # | |
| # "golf" is the sharpest test of grounding: it names a narrow vertical AND a | |
| # checkable property of the website, so a fabricated company is obvious and | |
| # fetch_page can confirm or refute the "static photos, no configurator" claim. | |
| ICP_PRESETS = { | |
| "golf": ( | |
| "US custom golf cart and luxury utility vehicle builders doing $1M-$10M revenue " | |
| "whose websites still show static photo galleries instead of an interactive 3D " | |
| "configurator" | |
| ), | |
| "btier": ( | |
| "US companies that are B Tier companies doing around 1M to 10M in profits and would be " | |
| "interested in Interactive 3D Configurators" | |
| ), | |
| "kitchens": ( | |
| "Luxury bespoke kitchen cabinet manufacturers in the UAE or UK without a 3D configurator" | |
| ), | |
| "yachts": ( | |
| "Superyacht and custom boat builders in Europe selling highly configurable vessels " | |
| "with only static photography on their websites" | |
| ), | |
| } | |
| DEFAULT_ICP = ICP_PRESETS["golf"] | |
| def _wrap_tools_with_counter() -> dict: | |
| """Count tool invocations by wrapping each tool's underlying callable.""" | |
| counts: dict[str, int] = {} | |
| for tool in provider_tools.DISCOVERY_TOOLS: | |
| name = getattr(tool, "name", None) or getattr(tool, "__name__", "tool") | |
| counts[name] = 0 | |
| inner = getattr(tool, "on_invoke_tool", None) | |
| if inner is None: | |
| continue | |
| def make(orig, key): | |
| async def wrapped(*a, **kw): | |
| counts[key] += 1 | |
| return await orig(*a, **kw) | |
| return wrapped | |
| tool.on_invoke_tool = make(inner, name) # type: ignore[attr-defined] | |
| return counts | |
| def _list_models(provider) -> list[str]: | |
| """Ask the provider which models it will actually serve this key.""" | |
| try: | |
| r = requests.get( | |
| f"{provider._base_url.rstrip('/')}/models", | |
| headers={"Authorization": f"Bearer {provider._api_key}"}, | |
| timeout=10, | |
| ) | |
| if r.status_code != 200: | |
| return [] | |
| data = r.json() | |
| rows = data.get("data") if isinstance(data, dict) else data | |
| return sorted(str(m.get("id")) for m in rows if isinstance(m, dict) and m.get("id")) | |
| except Exception: | |
| return [] | |
| def _diagnose(provider, error: str) -> None: | |
| """Turn an opaque provider error into something actionable.""" | |
| low = error.lower() | |
| key = provider._api_key or "" | |
| if "incorrect api key" in low or "invalid_api_key" in low or "401" in low: | |
| print("\n The endpoint rejected the key.") | |
| if key.startswith("gsk_") and "x.ai" in provider._base_url: | |
| print(" That key starts with 'gsk_', which is a GROQ key (groq.com).") | |
| print(" You are calling api.x.ai, which is xAI's Grok — a different company.") | |
| print(" Fix: set LLM_PROVIDER=groq, or put an xai- key in XAI_API_KEY.") | |
| elif key.startswith("xai-") and "groq.com" in provider._base_url: | |
| print(" That key starts with 'xai-', which is an xAI key, but this is Groq.") | |
| print(" Fix: set LLM_PROVIDER=grok.") | |
| else: | |
| print(f" Key prefix is {key[:4]!r}. Check it was copied in full and is still active.") | |
| return | |
| if "model" in low and ("not found" in low or "does not exist" in low or "decommission" in low): | |
| print(f"\n The model {provider._model!r} is not available to this key.") | |
| names = _list_models(provider) | |
| if names: | |
| print(" Models this key CAN use:") | |
| for n in names[:25]: | |
| print(f" {n}") | |
| print(f"\n Set one of those in .env, e.g. GROQ_MODEL=... or XAI_MODEL=...") | |
| else: | |
| print(" Could not list available models either — check the key first.") | |
| return | |
| if "tool" in low and "support" in low: | |
| print(f"\n {provider._model!r} appears not to support tool calling, which this") | |
| print(" agent requires. Pick a tool-capable model.") | |
| # A site that refuses an automated request is NOT a dead site. 403/406/429 and | |
| # friends come from bot protection and say nothing about whether the business is | |
| # real -- the same distinction scraper.looks_blocked() and site_matches_lead() | |
| # already make. Only 404/410 and a failure to connect mean the lead is bad. | |
| BLOCKED_STATUSES = {401, 403, 405, 406, 409, 418, 429, 503} | |
| GONE_STATUSES = {404, 410} | |
| # A bare "Mozilla/5.0" is itself enough to trigger some WAFs. Sending what a real | |
| # browser sends removes most false 406s. | |
| _BROWSER_HEADERS = { | |
| "User-Agent": ( | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " | |
| "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36" | |
| ), | |
| "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", | |
| "Accept-Language": "en-US,en;q=0.9", | |
| } | |
| def _site_is_live(url: str) -> tuple[str, bool]: | |
| """Return (human description, counts_as_dead).""" | |
| if not url: | |
| return "no website returned", True | |
| try: | |
| r = requests.head(url, timeout=8, allow_redirects=True, headers=_BROWSER_HEADERS) | |
| # Some servers reject HEAD specifically; retry properly before judging. | |
| if r.status_code >= 400: | |
| r = requests.get(url, timeout=10, allow_redirects=True, | |
| headers=_BROWSER_HEADERS, stream=True) | |
| code = r.status_code | |
| except requests.RequestException as exc: | |
| return f"unreachable ({type(exc).__name__}) <-- DEAD", True | |
| if code < 400: | |
| return f"HTTP {code}", False | |
| if code in BLOCKED_STATUSES: | |
| return f"HTTP {code} — bot-blocked, site is up (not a mark against the lead)", False | |
| if code in GONE_STATUSES: | |
| return f"HTTP {code} <-- DEAD", True | |
| return f"HTTP {code} <-- DEAD", True | |
| async def main() -> int: | |
| ap = argparse.ArgumentParser(description=__doc__) | |
| ap.add_argument("--provider", default="", help="grok | groq | gemini (default: auto)") | |
| ap.add_argument("--list-models", action="store_true", | |
| help="list the models this key can use, then exit") | |
| ap.add_argument("--icp", default=DEFAULT_ICP, | |
| help="a preset name (" + " | ".join(ICP_PRESETS) + ") or free text") | |
| ap.add_argument("--max", type=int, default=3) | |
| args = ap.parse_args() | |
| icp = ICP_PRESETS.get(args.icp.strip().lower(), args.icp) | |
| status = providers.provider_status() | |
| print("provider credentials:") | |
| for name, ok in status.items(): | |
| print(f" {name:8} {'configured' if ok else 'NOT configured'}") | |
| provider = providers.get_provider(args.provider) | |
| print(f"\nusing: {provider.name} (model {provider._model})") | |
| if not provider.available(): | |
| print(f"\n{provider.name} has no API key. Set XAI_API_KEY (or GEMINI_API_KEY) in .env.") | |
| return 1 | |
| if args.list_models: | |
| names = _list_models(provider) | |
| if not names: | |
| print("\nCould not list models — the key may be wrong for this endpoint.") | |
| return 1 | |
| print(f"\nmodels available to this key ({len(names)}):") | |
| for n in names: | |
| print(f" {n}") | |
| return 0 | |
| counts = _wrap_tools_with_counter() | |
| print(f"\nICP: {icp}\nasking for {args.max} companies...\n") | |
| result = await provider.discover(icp, args.max) | |
| if result.error: | |
| print(f"FAILED: {result.error}") | |
| _diagnose(provider, result.error) | |
| return 1 | |
| print(f"tool calls made: {sum(counts.values())}") | |
| for name, n in counts.items(): | |
| print(f" {name:22} {n}") | |
| print(f"\ncandidates returned: {len(result.candidates)}") | |
| dead = 0 | |
| for c in result.candidates: | |
| print(f"\n {c.company_name} [{c.country or '??'}] {c.industry_tag}") | |
| print(f" website : {c.company_website}") | |
| live, is_dead = _site_is_live(c.company_website) | |
| print(f" live? : {live}") | |
| dead += is_dead | |
| print(f" why fit : {(c.why_fit or '')[:140]}") | |
| print(f" cites : {len(c.citations)} url(s)") | |
| for u in c.citations[:3]: | |
| print(f" {u}") | |
| # ---- the assertions that matter ------------------------------------- | |
| print("\n" + "=" * 60) | |
| failures = [] | |
| if sum(counts.values()) == 0: | |
| failures.append( | |
| "the agent made NO tool calls — it answered from memory, which is exactly " | |
| "the ungrounded behaviour this provider exists to replace" | |
| ) | |
| if not result.candidates: | |
| failures.append("no candidates returned") | |
| if dead: | |
| failures.append( | |
| f"{dead} of {len(result.candidates)} websites are genuinely dead " | |
| f"(404/410/unreachable — bot-blocks do not count)" | |
| ) | |
| uncited = [c.company_name for c in result.candidates if not c.citations] | |
| if uncited: | |
| failures.append(f"no citations for: {', '.join(uncited)}") | |
| # A model must never volunteer contact details -- Candidate has no field for | |
| # them, but check the prose too in case it smuggled one into why_fit. | |
| leaked = [c.company_name for c in result.candidates if "@" in (c.why_fit or "")] | |
| if leaked: | |
| failures.append(f"an email address appeared in why_fit for: {', '.join(leaked)}") | |
| if failures: | |
| print("FAILED:") | |
| for f in failures: | |
| print(f" - {f}") | |
| return 1 | |
| print(f"PASS — {len(result.candidates)} grounded candidates, " | |
| f"{sum(counts.values())} tool calls, all websites live, all cited.") | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(asyncio.run(main())) | |