"""`/ask` helper: turn a natural-language question into structured filters — deterministically first (countries, industries, event types/subtypes, windows, intents: hiring / pricing / AI / launch / leadership / expansion / legal / developer), then optionally refined by the small model (`prompts/ask-router/v1.md`). The result is an `interpretation` object the API runs against its own tables; nothing here ever fabricates companies, events or numbers — `build_answer()` only phrases counts the API measured. interp = parse_question("companies hiring AI engineers in Canada last month") interp = await route_question(q) # + LLM refinement when configured answer = build_answer(interp, companies=12, events=48) """ from __future__ import annotations import logging import re from dataclasses import asdict, dataclass, field from datetime import UTC, datetime from typing import Any from companyatlas.config import settings from companyatlas.services.llm.schemas import AskRoute from companyatlas.taxonomy import EVENT_SUBTYPES, EventType log = logging.getLogger(__name__) ASK_PARSER_VERSION = "ask-parser-v1" # Compact built-in geography (the API may pass the full `countries` table for better coverage). COUNTRY_TERMS: dict[str, str] = { "canada": "CA", "canadian": "CA", "quebec": "CA", "québec": "CA", "ontario": "CA", "united states": "US", "usa": "US", "u.s.": "US", "us ": "US", "american": "US", "america": "US", "united kingdom": "GB", "uk": "GB", "britain": "GB", "british": "GB", "england": "GB", "france": "FR", "french": "FR", "germany": "DE", "german": "DE", "japan": "JP", "japanese": "JP", "south korea": "KR", "korea": "KR", "korean": "KR", "india": "IN", "indian": "IN", "australia": "AU", "australian": "AU", "brazil": "BR", "brazilian": "BR", "mexico": "MX", "mexican": "MX", "spain": "ES", "spanish": "ES", "italy": "IT", "italian": "IT", "netherlands": "NL", "dutch": "NL", "sweden": "SE", "swedish": "SE", "switzerland": "CH", "swiss": "CH", "singapore": "SG", "china": "CN", "chinese": "CN", "israel": "IL", "israeli": "IL", "ireland": "IE", "irish": "IE", "norway": "NO", "denmark": "DK", "finland": "FI", "poland": "PL", "portugal": "PT", "belgium": "BE", "austria": "AT", "uae": "AE", "emirates": "AE", "saudi": "SA", "south africa": "ZA", "nigeria": "NG", "kenya": "KE", "argentina": "AR", "chile": "CL", "colombia": "CO", "indonesia": "ID", "vietnam": "VN", "thailand": "TH", "taiwan": "TW", "new zealand": "NZ", "turkey": "TR", "türkiye": "TR", } REGION_TERMS = {"europe": "europe", "european": "europe", "asia": "asia", "asian": "asia", "latin america": "latam", "latam": "latam", "africa": "africa", "middle east": "middle-east", "nordic": "nordics", "nordics": "nordics", "apac": "apac", "emea": "emea", "north america": "north-america"} # Intent → (event types, subtypes, tags). Order matters: first match sets the primary intent. INTENTS: list[tuple[str, re.Pattern[str], list[str], list[str], list[str]]] = [ ("ai", re.compile(r"\b(ai|artificial intelligence|machine learning|ml|llm|generative|genai)\b", re.IGNORECASE), [], ["AI_HIRING", "AI_LAUNCH", "TECHNOLOGY_ADOPTION"], ["ai"]), ("hiring", re.compile(r"\b(hiring|hire[sd]?|recruit(ing|ment)?|job[s]?|positions?|openings?|careers?|headcount|talent)\b", re.IGNORECASE), [EventType.HIRING], [], []), ("pricing", re.compile(r"\b(pric(e|es|ing)|plans?|tiers?|cheaper|more expensive|raised prices|cost)\b", re.IGNORECASE), [EventType.PRICING], [], []), ("launch", re.compile(r"\b(launch(es|ed|ing)?|new products?|released?|introduc(ed|es|ing)|unveil(ed|s)?|ship(ped|ping)?)\b", re.IGNORECASE), [EventType.PRODUCT], [], []), ("leadership", re.compile(r"\b(ceo|cfo|cto|coo|executive[s]?|leadership|board|founder[s]?|appoint(ed|s|ment)?|management team)\b", re.IGNORECASE), [EventType.LEADERSHIP], [], []), ("expansion", re.compile(r"\b(expan(d|ded|ding|sion)|new (office|offices|countr(y|ies)|market[s]?|location[s]?)|open(ed|ing)? an? office|enter(ed|ing)?)\b", re.IGNORECASE), [EventType.LOCATION], [], []), ("legal", re.compile(r"\b(terms( of service)?|privacy( policy)?|legal|tos|gdpr|policy changes?)\b", re.IGNORECASE), [EventType.LEGAL], [], []), ("developer", re.compile(r"\b(api[s]?|sdk[s]?|developer[s]?|docs|documentation|changelog|open[- ]source)\b", re.IGNORECASE), [EventType.DEVELOPER], [], []), ("financing", re.compile(r"\b(funding|raise[sd]? (?:\$|€|£|[0-9]|an? )|series [a-f]\b|ipo|investors?|financing|venture capital)\b", re.IGNORECASE), [EventType.FINANCING], [], []), ("m&a", re.compile(r"\b(acqui(re|red|sition|sitions)|merger[s]?|merged|bought|takeover)\b", re.IGNORECASE), [EventType.MA], [], []), ("communication", re.compile(r"\b(news|press releases?|announc(e|ed|ements?)|blog)\b", re.IGNORECASE), [EventType.COMMUNICATION], [], []), ] _WINDOW_RE = re.compile(r"\b(?:in the |over the |during the |within the )?(?:last|past|previous)\s+(\d+)?\s*(day|days|week|weeks|month|months|quarter|quarters|year|years)\b", re.IGNORECASE) _THIS_RE = re.compile(r"\bthis (week|month|quarter|year)\b", re.IGNORECASE) _SINCE_RE = re.compile(r"\bsince (\d{4})(?:-(\d{2}))?(?:-(\d{2}))?\b", re.IGNORECASE) _COUNT_RE = re.compile(r"\b(how many|number of|count of)\b", re.IGNORECASE) _COMPARE_RE = re.compile(r"\b(vs\.?|versus|compare[d]?|comparison)\b", re.IGNORECASE) _TIMELINE_RE = re.compile(r"\b(when did|history of|timeline|over time)\b", re.IGNORECASE) _TREND_RE = re.compile(r"\b(trend(s|ing)?|fastest|most active|which (industries|countries|sectors))\b", re.IGNORECASE) _QUOTED_RE = re.compile(r"[\"“”']([^\"“”']{2,60})[\"“”']") _STOP = {"the", "a", "an", "of", "in", "on", "at", "for", "to", "and", "or", "with", "which", "what", "who", "are", "is", "that", "have", "has", "had", "companies", "company", "show", "me", "list", "find", "all", "any", "about", "from", "by", "their", "recently", "new", "did", "does", "do", "how", "many", "much", "last", "past", "this", "year", "years", "month", "months", "week", "weeks", "day", "days", "since", "were", "was", "been", "being", "into", "than", "more", "most", "less", "some", "there", "where", "when", "why", "get", "give", "tell"} _UNITS = {"day": 1, "days": 1, "week": 7, "weeks": 7, "month": 30, "months": 30, "quarter": 90, "quarters": 90, "year": 365, "years": 365} @dataclass(slots=True) class Interpretation: question: str intent: str = "search" intents: list[str] = field(default_factory=list) countries: list[str] = field(default_factory=list) regions: list[str] = field(default_factory=list) industries: list[str] = field(default_factory=list) event_types: list[str] = field(default_factory=list) event_subtypes: list[str] = field(default_factory=list) tags: list[str] = field(default_factory=list) companies: list[str] = field(default_factory=list) keywords: list[str] = field(default_factory=list) window_days: int | None = None answer_style: str = "list" min_importance: float | None = None confidence: float = 0.5 source: str = "deterministic" parser_version: str = ASK_PARSER_VERSION llm_model: str | None = None prompt_version: str | None = None def to_dict(self) -> dict[str, Any]: return asdict(self) @property def filters(self) -> dict[str, Any]: """Exactly what the API needs to query `events` / `companies`.""" return {k: v for k, v in {"countries": self.countries, "industries": self.industries, "event_types": self.event_types, "event_subtypes": self.event_subtypes, "tags": self.tags, "companies": self.companies, "keywords": self.keywords, "window_days": self.window_days, "min_importance": self.min_importance}.items() if v} def _find_countries(q: str, extra: dict[str, str] | None) -> list[str]: low = f" {q.lower()} " found: list[str] = [] terms = dict(COUNTRY_TERMS) for code, name in (extra or {}).items(): if name: terms[name.lower()] = code.upper() for term, code in sorted(terms.items(), key=lambda kv: -len(kv[0])): pattern = r"(? list[str]: if not industries: return [] low = q.lower() found: list[str] = [] for slug, name in sorted(industries.items(), key=lambda kv: -len(kv[1] or kv[0])): for needle in {slug.replace("-", " "), (name or "").lower()}: if needle and len(needle) >= 3 and re.search(r"(? int | None: m = _WINDOW_RE.search(q) if m: n = int(m.group(1) or 1) return n * _UNITS[m.group(2).lower()] m = _THIS_RE.search(q) if m: return _UNITS[m.group(1).lower()] if re.search(r"\btoday\b", q, re.IGNORECASE): return 1 if re.search(r"\byesterday\b", q, re.IGNORECASE): return 2 m = _SINCE_RE.search(q) if m: start = datetime(int(m.group(1)), int(m.group(2) or 1), int(m.group(3) or 1), tzinfo=UTC) return max(1, (datetime.now(UTC) - start).days) return None def _keywords(q: str, *, drop: set[str]) -> list[str]: words = re.findall(r"[a-zA-Z][a-zA-Z0-9\-\+\.]{1,}", q.lower()) out: list[str] = [] for w in words: w = w.strip(".") if w in _STOP or w in drop or len(w) < 3: continue if w not in out: out.append(w) return out[:8] def parse_question(q: str, *, industries: dict[str, str] | None = None, countries: dict[str, str] | None = None) -> Interpretation: """Deterministic parser. `industries` = {slug: name} and `countries` = {code: name} from the registry (optional).""" q = " ".join((q or "").split())[:400] it = Interpretation(question=q) it.countries = _find_countries(q, countries) it.regions = [v for k, v in REGION_TERMS.items() if re.search(r"(? Interpretation: """Deterministic parse, then an optional LLM refinement that may only tighten filters (never invents results).""" it = parse_question(q, industries=industries, countries=countries) enabled = settings.llm_configured if use_llm is None else (use_llm and settings.llm_configured) if not enabled or not q.strip(): return it try: from companyatlas.db import jsonb from companyatlas.services.llm.gateway import get_provider from companyatlas.services.llm.prompts import load_prompt prompt = load_prompt("ask-router") context = {"question": q, "parsed": it.filters | {"intent": it.intent, "answer_style": it.answer_style}, "allowed_event_types": [str(t) for t in EventType], "allowed_event_subtypes": sorted(EVENT_SUBTYPES), "industries": [{"slug": k, "name": v} for k, v in (industries or {}).items()][:200], "countries": [{"code": k, "name": v} for k, v in (countries or {}).items()][:250]} res = await get_provider().complete_json("small", prompt.system, jsonb(context), AskRoute, max_tokens=400) r: AskRoute = res.data allowed_ind = set(industries or {}) it.countries = list(dict.fromkeys(it.countries + [c for c in r.countries if not countries or c in countries]))[:8] it.industries = list(dict.fromkeys(it.industries + [s for s in r.industries if s in allowed_ind]))[:8] it.event_types = list(dict.fromkeys(it.event_types + [t for t in r.event_types if t in {str(x) for x in EventType}]))[:8] it.event_subtypes = list(dict.fromkeys(it.event_subtypes + r.event_subtypes))[:8] it.companies = list(dict.fromkeys(it.companies + r.companies))[:8] it.keywords = list(dict.fromkeys(r.keywords or it.keywords))[:8] it.window_days = it.window_days or r.window_days if it.intent == "search" and r.intent: it.intent = r.intent if r.answer_style in ("list", "count", "compare", "timeline"): it.answer_style = r.answer_style it.confidence = round(max(it.confidence, min(0.95, r.confidence)), 2) it.source = "llm" it.llm_model = res.model it.prompt_version = prompt.ref except Exception as exc: # noqa: BLE001 — the deterministic parse is always a valid fallback log.warning("ask-router llm refinement skipped", extra={"error": str(exc)[:200]}) return it def build_answer(it: Interpretation, *, companies: int, events: int) -> str: """Phrase measured counts only. The API fills `companies` / `events` from its own query.""" bits: list[str] = [] what = {"hiring": "hiring-related events", "pricing": "pricing events", "ai": "AI-related events", "launch": "product events", "leadership": "leadership events", "expansion": "location events", "legal": "legal page changes", "developer": "developer events", "financing": "financing mentions", "m&a": "M&A mentions", "communication": "communication events"}.get(it.intent, "events") scope = [] if it.countries: scope.append("in " + ", ".join(it.countries)) if it.industries: scope.append("within " + ", ".join(it.industries)) if it.window_days: scope.append(f"over the last {it.window_days} days") scope_txt = (" " + " ".join(scope)) if scope else "" if events == 0 and companies == 0: return f"No monitored evidence matches this question yet{scope_txt}. Results only include events detected on public company pages." bits.append(f"{events} {what}{scope_txt} across {companies} monitored {'company' if companies == 1 else 'companies'}.") bits.append("Every item links to its public source; interpretations are labelled with a confidence level.") return " ".join(bits) __all__ = ["ASK_PARSER_VERSION", "COUNTRY_TERMS", "Interpretation", "build_answer", "parse_question", "route_question"]