| 1 | """LLM-first query planning with deterministic guards for risky queries.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import re |
| 7 | import unicodedata |
| 8 | from collections import Counter |
| 9 | |
| 10 | from . import categories, competitors, entity_extract, http, log, providers, query, relevance, schema |
| 11 | |
| 12 | # Hebrew Unicode block: U+0590–U+05FF |
| 13 | _HEBREW_RE = re.compile(r'[\u0590-\u05FF]') |
| 14 | |
| 15 | DISCOVERY_SOURCE_ORDER = ("reddit", "hackernews", "digg", "x") |
| 16 | |
| 17 | |
| 18 | def detect_language(text: str) -> str | None: |
| 19 | """Return 'he' if the text contains Hebrew characters, else None.""" |
| 20 | return 'he' if _HEBREW_RE.search(text) else None |
| 21 | |
| 22 | |
| 23 | def build_discovery_plan( |
| 24 | domain: str, |
| 25 | *, |
| 26 | available_sources: list[str] | None = None, |
| 27 | subreddits: list[str] | None = None, |
| 28 | ) -> schema.DiscoveryPlan: |
| 29 | """Resolve a domain to the existing category-peer community feeds. |
| 30 | |
| 31 | An empty domain is global trending: sweep every river feed's own hot list |
| 32 | (r/all, HN front page, Digg) with no category scoping. Keyword-driven |
| 33 | sources (X, Techmeme, arXiv - none of which expose a river/front-page |
| 34 | lane) sit out of the global nominate stage and join per-topic at the |
| 35 | enrichment pass, where every nomination gets a full research run. |
| 36 | """ |
| 37 | normalized_domain = " ".join(domain.split()) |
| 38 | if not normalized_domain: |
| 39 | resolved = [ |
| 40 | subreddit.removeprefix("r/").strip() |
| 41 | for subreddit in (subreddits or ["all"]) |
| 42 | if subreddit.strip() |
| 43 | ] |
| 44 | allowed = set(DISCOVERY_SOURCE_ORDER if available_sources is None else available_sources) |
| 45 | allowed.discard("x") |
| 46 | sources = [source for source in DISCOVERY_SOURCE_ORDER if source in allowed] |
| 47 | if not sources: |
| 48 | raise ValueError("No listing sources are available for global trending") |
| 49 | return schema.DiscoveryPlan( |
| 50 | domain="", |
| 51 | category=None, |
| 52 | subreddits=resolved or ["all"], |
| 53 | sources=sources, |
| 54 | ) |
| 55 | |
| 56 | category = categories.detect_category(normalized_domain) |
| 57 | candidate_subreddits = list(subreddits or categories.peer_subs_for(category)) |
| 58 | seen_subreddits: set[str] = set() |
| 59 | resolved_subreddits: list[str] = [] |
| 60 | for subreddit in candidate_subreddits: |
| 61 | normalized_subreddit = subreddit.removeprefix("r/").strip() |
| 62 | key = normalized_subreddit.lower() |
| 63 | if not normalized_subreddit or key in seen_subreddits: |
| 64 | continue |
| 65 | seen_subreddits.add(key) |
| 66 | resolved_subreddits.append(normalized_subreddit) |
| 67 | # The curated map intentionally stays small. Keep discovery's keyless floor |
| 68 | # for uncategorized domains by sweeping r/all and applying domain relevance |
| 69 | # during normalization instead of inventing a second category resolver. |
| 70 | if not resolved_subreddits: |
| 71 | resolved_subreddits = ["all"] |
| 72 | |
| 73 | allowed = set(DISCOVERY_SOURCE_ORDER if available_sources is None else available_sources) |
| 74 | sources = [source for source in DISCOVERY_SOURCE_ORDER if source in allowed] |
| 75 | if not sources: |
| 76 | raise ValueError(f"No listing sources are available for {normalized_domain!r}") |
| 77 | |
| 78 | return schema.DiscoveryPlan( |
| 79 | domain=normalized_domain, |
| 80 | category=category, |
| 81 | subreddits=resolved_subreddits, |
| 82 | sources=sources, |
| 83 | ) |
| 84 | |
| 85 | ALLOWED_INTENTS = { |
| 86 | "factual", |
| 87 | "product", |
| 88 | "concept", |
| 89 | "opinion", |
| 90 | "how_to", |
| 91 | "comparison", |
| 92 | "breaking_news", |
| 93 | "prediction", |
| 94 | } |
| 95 | ALLOWED_CLUSTER_MODES = {"none", "story", "workflow", "market", "debate"} |
| 96 | |
| 97 | QUICK_SOURCE_PRIORITY = { |
| 98 | "factual": ["hackernews", "reddit", "x", "xquik", "youtube"], |
| 99 | "product": ["jobs", "youtube", "reddit", "x", "xquik", "tiktok"], |
| 100 | "concept": ["hackernews", "reddit", "x", "xquik", "youtube"], |
| 101 | "opinion": ["reddit", "x", "xquik", "youtube", "hackernews"], |
| 102 | "how_to": ["youtube", "reddit", "x", "xquik", "hackernews"], |
| 103 | "comparison": ["reddit", "x", "xquik", "hackernews", "youtube"], |
| 104 | "breaking_news": ["x", "xquik", "reddit", "hackernews", "youtube", "polymarket"], |
| 105 | "prediction": ["polymarket", "x", "xquik", "hackernews", "reddit", "youtube"], |
| 106 | } |
| 107 | SOURCE_PRIORITY = { |
| 108 | "factual": ["hackernews", "reddit", "x", "youtube"], |
| 109 | "product": ["jobs", "youtube", "reddit", "x", "tiktok", "hackernews"], |
| 110 | "concept": ["hackernews", "reddit", "x", "youtube"], |
| 111 | "opinion": ["reddit", "x", "stocktwits", "dripstack", "youtube", "hackernews"], |
| 112 | "how_to": ["youtube", "reddit", "x", "hackernews"], |
| 113 | "comparison": ["reddit", "x", "hackernews", "youtube"], |
| 114 | "breaking_news": ["x", "stocktwits", "reddit", "hackernews", "youtube", "polymarket"], |
| 115 | "prediction": ["polymarket", "stocktwits", "dripstack", "x", "hackernews", "reddit", "youtube"], |
| 116 | } |
| 117 | SOURCE_LIMITS = { |
| 118 | "quick": { |
| 119 | "factual": 2, |
| 120 | "product": 2, |
| 121 | "concept": 2, |
| 122 | "opinion": 2, |
| 123 | "how_to": 2, |
| 124 | "comparison": 2, |
| 125 | "breaking_news": 2, |
| 126 | "prediction": 2, |
| 127 | }, |
| 128 | # "default" intentionally absent: all available sources are searched |
| 129 | # at default depth. Fusion and reranking handle quality. quick mode |
| 130 | # uses tight budgets above for latency. |
| 131 | } |
| 132 | INTENT_SOURCE_EXCLUSIONS: dict[str, set[str]] = { |
| 133 | "concept": {"polymarket"}, |
| 134 | "how_to": {"polymarket"}, |
| 135 | } |
| 136 | SOURCE_CAPABILITIES = { |
| 137 | "reddit": {"discussion", "social"}, |
| 138 | "x": {"discussion", "social"}, |
| 139 | "xquik": {"discussion", "social"}, |
| 140 | "youtube": {"video", "video_longform", "discussion"}, |
| 141 | "tiktok": {"video", "video_shortform", "social"}, |
| 142 | "instagram": {"video", "video_shortform", "social"}, |
| 143 | "hackernews": {"discussion", "link"}, |
| 144 | "bluesky": {"discussion", "social"}, |
| 145 | "truthsocial": {"discussion", "social"}, |
| 146 | "polymarket": {"market"}, |
| 147 | "stocktwits": {"social", "market", "finance_social"}, |
| 148 | "dripstack": {"reference", "analysis", "link"}, |
| 149 | "digg": {"discussion", "social", "link"}, |
| 150 | "arxiv": {"reference", "analysis", "link"}, |
| 151 | "techmeme": {"discussion", "link", "reference"}, |
| 152 | "trustpilot": {"reference", "company_signal", "social"}, |
| 153 | "amazon": {"reference", "company_signal", "product_signal"}, |
| 154 | "meta_ads": {"reference", "company_signal", "product_signal"}, |
| 155 | "xiaohongshu": {"video", "video_shortform", "social"}, |
| 156 | "telegram": {"discussion", "social"}, |
| 157 | "github": {"discussion", "link"}, |
| 158 | "grounding": {"web", "reference", "link"}, |
| 159 | "perplexity": {"web", "reference", "analysis"}, |
| 160 | "jobs": {"jobs", "company_signal", "link"}, |
| 161 | "corpus": {"reference", "analysis"}, |
| 162 | } |
| 163 | |
| 164 | |
| 165 | def validate_external_plan(raw: dict) -> None: |
| 166 | """Validate explicit-plan structure before permissive sanitization. |
| 167 | |
| 168 | Enum-like metadata stays permissive because direct pipeline callers rely on |
| 169 | the sanitizer to canonicalize those values. |
| 170 | """ |
| 171 | if not isinstance(raw, dict): |
| 172 | raise ValueError("top-level plan must be an object") |
| 173 | for field in ("intent", "freshness_mode", "cluster_mode", "subqueries"): |
| 174 | if field not in raw: |
| 175 | raise ValueError(f"missing required field '{field}'") |
| 176 | for field in ("intent", "freshness_mode", "cluster_mode"): |
| 177 | if not isinstance(raw[field], str) or not raw[field].strip(): |
| 178 | raise ValueError(f"field '{field}' must be a non-empty string") |
| 179 | |
| 180 | source_weights = raw.get("source_weights") |
| 181 | if source_weights is not None and not isinstance(source_weights, dict): |
| 182 | raise ValueError("field 'source_weights' must be an object when provided") |
| 183 | for source, weight in (source_weights or {}).items(): |
| 184 | if ( |
| 185 | not isinstance(source, str) |
| 186 | or not source.strip() |
| 187 | or isinstance(weight, bool) |
| 188 | or not isinstance(weight, (int, float)) |
| 189 | ): |
| 190 | raise ValueError("field 'source_weights' must map source names to numbers") |
| 191 | subqueries = raw["subqueries"] |
| 192 | if not isinstance(subqueries, list) or not subqueries: |
| 193 | raise ValueError("field 'subqueries' must be a non-empty array") |
| 194 | for index, subquery in enumerate(subqueries): |
| 195 | if not isinstance(subquery, dict): |
| 196 | raise ValueError(f"subqueries[{index}] must be an object") |
| 197 | for field in ("search_query", "ranking_query"): |
| 198 | if not isinstance(subquery.get(field), str) or not subquery[field].strip(): |
| 199 | raise ValueError(f"subqueries[{index}].{field} must be a non-empty string") |
| 200 | sources = subquery.get("sources") |
| 201 | if not isinstance(sources, list) or not sources or not all( |
| 202 | isinstance(source, str) and source.strip() for source in sources |
| 203 | ): |
| 204 | raise ValueError(f"subqueries[{index}].sources must be a non-empty string array") |
| 205 | weight = subquery.get("weight") |
| 206 | if weight is not None and ( |
| 207 | isinstance(weight, bool) or not isinstance(weight, (int, float)) |
| 208 | ): |
| 209 | raise ValueError(f"subqueries[{index}].weight must be a number when provided") |
| 210 | |
| 211 | |
| 212 | DEFAULT_INTENT_CAPABILITIES = { |
| 213 | "comparison": {"discussion", "video", "web", "reference", "social", "link", "market"}, |
| 214 | "how_to": {"discussion", "video", "web", "reference", "link"}, |
| 215 | } |
| 216 | |
| 217 | |
| 218 | class DrillTargetError(ValueError): |
| 219 | """Raised when a follow-up target cannot be resolved to a report cluster.""" |
| 220 | |
| 221 | def __init__(self, target: str, clusters: list[schema.Cluster]) -> None: |
| 222 | candidates = ", ".join( |
| 223 | f"{index}. {cluster.title}" |
| 224 | for index, cluster in enumerate(clusters, start=1) |
| 225 | ) or "(no clusters in the cached report)" |
| 226 | super().__init__(f"No cluster matched {target!r}. Available clusters: {candidates}") |
| 227 | |
| 228 | |
| 229 | def _drill_cluster_text(report: schema.Report, cluster: schema.Cluster) -> str: |
| 230 | candidates = {candidate.candidate_id: candidate for candidate in report.ranked_candidates} |
| 231 | parts = [cluster.title] |
| 232 | for candidate_id in cluster.candidate_ids: |
| 233 | candidate = candidates.get(candidate_id) |
| 234 | if candidate: |
| 235 | parts.extend((candidate.title, candidate.snippet)) |
| 236 | return " ".join(part for part in parts if part) |
| 237 | |
| 238 | |
| 239 | def resolve_drill_clusters(report: schema.Report, target: str) -> list[schema.Cluster]: |
| 240 | """Resolve a 1-based cluster index or fuzzy title/entity description.""" |
| 241 | cleaned = target.strip() |
| 242 | numeric = re.fullmatch(r"(?:cluster\s*)?#?(\d+)", cleaned, flags=re.IGNORECASE) |
| 243 | if numeric: |
| 244 | index = int(numeric.group(1)) |
| 245 | if 1 <= index <= len(report.clusters): |
| 246 | return [report.clusters[index - 1]] |
| 247 | raise DrillTargetError(target, report.clusters) |
| 248 | |
| 249 | target_entities = entity_extract.extract_text_entities(cleaned) |
| 250 | scored: list[tuple[float, schema.Cluster]] = [] |
| 251 | for cluster in report.clusters: |
| 252 | cluster_text = _drill_cluster_text(report, cluster) |
| 253 | title_score = relevance.token_overlap_relevance(cleaned, cluster.title) |
| 254 | body_score = relevance.token_overlap_relevance(cleaned, cluster_text) |
| 255 | entity_score = entity_extract.entity_overlap( |
| 256 | target_entities, |
| 257 | entity_extract.extract_text_entities(cluster_text), |
| 258 | ) |
| 259 | score = max(title_score, (0.75 * body_score) + (0.25 * entity_score)) |
| 260 | scored.append((score, cluster)) |
| 261 | |
| 262 | scored.sort(key=lambda entry: entry[0], reverse=True) |
| 263 | if not scored or scored[0][0] < 0.35: |
| 264 | raise DrillTargetError(target, report.clusters) |
| 265 | return [scored[0][1]] |
| 266 | |
| 267 | |
| 268 | def build_drill_plan( |
| 269 | report: schema.Report, |
| 270 | target: str, |
| 271 | *, |
| 272 | clusters: list[schema.Cluster] | None = None, |
| 273 | ) -> schema.QueryPlan: |
| 274 | """Build a deep follow-up plan limited to the matched clusters' sources.""" |
| 275 | matched = clusters or resolve_drill_clusters(report, target) |
| 276 | candidates = {candidate.candidate_id: candidate for candidate in report.ranked_candidates} |
| 277 | |
| 278 | sources: list[str] = [] |
| 279 | for cluster in matched: |
| 280 | for source in cluster.sources: |
| 281 | if source and source not in sources: |
| 282 | sources.append(source) |
| 283 | for candidate_id in cluster.candidate_ids: |
| 284 | candidate = candidates.get(candidate_id) |
| 285 | if not candidate: |
| 286 | continue |
| 287 | for source in schema.candidate_sources(candidate): |
| 288 | if source and source not in sources: |
| 289 | sources.append(source) |
| 290 | if not sources: |
| 291 | raise DrillTargetError(target, report.clusters) |
| 292 | |
| 293 | titles: list[str] = [] |
| 294 | entity_counts: Counter[str] = Counter() |
| 295 | for cluster in matched: |
| 296 | titles.append(cluster.title) |
| 297 | entity_counts.update(entity_extract.extract_text_entities(cluster.title)) |
| 298 | for candidate_id in cluster.representative_ids: |
| 299 | candidate = candidates.get(candidate_id) |
| 300 | if candidate: |
| 301 | titles.append(candidate.title) |
| 302 | entity_counts.update(entity_extract.extract_text_entities(candidate.title)) |
| 303 | |
| 304 | queries: list[str] = [] |
| 305 | for query_text in [ |
| 306 | " ".join(titles[: len(matched)]), |
| 307 | " ".join(entity for entity, _ in entity_counts.most_common(8)), |
| 308 | *titles[len(matched):], |
| 309 | ]: |
| 310 | query_text = " ".join(query_text.split()).strip() |
| 311 | if query_text and query_text.lower() not in {item.lower() for item in queries}: |
| 312 | queries.append(query_text) |
| 313 | if len(queries) == 3: |
| 314 | break |
| 315 | |
| 316 | subqueries = [ |
| 317 | schema.SubQuery( |
| 318 | label=f"drill-{index}", |
| 319 | search_query=search_query, |
| 320 | ranking_query=( |
| 321 | "What deeper evidence, firsthand discussion, comments, and transcripts " |
| 322 | f"explain {search_query}?" |
| 323 | ), |
| 324 | sources=list(sources), |
| 325 | weight=1.0 if index == 1 else 0.85, |
| 326 | ) |
| 327 | for index, search_query in enumerate(queries, start=1) |
| 328 | ] |
| 329 | return schema.QueryPlan( |
| 330 | intent=report.query_plan.intent, |
| 331 | freshness_mode=report.query_plan.freshness_mode, |
| 332 | cluster_mode=report.query_plan.cluster_mode, |
| 333 | raw_topic=report.topic, |
| 334 | subqueries=subqueries, |
| 335 | source_weights={ |
| 336 | source: report.query_plan.source_weights.get(source, 1.0) |
| 337 | for source in sources |
| 338 | }, |
| 339 | notes=[ |
| 340 | "drill-mode", |
| 341 | "drill-targets:" + ",".join(cluster.cluster_id for cluster in matched), |
| 342 | ], |
| 343 | ) |
| 344 | |
| 345 | def plan_query( |
| 346 | *, |
| 347 | topic: str, |
| 348 | available_sources: list[str], |
| 349 | requested_sources: list[str] | None, |
| 350 | depth: str, |
| 351 | provider: providers.ReasoningClient | None, |
| 352 | model: str | None, |
| 353 | context: str = "", |
| 354 | internal_subrun: bool = False, |
| 355 | ) -> schema.QueryPlan: |
| 356 | """Create a query plan. Comparison queries with extractable entities use a |
| 357 | deterministic plan; other intents prefer the configured reasoning provider. |
| 358 | |
| 359 | internal_subrun: when True, suppress the LAW 7 "No --plan passed" stderr |
| 360 | warning. LAW 7 targets the hosting-reasoning-model path; competitor |
| 361 | fan-out sub-runs are engine-internal and the warning is a false positive |
| 362 | there. Default False preserves the warning on every user-facing invocation. |
| 363 | """ |
| 364 | if _should_force_deterministic_plan(topic): |
| 365 | return _fallback_plan( |
| 366 | topic, |
| 367 | available_sources, |
| 368 | requested_sources, |
| 369 | depth, |
| 370 | note="deterministic-comparison-plan", |
| 371 | ) |
| 372 | prompt = _build_prompt(topic, available_sources, requested_sources, depth) |
| 373 | if context: |
| 374 | prompt += f"\n\nCurrent context (from web search): {context}" |
| 375 | if provider and model: |
| 376 | try: |
| 377 | raw = provider.generate_json(model, prompt) |
| 378 | plan = _sanitize_plan( |
| 379 | raw, topic, available_sources, requested_sources, depth, |
| 380 | ) |
| 381 | if plan.subqueries: |
| 382 | return plan |
| 383 | except (ValueError, KeyError, json.JSONDecodeError, OSError, http.HTTPError) as exc: |
| 384 | import sys |
| 385 | print(f"[Planner] LLM planning failed, using deterministic fallback: {type(exc).__name__}: {exc}", file=sys.stderr) |
| 386 | return _fallback_plan( |
| 387 | topic, available_sources, requested_sources, depth, |
| 388 | note=f"fallback-plan (LLM error: {type(exc).__name__})", |
| 389 | ) |
| 390 | # No --plan was passed and no engine-internal provider is configured. |
| 391 | # The deterministic fallback path produces weaker plans than either the |
| 392 | # --plan path or the engine-internal LLM planner path. Surface this so |
| 393 | # the HOSTING reasoning model (Claude Code, Codex, Hermes, Gemini, etc.) |
| 394 | # realizes it is supposed to generate the plan itself and pass --plan. |
| 395 | # The word "provider" in engine messages refers to the engine's INTERNAL |
| 396 | # planner credentials - NOT a prerequisite the caller needs. If you are |
| 397 | # the hosting reasoning model, YOU are the provider. LAW 7 / 2026-04-19 |
| 398 | # Hermes Agent Use Cases failure mode. |
| 399 | if not internal_subrun: |
| 400 | import sys |
| 401 | print( |
| 402 | "[Planner] No --plan passed. If you are the reasoning model hosting " |
| 403 | "this skill (Claude Code, Codex, Hermes, Gemini, or any agent runtime), " |
| 404 | "YOU ARE the planner: generate a JSON query plan yourself and pass it " |
| 405 | "via --plan. You do not need an API key or credentials; you ARE the " |
| 406 | "LLM. The deterministic fallback below is the headless/cron path only. " |
| 407 | "See LAW 7 in SKILL.md and Step 0.75 for the plan schema.", |
| 408 | file=sys.stderr, |
| 409 | ) |
| 410 | return _fallback_plan(topic, available_sources, requested_sources, depth) |
| 411 | |
| 412 | |
| 413 | def _build_prompt( |
| 414 | topic: str, |
| 415 | available_sources: list[str], |
| 416 | requested_sources: list[str] | None, |
| 417 | depth: str, |
| 418 | ) -> str: |
| 419 | requested = ", ".join(requested_sources or ["auto"]) |
| 420 | available = ", ".join(available_sources) |
| 421 | return f""" |
| 422 | You are the query planner for a live last-30-days research pipeline. |
| 423 | |
| 424 | Topic: {topic} |
| 425 | Depth: {depth} |
| 426 | Available sources: {available} |
| 427 | Requested sources: {requested} |
| 428 | |
| 429 | Return JSON only with this shape: |
| 430 | {{ |
| 431 | "intent": "factual|product|concept|opinion|how_to|comparison|breaking_news|prediction", |
| 432 | "freshness_mode": "strict_recent|balanced_recent|evergreen_ok", |
| 433 | "cluster_mode": "none|story|workflow|market|debate", |
| 434 | "source_weights": {{"source_name": 0.0}}, |
| 435 | "subqueries": [ |
| 436 | {{ |
| 437 | "label": "short label", |
| 438 | "search_query": "keyword style query for search APIs", |
| 439 | "ranking_query": "natural language rewrite for reranking", |
| 440 | "sources": ["reddit", "x", "grounding"], |
| 441 | "weight": 1.0 |
| 442 | }} |
| 443 | ], |
| 444 | "notes": ["optional short notes"] |
| 445 | }} |
| 446 | |
| 447 | Rules: |
| 448 | - emit 1 to 5 subqueries (how_to/opinion/product/breaking_news intents benefit from 4-5; factual/concept from 2) |
| 449 | - every subquery must include both search_query and ranking_query |
| 450 | - sources must be drawn from Available sources only |
| 451 | - use cluster_mode=none for factual or many how-to queries |
| 452 | - use strict_recent for breaking news and most predictions |
| 453 | - use debate for comparison/opinion, market for prediction, workflow for how_to, story for breaking_news |
| 454 | - search_query should be concise and keyword-heavy |
| 455 | - ranking_query should read like a natural-language question |
| 456 | - preserve exact proper nouns and entity strings from the topic |
| 457 | - NEVER include temporal phrases in search_query: no 'last 30 days', 'recent', month names, year numbers |
| 458 | - NEVER include meta-research phrases: no 'news', 'updates', 'public appearances', 'latest developments' |
| 459 | - INTENT-MODIFIER HANDLING: when the topic contains one of {{use cases, use case, workflows, workflow, examples, tutorial, tutorials, review, reviews, comparison, applications, in practice, production, production use, how i use}}, STRIP that phrase from every search_query (keep its meaning in ranking_query). Emit 4-5 paraphrased subqueries that each express the intent differently (e.g., 'production', 'workflow OR pipeline', 'review OR experience', 'vs COMPETITOR', 'community discussion'). Broad retrieval, narrow ranking. This was the 2026-04-19 Hermes Agent Use Cases failure mode: the planner echoed "hermes agent use cases" as a literal search string and returned near-zero results because nobody posts that exact phrase. |
| 460 | - DO NOT quote the user's full topic verbatim in search_query. Quote only multi-word proper nouns like "Hermes Agent", "Claude Code", "Nous Research". Bare keywords OR'd together retrieve more than exact-phrase searches. |
| 461 | - search_query should match how content is TITLED on platforms |
| 462 | - GitHub (Issues/PRs) is best for engineering, developer tools, and open source topics: 'kanye west bully' not 'kanye west album news March 2026' |
| 463 | """.strip() |
| 464 | |
| 465 | |
| 466 | def _sanitize_plan( |
| 467 | raw: dict, |
| 468 | topic: str, |
| 469 | available_sources: list[str], |
| 470 | requested_sources: list[str] | None, |
| 471 | depth: str, |
| 472 | *, |
| 473 | honor_plan_sources: bool = False, |
| 474 | ) -> schema.QueryPlan: |
| 475 | intent_hint = str(raw.get("intent") or _infer_intent(topic)).strip() |
| 476 | if intent_hint not in ALLOWED_INTENTS: |
| 477 | intent_hint = _infer_intent(topic) |
| 478 | requested = set(requested_sources or []) |
| 479 | available = set(available_sources) |
| 480 | eligible_sources = [ |
| 481 | source for source in available_sources |
| 482 | if (not requested or source in requested) |
| 483 | ] |
| 484 | source_weights = { |
| 485 | source: float(weight) |
| 486 | for source, weight in (raw.get("source_weights") or {}).items() |
| 487 | if source in available |
| 488 | } |
| 489 | if requested: |
| 490 | source_weights = {source: weight for source, weight in source_weights.items() if source in requested} |
| 491 | if not source_weights: |
| 492 | source_weights = _default_source_weights(_infer_intent(topic), eligible_sources) |
| 493 | # Ensure all eligible sources are available for subqueries. The LLM may |
| 494 | # assign high weights to its preferred sources, but omitted sources still |
| 495 | # participate with base weight so retrieval can overfetch and let fusion |
| 496 | # decide quality. |
| 497 | for source in eligible_sources: |
| 498 | source_weights.setdefault(source, 1.0) |
| 499 | if intent_hint in DEFAULT_INTENT_CAPABILITIES and depth != "quick": |
| 500 | for source in _default_sources_for_intent(intent_hint, eligible_sources): |
| 501 | source_weights.setdefault(source, 1.0) |
| 502 | source_weights = _normalize_weights(source_weights) |
| 503 | |
| 504 | subqueries: list[schema.SubQuery] = [] |
| 505 | for index, subquery in enumerate((raw.get("subqueries") or [])[:_max_subqueries(intent_hint, topic)], start=1): |
| 506 | if not isinstance(subquery, dict): |
| 507 | continue |
| 508 | sources = [source for source in subquery.get("sources") or [] if source in source_weights] |
| 509 | if requested: |
| 510 | sources = [source for source in sources if source in requested] |
| 511 | if not sources: |
| 512 | if honor_plan_sources: |
| 513 | label = str(subquery.get("label") or f"q{index}") |
| 514 | log.source_log( |
| 515 | "Planner", |
| 516 | f"Skipping external-plan subquery {label}: none of its planned " |
| 517 | "sources are available under the current source configuration.", |
| 518 | tty_only=False, |
| 519 | ) |
| 520 | continue |
| 521 | sources = list(source_weights) |
| 522 | search_query = str(subquery.get("search_query") or "").strip() |
| 523 | ranking_query = str(subquery.get("ranking_query") or "").strip() |
| 524 | if not search_query or not ranking_query: |
| 525 | continue |
| 526 | subqueries.append( |
| 527 | schema.SubQuery( |
| 528 | label=str(subquery.get("label") or f"q{index}").strip() or f"q{index}", |
| 529 | search_query=search_query, |
| 530 | ranking_query=ranking_query, |
| 531 | sources=sources, |
| 532 | weight=max(0.05, float(subquery.get("weight") or 1.0)), |
| 533 | ) |
| 534 | ) |
| 535 | if depth == "quick" and subqueries: |
| 536 | subqueries = subqueries[:1] |
| 537 | if not subqueries: |
| 538 | if honor_plan_sources: |
| 539 | raise ValueError( |
| 540 | "No available planned sources remain. Enable a source named in " |
| 541 | "--plan or revise the plan/source configuration; no retrieval was started." |
| 542 | ) |
| 543 | return _fallback_plan(topic, available_sources, requested_sources, depth) |
| 544 | |
| 545 | intent = intent_hint |
| 546 | freshness_mode = str(raw.get("freshness_mode") or _default_freshness(intent)).strip() |
| 547 | if intent == "how_to": |
| 548 | freshness_mode = "evergreen_ok" |
| 549 | cluster_mode = str(raw.get("cluster_mode") or _default_cluster_mode(intent)).strip() |
| 550 | if cluster_mode not in ALLOWED_CLUSTER_MODES: |
| 551 | cluster_mode = _default_cluster_mode(intent) |
| 552 | |
| 553 | return schema.QueryPlan( |
| 554 | intent=intent, |
| 555 | freshness_mode=freshness_mode, |
| 556 | cluster_mode=cluster_mode, |
| 557 | raw_topic=topic, |
| 558 | subqueries=_normalize_subquery_weights( |
| 559 | _trim_subqueries_for_depth( |
| 560 | subqueries, |
| 561 | intent, |
| 562 | depth, |
| 563 | eligible_sources, |
| 564 | requested_sources=requested_sources, |
| 565 | honor_plan_sources=honor_plan_sources, |
| 566 | ) |
| 567 | ), |
| 568 | source_weights=source_weights, |
| 569 | notes=[str(note).strip() for note in raw.get("notes") or [] if str(note).strip()], |
| 570 | ) |
| 571 | |
| 572 | |
| 573 | def _normalize_subquery_weights(subqueries: list[schema.SubQuery]) -> list[schema.SubQuery]: |
| 574 | total = sum(subquery.weight for subquery in subqueries) or 1.0 |
| 575 | return [ |
| 576 | schema.SubQuery( |
| 577 | label=subquery.label, |
| 578 | search_query=subquery.search_query, |
| 579 | ranking_query=subquery.ranking_query, |
| 580 | sources=subquery.sources, |
| 581 | weight=subquery.weight / total, |
| 582 | ) |
| 583 | for subquery in subqueries |
| 584 | ] |
| 585 | |
| 586 | |
| 587 | def _normalize_weights(weights: dict[str, float]) -> dict[str, float]: |
| 588 | total = sum(max(weight, 0.0) for weight in weights.values()) or 1.0 |
| 589 | return { |
| 590 | source: max(weight, 0.0) / total |
| 591 | for source, weight in weights.items() |
| 592 | } |
| 593 | |
| 594 | |
| 595 | def _trim_subqueries_for_depth( |
| 596 | subqueries: list[schema.SubQuery], |
| 597 | intent: str, |
| 598 | depth: str, |
| 599 | available_sources: list[str], |
| 600 | requested_sources: list[str] | None = None, |
| 601 | honor_plan_sources: bool = False, |
| 602 | ) -> list[schema.SubQuery]: |
| 603 | # At non-quick depth, expand sources: use capability routing for intents |
| 604 | # that define it, or all available sources otherwise. The LLM planner may |
| 605 | # assign narrow source lists; we override to let fusion decide quality. |
| 606 | # Operator-supplied --plan is a contract: keep per-subquery sources. |
| 607 | if depth != "quick" and not honor_plan_sources: |
| 608 | expanded_sources = _default_sources_for_intent(intent, available_sources) |
| 609 | return [ |
| 610 | schema.SubQuery( |
| 611 | label=subquery.label, |
| 612 | search_query=subquery.search_query, |
| 613 | ranking_query=subquery.ranking_query, |
| 614 | sources=expanded_sources, |
| 615 | weight=subquery.weight, |
| 616 | ) |
| 617 | for subquery in subqueries |
| 618 | ] |
| 619 | limits = SOURCE_LIMITS.get(depth) |
| 620 | if not limits: |
| 621 | return subqueries |
| 622 | priority_table = QUICK_SOURCE_PRIORITY |
| 623 | priority = priority_table.get(intent, priority_table["breaking_news"]) |
| 624 | limit = limits.get(intent, 3) |
| 625 | ranked_sources = [source for source in priority if source in available_sources] |
| 626 | if not ranked_sources: |
| 627 | ranked_sources = list(available_sources) |
| 628 | trimmed = [] |
| 629 | for subquery in subqueries: |
| 630 | # Quick depth only reaches this block. Honor the plan's explicit |
| 631 | # per-subquery sources: prefer priority-ranked plan sources first, then |
| 632 | # append any plan sources absent from the priority table (e.g. |
| 633 | # instagram). Explicit --search sources are user overrides, so they get |
| 634 | # first claim on the quick slots when present. The final list remains |
| 635 | # capped to the quick-depth limit. |
| 636 | plan_sources = [s for s in ranked_sources if s in subquery.sources] |
| 637 | for source in subquery.sources: |
| 638 | if source not in plan_sources: |
| 639 | plan_sources.append(source) |
| 640 | if not plan_sources: |
| 641 | plan_sources = ranked_sources[:limit] |
| 642 | preferred_sources: list[str] = [] |
| 643 | if requested_sources: |
| 644 | for source in requested_sources: |
| 645 | if ( |
| 646 | source in available_sources |
| 647 | and source in subquery.sources |
| 648 | and source not in preferred_sources |
| 649 | ): |
| 650 | preferred_sources.append(source) |
| 651 | if len(preferred_sources) >= limit: |
| 652 | break |
| 653 | for source in plan_sources: |
| 654 | if len(preferred_sources) >= limit: |
| 655 | break |
| 656 | if source not in preferred_sources: |
| 657 | preferred_sources.append(source) |
| 658 | trimmed.append( |
| 659 | schema.SubQuery( |
| 660 | label=subquery.label, |
| 661 | search_query=subquery.search_query, |
| 662 | ranking_query=subquery.ranking_query, |
| 663 | sources=preferred_sources, |
| 664 | weight=subquery.weight, |
| 665 | ) |
| 666 | ) |
| 667 | return trimmed |
| 668 | |
| 669 | |
| 670 | def _fallback_plan( |
| 671 | topic: str, |
| 672 | available_sources: list[str], |
| 673 | requested_sources: list[str] | None, |
| 674 | depth: str, |
| 675 | note: str = "fallback-plan", |
| 676 | ) -> schema.QueryPlan: |
| 677 | intent = _infer_intent(topic) |
| 678 | # Hebrew-language topics: elevate web search (grounding) to the front of |
| 679 | # the source list since Reddit/HN/GitHub are English-dominant platforms. |
| 680 | # Grounding covers Ynet, Walla, Mako, N12 etc. if a web search key is set. |
| 681 | if detect_language(topic) == 'he' and 'grounding' in available_sources: |
| 682 | ordered = ['grounding'] + [s for s in available_sources if s != 'grounding'] |
| 683 | available_sources = ordered |
| 684 | if requested_sources: |
| 685 | requested_sources = ['grounding'] + [s for s in requested_sources if s != 'grounding'] |
| 686 | allowed_sources = requested_sources or available_sources |
| 687 | source_weights = _default_source_weights(intent, allowed_sources) |
| 688 | core = query.extract_core_subject(topic, max_words=6, strip_suffixes=True) |
| 689 | base_search = _keyword_query(topic, core) |
| 690 | base_ranking = _ranking_query(topic, core) |
| 691 | |
| 692 | subqueries = [schema.SubQuery( |
| 693 | label="primary", |
| 694 | search_query=base_search, |
| 695 | ranking_query=base_ranking, |
| 696 | sources=list(source_weights), |
| 697 | weight=1.0, |
| 698 | )] |
| 699 | |
| 700 | if depth != "quick" and intent == "comparison": |
| 701 | entities = _comparison_entities(topic) |
| 702 | if entities: |
| 703 | for index, entity in enumerate(entities, start=1): |
| 704 | subqueries.append( |
| 705 | schema.SubQuery( |
| 706 | label=f"entity-{index}", |
| 707 | search_query=entity, |
| 708 | ranking_query=f"What recent evidence from the last 30 days is most relevant to {entity} in the comparison '{topic}'?", |
| 709 | sources=list(source_weights), |
| 710 | weight=0.65, |
| 711 | ) |
| 712 | ) |
| 713 | elif depth != "quick" and intent == "prediction": |
| 714 | subqueries.append( |
| 715 | schema.SubQuery( |
| 716 | label="odds", |
| 717 | search_query=f"{base_search} odds forecast", |
| 718 | ranking_query=f"What are the current odds, forecasts, or market signals about {topic}?", |
| 719 | sources=[source for source in source_weights if source in {"polymarket", "grounding", "x", "reddit"}] or list(source_weights), |
| 720 | weight=0.7, |
| 721 | ) |
| 722 | ) |
| 723 | elif depth != "quick" and intent == "breaking_news": |
| 724 | subqueries.append( |
| 725 | schema.SubQuery( |
| 726 | label="reaction", |
| 727 | search_query=f"{base_search} reaction update", |
| 728 | ranking_query=f"What new reactions or follow-up reporting from the last 30 days matter for {topic}?", |
| 729 | sources=[source for source in source_weights if source in {"x", "reddit", "grounding", "hackernews"}] or list(source_weights), |
| 730 | weight=0.7, |
| 731 | ) |
| 732 | ) |
| 733 | |
| 734 | # Intent-modifier fanout: when topic contains a phrase like "use cases", |
| 735 | # "workflows", "examples", "review" (see _INTENT_MODIFIER_PATTERNS), |
| 736 | # paraphrase the intent across 3 extra subqueries rather than echoing |
| 737 | # the literal phrase. Fixes 2026-04-19 Hermes Agent Use Cases failure. |
| 738 | # Excluded for comparison/prediction since those already have dedicated |
| 739 | # fanout (entity-per-subquery / odds). |
| 740 | if depth != "quick" and intent not in {"comparison", "prediction"} and _has_intent_modifier(topic): |
| 741 | subqueries.extend(_intent_modifier_subqueries(topic, core, base_search, source_weights)) |
| 742 | |
| 743 | return schema.QueryPlan( |
| 744 | intent=intent, |
| 745 | freshness_mode=_default_freshness(intent), |
| 746 | cluster_mode=_default_cluster_mode(intent), |
| 747 | raw_topic=topic, |
| 748 | subqueries=_normalize_subquery_weights( |
| 749 | _trim_subqueries_for_depth( |
| 750 | subqueries[:_max_subqueries(intent, topic)], |
| 751 | intent, |
| 752 | depth, |
| 753 | list(source_weights), |
| 754 | requested_sources=requested_sources, |
| 755 | ) |
| 756 | ), |
| 757 | source_weights=_normalize_weights(source_weights), |
| 758 | notes=[note], |
| 759 | ) |
| 760 | |
| 761 | |
| 762 | def _infer_intent(topic: str) -> str: |
| 763 | text = topic.lower().strip() |
| 764 | if re.search(r"\b(vs|versus|compare|compared to|difference between)\b", text): |
| 765 | return "comparison" |
| 766 | # Slash-separated proper nouns: "React/Vue/Svelte" (not URLs, not acronyms like CI/CD or I/O) |
| 767 | if not re.search(r"https?://", topic) and re.search(r"\b[A-Z][a-z]{2,}(?:/[A-Z][a-z]{2,})+\b", topic): |
| 768 | return "comparison" |
| 769 | if re.search(r"\b(odds|predict|prediction|forecast|chance|probability|will .* win)\b", text): |
| 770 | return "prediction" |
| 771 | if re.search(r"\b(how to|tutorial|guide|setup|step by step|deploy|install)\b", text): |
| 772 | return "how_to" |
| 773 | if re.search(r"\b(what is|what are|who is|who acquired|when did|parameter count|release date)\b", text): |
| 774 | return "factual" |
| 775 | if re.search(r"\b(thoughts on|worth it|should i|opinion|review)\b", text): |
| 776 | return "opinion" |
| 777 | if re.search(r"\b(latest|news|announced|just shipped|launched|released|update)\b", text): |
| 778 | return "breaking_news" |
| 779 | if re.search(r"\b(pricing|feature|features|best .* for|top .* for)\b", text): |
| 780 | return "product" |
| 781 | if re.search(r"\b(explain|concept|protocol|architecture|what does)\b", text): |
| 782 | return "concept" |
| 783 | if re.search(r"\b(tournament|championship|playoffs|march madness|world cup|olympics|super bowl|final four|ceremony|awards|keynote)\b", text): |
| 784 | return "breaking_news" |
| 785 | # Recency signals take priority when nothing more specific matched. |
| 786 | if re.search(r"\b(trending|this week|right now|today|this month)\b", text): |
| 787 | return "breaking_news" |
| 788 | # Default changed from "breaking_news" to "concept" on 2026-04-19 after |
| 789 | # the Hermes Agent Use Cases failure: unclassified topics were getting |
| 790 | # strict_recent freshness, which over-weighted the last 7 days and |
| 791 | # under-weighted older relevant material. "concept" defaults to |
| 792 | # evergreen_ok freshness, a safer posture for unknown topics. |
| 793 | return "concept" |
| 794 | |
| 795 | |
| 796 | def _default_freshness(intent: str) -> str: |
| 797 | if intent in {"breaking_news", "prediction"}: |
| 798 | return "strict_recent" |
| 799 | if intent in {"concept", "how_to"}: |
| 800 | return "evergreen_ok" |
| 801 | return "balanced_recent" |
| 802 | |
| 803 | |
| 804 | def _default_cluster_mode(intent: str) -> str: |
| 805 | return { |
| 806 | "breaking_news": "story", |
| 807 | "comparison": "debate", |
| 808 | "opinion": "debate", |
| 809 | "prediction": "market", |
| 810 | "how_to": "workflow", |
| 811 | "factual": "none", |
| 812 | "product": "none", |
| 813 | "concept": "none", |
| 814 | }.get(intent, "none") |
| 815 | |
| 816 | |
| 817 | def _default_source_weights(intent: str, sources: list[str]) -> dict[str, float]: |
| 818 | base = {source: 1.0 for source in sources} |
| 819 | if intent == "prediction": |
| 820 | for source, bonus in {"polymarket": 2.5, "x": 1.3}.items(): |
| 821 | if source in base: |
| 822 | base[source] += bonus |
| 823 | elif intent == "breaking_news": |
| 824 | for source, bonus in {"x": 1.5, "reddit": 1.3, "hackernews": 0.8}.items(): |
| 825 | if source in base: |
| 826 | base[source] += bonus |
| 827 | elif intent == "how_to": |
| 828 | for source, bonus in {"youtube": 2.0, "hackernews": 0.8}.items(): |
| 829 | if source in base: |
| 830 | base[source] += bonus |
| 831 | elif intent == "factual": |
| 832 | for source, bonus in {"reddit": 0.8, "x": 0.5}.items(): |
| 833 | if source in base: |
| 834 | base[source] += bonus |
| 835 | elif intent == "product": |
| 836 | for source, bonus in {"jobs": 0.8, "youtube": 0.5}.items(): |
| 837 | if source in base: |
| 838 | base[source] += bonus |
| 839 | return base |
| 840 | |
| 841 | |
| 842 | def _keyword_query(topic: str, core: str) -> str: |
| 843 | """Build a search_query string for the deterministic fallback. |
| 844 | |
| 845 | Quote ONLY title-cased multi-word proper nouns ("Hermes Agent", |
| 846 | "Claude Code", "Nous Research") so platform search engines preserve the |
| 847 | name as a phrase. Hyphenated compounds and lowercase terms are left as |
| 848 | bare keywords, which broadens retrieval instead of narrowing it. |
| 849 | |
| 850 | Prior behavior quoted the entire compound including the user's typed |
| 851 | topic, producing searches like `"Hermes Agent Actual Use Cases" hermes agent actual` |
| 852 | that returned near-zero matches on X and Reddit because nobody posts |
| 853 | that exact phrase. See 2026-04-19 Hermes Agent Use Cases failure. |
| 854 | """ |
| 855 | compounds = query.extract_compound_terms(topic) |
| 856 | # Only quote title-cased proper nouns (multi-word names). Hyphenated |
| 857 | # compounds go unquoted so platform tokenizers can split and match. |
| 858 | title_cased = [ |
| 859 | term for term in compounds |
| 860 | if re.match(r"^(?:[A-Z][a-z]+\s+){1,}[A-Z][a-z]+$", term) |
| 861 | ] |
| 862 | selected = title_cased[:2] |
| 863 | quoted = " ".join(f'"{term}"' for term in selected) |
| 864 | remainder = core.strip() or topic.strip() |
| 865 | # Drop words already carried by a quoted phrase. Emitting both produced |
| 866 | # '"Peter Steinberger" peter steinberger steipete', which reads to a |
| 867 | # provider as the phrase AND each of its words again -- strictly narrower |
| 868 | # than the phrase alone, and on X it degraded to a bare token conjunction |
| 869 | # once the quotes were stripped downstream. Distinct tokens (here |
| 870 | # "steipete") are preserved. |
| 871 | if selected and remainder: |
| 872 | phrase_words = { |
| 873 | word.lower() |
| 874 | for term in selected |
| 875 | for word in term.split() |
| 876 | } |
| 877 | remainder = " ".join( |
| 878 | word for word in remainder.split() |
| 879 | if word.strip('"').lower() not in phrase_words |
| 880 | ) |
| 881 | keywords = [quoted.strip(), remainder.strip()] |
| 882 | return " ".join(part for part in keywords if part).strip() |
| 883 | |
| 884 | |
| 885 | def _ranking_query(topic: str, core: str) -> str: |
| 886 | if topic.strip().endswith("?"): |
| 887 | return topic.strip() |
| 888 | if core and core.lower() != topic.lower(): |
| 889 | return f"What recent evidence from the last 30 days is most relevant to {topic}, especially about {core}?" |
| 890 | return f"What recent evidence from the last 30 days is most relevant to {topic}?" |
| 891 | |
| 892 | |
| 893 | _TRAILING_CONTEXT = re.compile( |
| 894 | r"\s+\b(?:for|in|on|at|to|with|about|from|by|during|since|after|before|using|via)\b.*$", |
| 895 | re.I, |
| 896 | ) |
| 897 | |
| 898 | |
| 899 | def _comparison_entities(topic: str, *, uncapped: bool = False) -> list[str]: |
| 900 | """Split a comparison topic into entity names. |
| 901 | |
| 902 | Caps at ``competitors.COMPARISON_ENTITY_MAX`` unless ``uncapped`` (caller |
| 903 | truncates and may warn about dropped entities). |
| 904 | """ |
| 905 | # "difference between X and Y" -> "X vs Y" (replace "and" only in this context) |
| 906 | normalized = re.sub( |
| 907 | r"\bdifference between\s+(.+?)\s+and\s+", |
| 908 | r"\1 vs ", |
| 909 | topic, |
| 910 | flags=re.I, |
| 911 | ) |
| 912 | normalized = re.sub(r"\b(compared to)\b", " vs ", normalized, flags=re.I) |
| 913 | parts = [ |
| 914 | part.strip(" \t\r\n?.,:;!()[]{}\"'") |
| 915 | for part in re.split(r"\bvs\.?\b|\bversus\b|/", normalized, flags=re.I) |
| 916 | if part.strip(" \t\r\n?.,:;!()[]{}\"'") |
| 917 | ] |
| 918 | # Strip trailing context from parts ("Svelte for frontend in 2026" -> "Svelte") |
| 919 | if len(parts) < 2: |
| 920 | return [] |
| 921 | parts = [_TRAILING_CONTEXT.sub("", part).strip() or part for part in parts] |
| 922 | deduped: list[str] = [] |
| 923 | for part in parts: |
| 924 | if part and part not in deduped: |
| 925 | deduped.append(part) |
| 926 | if uncapped: |
| 927 | return deduped |
| 928 | return deduped[: competitors.COMPARISON_ENTITY_MAX] |
| 929 | |
| 930 | |
| 931 | def _should_force_deterministic_plan(topic: str) -> bool: |
| 932 | return _infer_intent(topic) == "comparison" and len(_comparison_entities(topic)) >= 2 |
| 933 | |
| 934 | |
| 935 | _INTENT_MODIFIER_PATTERNS = ( |
| 936 | "use cases", "use case", "workflows", "workflow", |
| 937 | "examples", "example", "tutorial", "tutorials", |
| 938 | "review", "reviews", "comparison", "applications", |
| 939 | "in practice", "production use", "production", |
| 940 | "how i use", |
| 941 | ) |
| 942 | |
| 943 | |
| 944 | def _has_intent_modifier(topic: str) -> bool: |
| 945 | """Return True if the topic contains an intent modifier phrase. |
| 946 | |
| 947 | See 2026-04-19 Hermes Agent Use Cases failure: a literal "Hermes Agent |
| 948 | use cases" search returns near-zero matches because nobody posts that |
| 949 | exact phrase. Intent modifiers should be stripped from search_query |
| 950 | and paraphrased across multiple subqueries. |
| 951 | """ |
| 952 | text = topic.lower() |
| 953 | return any(pattern in text for pattern in _INTENT_MODIFIER_PATTERNS) |
| 954 | |
| 955 | |
| 956 | def _intent_modifier_subqueries( |
| 957 | topic: str, |
| 958 | core: str, |
| 959 | base_search: str, |
| 960 | source_weights: dict[str, float], |
| 961 | ) -> list[schema.SubQuery]: |
| 962 | """Produce paraphrased subqueries for intent-modifier topics. |
| 963 | |
| 964 | The deterministic fallback used to echo the user's literal phrase |
| 965 | (e.g., "hermes agent use cases") into every search_query. This helper |
| 966 | fans out 3 extra subqueries that each express the intent differently |
| 967 | so retrieval pulls a broader corpus for reranking. |
| 968 | """ |
| 969 | entity = core or topic.strip() |
| 970 | sources = list(source_weights) |
| 971 | return [ |
| 972 | schema.SubQuery( |
| 973 | label="workflows", |
| 974 | search_query=f"{entity} workflow pipeline", |
| 975 | ranking_query=f"What real-world workflows or pipelines are people running with {entity}?", |
| 976 | sources=sources, |
| 977 | weight=0.6, |
| 978 | ), |
| 979 | schema.SubQuery( |
| 980 | label="production", |
| 981 | search_query=f"{entity} production real-world", |
| 982 | ranking_query=f"What production deployments or real-world use cases of {entity} are people describing?", |
| 983 | sources=sources, |
| 984 | weight=0.55, |
| 985 | ), |
| 986 | schema.SubQuery( |
| 987 | label="experience", |
| 988 | search_query=f"{entity} experience review", |
| 989 | ranking_query=f"What hands-on experience reports or reviews of {entity} exist in the last 30 days?", |
| 990 | sources=sources, |
| 991 | weight=0.5, |
| 992 | ), |
| 993 | ] |
| 994 | |
| 995 | |
| 996 | def _max_subqueries(intent: str, topic: str | None = None) -> int: |
| 997 | # how_to/opinion/product/breaking_news/prediction benefit from 4-5 |
| 998 | # paraphrased subqueries when the topic carries an intent modifier |
| 999 | # (use cases, workflows, examples, review, etc.). See 2026-04-19 |
| 1000 | # Hermes Agent Use Cases failure: prior cap of 3 produced near-literal |
| 1001 | # echoes of the topic instead of a paraphrase fanout. |
| 1002 | if intent == "comparison": |
| 1003 | # primary + one dedicated subquery per entity (up to COMPARISON_ENTITY_MAX) |
| 1004 | return competitors.COMPARISON_ENTITY_MAX + 1 |
| 1005 | # Intent-modifier topics get headroom for paraphrase fanout even when |
| 1006 | # the intent itself is factual/concept. Without this, a "Hermes Agent |
| 1007 | # use cases" query (classified "concept" after the 2026-04-19 default |
| 1008 | # change) would be capped at 2 and drop the fanout. |
| 1009 | if topic and _has_intent_modifier(topic): |
| 1010 | return 5 |
| 1011 | if intent in {"factual", "concept"}: |
| 1012 | return 2 |
| 1013 | return 5 |
| 1014 | |
| 1015 | |
| 1016 | def _default_sources_for_intent(intent: str, available_sources: list[str]) -> list[str]: |
| 1017 | if intent == "how_to": |
| 1018 | sources = _how_to_sources(available_sources) |
| 1019 | else: |
| 1020 | target_capabilities = DEFAULT_INTENT_CAPABILITIES.get(intent) |
| 1021 | if not target_capabilities: |
| 1022 | sources = list(available_sources) |
| 1023 | else: |
| 1024 | matched = [ |
| 1025 | source |
| 1026 | for source in available_sources |
| 1027 | if SOURCE_CAPABILITIES.get(source, set()) & target_capabilities |
| 1028 | ] |
| 1029 | sources = matched or list(available_sources) |
| 1030 | excluded = INTENT_SOURCE_EXCLUSIONS.get(intent, set()) |
| 1031 | if excluded: |
| 1032 | filtered = [s for s in sources if s not in excluded] |
| 1033 | return filtered or sources |
| 1034 | return sources |
| 1035 | |
| 1036 | |
| 1037 | def _how_to_sources(available_sources: list[str]) -> list[str]: |
| 1038 | """Pick one source per role: web/reference, video (prefer longform), discussion.""" |
| 1039 | selected: set[str] = set() |
| 1040 | has_video = False |
| 1041 | # Order matters: web first, then longform video, generic video, discussion. |
| 1042 | role_capabilities = [ |
| 1043 | {"web", "reference"}, |
| 1044 | {"video_longform"}, |
| 1045 | {"video"}, |
| 1046 | {"discussion"}, |
| 1047 | ] |
| 1048 | for role in role_capabilities: |
| 1049 | is_video_role = role & {"video", "video_longform"} |
| 1050 | if is_video_role and has_video: |
| 1051 | continue |
| 1052 | for source in available_sources: |
| 1053 | if source in selected: |
| 1054 | continue |
| 1055 | if SOURCE_CAPABILITIES.get(source, set()) & role: |
| 1056 | selected.add(source) |
| 1057 | if is_video_role: |
| 1058 | has_video = True |
| 1059 | break |
| 1060 | # After core role-based selection, include remaining sources with any |
| 1061 | # how_to-relevant capability (video, discussion, web, reference, link). |
| 1062 | how_to_caps = DEFAULT_INTENT_CAPABILITIES.get("how_to", set()) |
| 1063 | for source in available_sources: |
| 1064 | if source not in selected and SOURCE_CAPABILITIES.get(source, set()) & how_to_caps: |
| 1065 | selected.add(source) |
| 1066 | if not selected: |
| 1067 | return list(available_sources) |
| 1068 | return [source for source in available_sources if source in selected] |
| 1069 |