| from __future__ import annotations |
|
|
| import re |
| import threading |
| import time |
| from functools import lru_cache |
| from pathlib import Path |
| from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple |
|
|
| _CONCEPT_ASK_HINTS: Tuple[str, ...] = ( |
| "difference between", |
| "what is the difference", |
| "what's the difference", |
| "explain", |
| "plain english", |
| "in plain english", |
| "overview of", |
| "basics of", |
| "how does", |
| "how do", |
| "what is", |
| "what's", |
| "compare", |
| "comparison", |
| "versus", |
| " vs ", |
| "why would", |
| "when should i use", |
| "when do i use", |
| ) |
|
|
| _CONCEPT_BLOCKED_HINTS: Tuple[str, ...] = ( |
| "price", |
| "pricing", |
| "cost", |
| "msrp", |
| "discount", |
| "lead time", |
| "availability", |
| "in stock", |
| "stock", |
| "verizon policy", |
| "plan pricing", |
| "promotion", |
| "promo", |
| "end of sale", |
| "end-of-sale", |
| "end of life", |
| "end-of-life", |
| "eos", |
| "eol", |
| "firmware", |
| "certification", |
| "certified", |
| "fcc id", |
| "compatib", |
| "supported band", |
| "band support", |
| "n77", |
| "n78", |
| "n41", |
| "b66", |
| "b71", |
| ) |
|
|
| _HIGH_RISK_ADJUDICATION_HINTS: Tuple[str, ...] = ( |
| "pass inspection", |
| "inspection", |
| "code compliant", |
| "compliant with code", |
| "code requires", |
| "code require", |
| "required by code", |
| "meets code", |
| "meet code", |
| "code-approved", |
| "approved", |
| "approval", |
| "authority having jurisdiction", |
| "ahj", |
| "permit", |
| "permitted", |
| "permitting", |
| "sign off", |
| "sign-off", |
| ) |
|
|
| _HIGH_RISK_REGULATORY_HINTS: Tuple[str, ...] = ( |
| "code", |
| "compliance", |
| "kari's law", |
| "karis law", |
| "ray baum", |
| "ray baum's act", |
| "ray baums act", |
| "nfpa", |
| "ul", |
| "ahj", |
| "fire code", |
| "life safety", |
| "e911", |
| ) |
|
|
| _HIGH_RISK_OUTCOME_HINTS: Tuple[str, ...] = ( |
| "require", |
| "requires", |
| "required", |
| "must", |
| "need to", |
| "pass", |
| "approve", |
| "approved", |
| "approval", |
| "comply", |
| "compliant", |
| "compliance", |
| "certified", |
| "guarantee", |
| "guaranteed", |
| ) |
|
|
| _CURRENT_INFO_HINTS: Tuple[str, ...] = ( |
| "latest", |
| "most recent", |
| "today", |
| "right now", |
| "recent", |
| "updated", |
| "as of", |
| "2026", |
| "2027", |
| ) |
|
|
| _CONCEPT_WEAK_HINTS: Tuple[str, ...] = ( |
| "needs confirmation", |
| "not enough evidence", |
| "uncertain", |
| "depends on", |
| "varies by carrier", |
| "varies by deployment", |
| "ask for internal documentation", |
| "treat this as a concept explainer", |
| ) |
|
|
| _AUTHORITATIVE_FACT_HINTS: Tuple[str, ...] = ( |
| "default port", |
| "default tcp", |
| "default udp", |
| "tcp", |
| "udp", |
| "dimension", |
| "dimensions", |
| "size", |
| "weight", |
| "temperature", |
| "temperatures", |
| "operating temperature", |
| "storage temperature", |
| "battery capacity", |
| "runtime", |
| "certification", |
| "certifications", |
| "certified", |
| "approval", |
| "cloud management", |
| "management platform", |
| "diagnostic", |
| "diagnostics", |
| "monitoring tool", |
| "monitoring tools", |
| "firewall", |
| "port forwarding", |
| "vpn", |
| "overhead", |
| "imix", |
| "listed", |
| "documented", |
| "supported", |
| "what does the document say", |
| "what does the datasheet say", |
| "what does the manual say", |
| "what does the whitepaper say", |
| ) |
|
|
| _DOCUMENT_REFERENCE_HINTS: Tuple[str, ...] = ( |
| "document", |
| "documents", |
| "documentation", |
| "doc", |
| "docs", |
| "manual", |
| "datasheet", |
| "data sheet", |
| "whitepaper", |
| "pdf", |
| "guide", |
| "best practices", |
| "best practice", |
| ) |
|
|
| _PROCEDURAL_FACT_HINTS: Tuple[str, ...] = ( |
| "step", |
| "steps", |
| "setup", |
| "configure", |
| "configuration", |
| "install", |
| "installation", |
| "workflow", |
| "procedure", |
| "process", |
| "port forwarding", |
| ) |
|
|
| _NUMERIC_QUERY_HINTS: Tuple[str, ...] = ( |
| "how many", |
| "how much", |
| "maximum", |
| "minimum", |
| "max ", |
| "min ", |
| "bytes", |
| "mbps", |
| "gbps", |
| "hours", |
| "mah", |
| "degrees", |
| "celsius", |
| "fahrenheit", |
| "percent", |
| ) |
|
|
|
|
| def _source( |
| *, |
| id: str, |
| domain: str, |
| doc: str, |
| relative_path: str, |
| chunk_id: str, |
| excerpt: str, |
| score: float, |
| location: str = "", |
| ) -> Dict[str, Any]: |
| return { |
| "id": id, |
| "domain": domain, |
| "doc": doc, |
| "relative_path": relative_path, |
| "chunk_id": chunk_id, |
| "location": location, |
| "excerpt": excerpt, |
| "score": float(score), |
| } |
|
|
|
|
| def _normalize_doc_names(available_docs: Iterable[str]) -> set[str]: |
| names: set[str] = set() |
| for raw in available_docs: |
| text = str(raw or "").strip() |
| if not text: |
| continue |
| names.add(Path(text).name.lower()) |
| names.add(text.lower()) |
| return names |
|
|
|
|
| def _doc_available(available_docs: set[str], *candidates: str) -> bool: |
| if not candidates: |
| return True |
| if not available_docs: |
| return True |
| return any(str(candidate or "").strip().lower() in available_docs for candidate in candidates) |
|
|
|
|
| def _deterministic_payload( |
| *, |
| domain: str, |
| retrieval_mode: str, |
| result: str, |
| why: List[str], |
| next_actions: List[str], |
| sources: Optional[List[Dict[str, Any]]] = None, |
| files: Optional[List[str]] = None, |
| ) -> Dict[str, Any]: |
| return { |
| "result": str(result or "").strip(), |
| "why": [str(line or "").strip() for line in (why or []) if str(line or "").strip()], |
| "next_actions": [str(line or "").strip() for line in (next_actions or []) if str(line or "").strip()], |
| "sources": list(sources or []), |
| "files": list(files or []), |
| "meta": { |
| "domain": str(domain or "").strip().lower(), |
| "retrieval_mode": str(retrieval_mode or "").strip(), |
| "web_assisted": False, |
| "llm_assisted": False, |
| "non_internal_generated": False, |
| }, |
| } |
|
|
|
|
| def normalize_concept_question(value: Any) -> str: |
| text = str(value or "").lower() |
| text = text.replace("’", "'") |
| return " ".join(text.split()).strip() |
|
|
|
|
| @lru_cache(maxsize=512) |
| def _hint_pattern(hint: str) -> Optional[re.Pattern[str]]: |
| normalized = normalize_concept_question(hint) |
| if not normalized: |
| return None |
| if not re.fullmatch(r"[a-z0-9](?:[a-z0-9 '/-]*[a-z0-9])?|[a-z0-9]", normalized): |
| return None |
| escaped = re.escape(normalized).replace(r"\ ", r"\s+") |
| return re.compile(rf"(?<![a-z0-9]){escaped}(?![a-z0-9])") |
|
|
|
|
| def contains_any(text: str, hints: Iterable[str]) -> bool: |
| low = normalize_concept_question(text) |
| for raw_hint in hints: |
| hint = normalize_concept_question(raw_hint) |
| if not hint: |
| continue |
| pattern = _hint_pattern(hint) |
| if pattern: |
| if pattern.search(low): |
| return True |
| continue |
| if hint in low: |
| return True |
| return False |
|
|
|
|
| def is_high_risk_concept_question(question: str, blocked_terms: Iterable[str] = ()) -> bool: |
| low = normalize_concept_question(question) |
| if contains_any(low, _CONCEPT_BLOCKED_HINTS) or contains_any(low, blocked_terms): |
| return True |
| if contains_any(low, _HIGH_RISK_ADJUDICATION_HINTS): |
| return True |
| if contains_any(low, _HIGH_RISK_REGULATORY_HINTS) and contains_any(low, _HIGH_RISK_OUTCOME_HINTS): |
| return True |
| if re.search( |
| r"\b(will|would|can|does|is)\b.*\b(pass|meet|comply|require|required|approved|approval)\b", |
| low, |
| ) and contains_any(low, _HIGH_RISK_REGULATORY_HINTS): |
| return True |
| return False |
|
|
|
|
| def classify_concept_request( |
| question: str, |
| *, |
| domain: str, |
| enabled: bool, |
| scope_terms: Iterable[str], |
| blocked_terms: Iterable[str] = (), |
| strict_citation_required: bool = False, |
| extra_blocked: bool = False, |
| ) -> Dict[str, Any]: |
| low = normalize_concept_question(question) |
| in_scope = contains_any(low, scope_terms) |
| asks_concept = contains_any(low, _CONCEPT_ASK_HINTS) or ("?" in str(question or "") and in_scope) |
| blocked = bool( |
| extra_blocked |
| or strict_citation_required |
| or is_high_risk_concept_question(low, blocked_terms) |
| ) |
| return { |
| "domain": str(domain or "").strip().lower(), |
| "blocked": blocked, |
| "in_scope": in_scope, |
| "asks_concept": asks_concept, |
| "allow_concept": bool(enabled and in_scope and asks_concept and not blocked), |
| "wants_current_refinement": contains_any(low, _CURRENT_INFO_HINTS), |
| } |
|
|
|
|
| def concept_answer_needs_web_refinement(question: str, assistant: str, meta: Dict[str, Any]) -> bool: |
| if not bool((meta or {}).get("llm_assisted")): |
| return False |
| if contains_any(question, _CURRENT_INFO_HINTS): |
| return True |
| return contains_any(assistant, _CONCEPT_WEAK_HINTS) |
|
|
|
|
| def question_prefers_authoritative_evidence(question: str, domain: str = "") -> bool: |
| low = normalize_concept_question(question) |
| if not low: |
| return False |
| asks_answer = contains_any(low, _CONCEPT_ASK_HINTS) or ("?" in str(question or "")) |
| if not asks_answer: |
| return False |
| if contains_any(low, _DOCUMENT_REFERENCE_HINTS): |
| return True |
| if contains_any(low, _AUTHORITATIVE_FACT_HINTS): |
| return True |
| if contains_any(low, _PROCEDURAL_FACT_HINTS) and contains_any(low, _NUMERIC_QUERY_HINTS): |
| return True |
| if re.search(r"\b\d+(?:\.\d+)?\b", low) and ( |
| contains_any(low, _AUTHORITATIVE_FACT_HINTS) |
| or contains_any(low, _DOCUMENT_REFERENCE_HINTS) |
| or contains_any(low, _PROCEDURAL_FACT_HINTS) |
| ): |
| return True |
| dom = str(domain or "").strip().lower() |
| if dom == "router_docs" and contains_any( |
| low, |
| ( |
| "speedfusion", |
| "incontrol", |
| "cloud management", |
| "firewall", |
| "port forwarding", |
| "vpn", |
| ), |
| ): |
| return contains_any(low, _AUTHORITATIVE_FACT_HINTS) or contains_any(low, _DOCUMENT_REFERENCE_HINTS) |
| return False |
|
|
|
|
| def annotate_provenance(meta: Dict[str, Any]) -> Dict[str, Any]: |
| out = dict(meta or {}) |
| if bool(out.get("web_assisted")): |
| out["provenance_kind"] = "web" |
| out["provenance_label"] = "Web-sourced (not from internal docs)" |
| elif bool(out.get("llm_assisted")) or bool(out.get("non_internal_generated")): |
| out["provenance_kind"] = "model_generated" |
| out["provenance_label"] = "Model-generated (not from internal docs)" |
| else: |
| out["provenance_kind"] = "internal" |
| out["provenance_label"] = "Internal docs" |
| return out |
|
|
|
|
| def deterministic_concept_payload( |
| question: str, |
| *, |
| domain: str, |
| available_docs: Iterable[str] = (), |
| ) -> Optional[Dict[str, Any]]: |
| if question_prefers_authoritative_evidence(question, domain): |
| return None |
| low = normalize_concept_question(question) |
| dom = str(domain or "").strip().lower() |
| docs = _normalize_doc_names(available_docs) |
|
|
| if dom == "router_docs": |
| asks_failover_compare = bool( |
| ("failover" in low) |
| and ("cellular" in low) |
| and ("wired" in low) |
| and any(token in low for token in ("difference", "compare", "comparison", "versus", " vs ", "plain english", "in plain english")) |
| ) |
| asks_cellular_failover_basics = bool( |
| ("cellular failover" in low) |
| and any(token in low for token in ("what is", "what's", "explain", "plain english", "in plain english")) |
| ) |
| asks_wired_failover_basics = bool( |
| ("wired failover" in low or "inline failover" in low) |
| and any(token in low for token in ("what is", "what's", "explain", "plain english", "in plain english")) |
| ) |
| asks_4g_vs_5g = bool( |
| ( |
| (("cat 4" in low) or ("cat4" in low) or ("lte cat 4" in low)) |
| and ("5g" in low) |
| and any(x in low for x in ("difference", "practical difference", "branch backup", "backup")) |
| ) |
| or (("4g router" in low) and ("instead of a 5g" in low or "instead of 5g" in low)) |
| or ( |
| ("4g" in low) |
| and ("5g" in low) |
| and any( |
| x in low |
| for x in ( |
| "difference between 4g and 5g", |
| "difference between 5g and 4g", |
| "4g vs 5g", |
| "5g vs 4g", |
| "4g versus 5g", |
| "5g versus 4g", |
| "what is the difference between 4g and 5g", |
| "what's the difference between 4g and 5g", |
| ) |
| ) |
| ) |
| ) |
| asks_wan_vs_lan = any( |
| x in low |
| for x in ( |
| "wan vs lan", |
| "lan vs wan", |
| "difference between wan and lan", |
| "difference between lan and wan", |
| "what is the difference between wan and lan", |
| "what's the difference between wan and lan", |
| ) |
| ) |
| asks_poe_basics = bool( |
| ("poe" in low or "power over ethernet" in low) |
| and any( |
| x in low |
| for x in ( |
| "what is", |
| "what's", |
| "what does", |
| "explain", |
| "mean", |
| "plain english", |
| "run on it", |
| "difference", |
| ) |
| ) |
| ) |
| asks_esim_basics = bool( |
| ("esim" in low or "e sim" in low) |
| and any( |
| x in low |
| for x in ( |
| "what is", |
| "what's", |
| "what does", |
| "difference", |
| "versus", |
| " vs ", |
| "plain english", |
| "explain", |
| "physical sim", |
| ) |
| ) |
| ) |
|
|
| if asks_failover_compare: |
| return _deterministic_payload( |
| domain="router_docs", |
| retrieval_mode="router_failover_concept_fast", |
| result="\n".join( |
| [ |
| "Cellular failover vs wired failover for a branch router:", |
| "", |
| "| Topic | Cellular failover | Wired failover |", |
| "| --- | --- | --- |", |
| "| Backup path | Uses a cellular link when the primary wired circuit is unhealthy or down. | Uses a second wired WAN or wired provider path when the primary wired circuit is unhealthy or down. |", |
| "| Typical use | Branch resiliency when a wired backup is unavailable, slow to install, or too expensive for the site. | Branch resiliency when the site can justify dual wired paths for lower recurring risk and more predictable performance. |", |
| "| Performance expectation | Better continuity than no backup, but outcome still depends on signal quality, carrier conditions, and outage traffic load. | Usually steadier for large outage traffic loads if the secondary wired path is truly diverse and properly sized. |", |
| "| Qualification note | Validate signal, antenna path, hold timers, and what must stay up during an outage. | Validate circuit diversity, failover policy, and what must stay up during an outage. |", |
| ] |
| ), |
| why=[ |
| "Internal branch-backup guidance treats failover design as a traffic-profile and resilience decision, not a headline-speed decision.", |
| "Use cellular failover when fast deployment and practical survivability matter more than having a second wired circuit everywhere.", |
| ], |
| next_actions=[ |
| "Next step: confirm outage traffic load, acceptable failover behavior, and whether the site can support a diverse wired backup circuit.", |
| ], |
| sources=[ |
| _source( |
| id="RCFAIL1", |
| domain="router_docs", |
| doc="FAQ_master_updated.csv", |
| relative_path="docs/faq/FAQ_master_updated.csv", |
| chunk_id="faq:branch_backup_baseline", |
| excerpt="Branch-backup FAQ guidance focuses on cellular failover behind the main firewall with health checks and hold timers, so fit depends on outage traffic profile rather than headline radio generation alone.", |
| score=0.95, |
| ), |
| _source( |
| id="RCFAIL2", |
| domain="router_docs", |
| doc="InHand Networks-FWA02-Manual-1.pdf", |
| relative_path="/router_rag_files/01_documents/routers/inhand_networks/InHand%20Networks-FWA02-Manual-1.pdf", |
| chunk_id="InHand-Networks-FWA02-Manual-1__chunk_0002", |
| excerpt="Internal 5G FWA excerpt positions wired/cellular failover for branch-style deployments.", |
| score=0.90, |
| ), |
| ], |
| files=[ |
| "docs/faq/FAQ_master_updated.csv", |
| "/router_rag_files/01_documents/routers/inhand_networks/InHand%20Networks-FWA02-Manual-1.pdf", |
| ], |
| ) |
|
|
| if asks_cellular_failover_basics: |
| return _deterministic_payload( |
| domain="router_docs", |
| retrieval_mode="router_failover_concept_fast", |
| result="Cellular failover means the router keeps a cellular path ready so traffic can switch there if the primary wired connection fails or health checks say it is no longer usable.", |
| why=[ |
| "Rep-safe framing: this is a survivability pattern for branch continuity, not a promise of zero interruption.", |
| "Internal branch-backup guidance says the right design depends on outage traffic load, coverage confidence, and failover timer policy.", |
| ], |
| next_actions=[ |
| "Next step: qualify what must stay up during an outage and how much traffic the backup path really needs to carry.", |
| ], |
| sources=[ |
| _source( |
| id="RCFAIL3", |
| domain="router_docs", |
| doc="FAQ_master_updated.csv", |
| relative_path="docs/faq/FAQ_master_updated.csv", |
| chunk_id="faq:branch_backup_baseline", |
| excerpt="Branch-backup FAQ guidance focuses on cellular failover behind the main firewall with health checks and hold timers.", |
| score=0.95, |
| ) |
| ], |
| files=["docs/faq/FAQ_master_updated.csv"], |
| ) |
|
|
| if asks_wired_failover_basics: |
| return _deterministic_payload( |
| domain="router_docs", |
| retrieval_mode="router_failover_concept_fast", |
| result="Wired failover means the branch keeps a secondary wired path available so the router can switch away from the primary circuit when the main WAN fails or health checks mark it unusable. Inline failover is a common design where the device sits in the traffic path so cutover can happen without re-cabling many LAN devices.", |
| why=[ |
| "Internal FAQ guidance distinguishes simple failover from load balancing and explains inline failover as a practical branch-backup pattern.", |
| "Rep-safe framing: the value is continuity and cleaner cutover design, not a guarantee that every application session will stay up.", |
| ], |
| next_actions=[ |
| "Next step: confirm whether the site needs a second wired carrier path or an inline cellular backup design before recommending hardware.", |
| ], |
| sources=[ |
| _source( |
| id="RCFAIL4", |
| domain="router_docs", |
| doc="FAQ_200_ansers_set_3.csv", |
| relative_path="docs/faq/FAQ_200_ansers_set_3.csv", |
| chunk_id="faq:load_balancing_vs_failover", |
| excerpt="Load balancing spreads sessions across links, while failover keeps one link primary and switches to backup on outage.", |
| score=0.95, |
| ), |
| _source( |
| id="RCFAIL5", |
| domain="router_docs", |
| doc="FAQ_200_ansers_set_3.csv", |
| relative_path="docs/faq/FAQ_200_ansers_set_3.csv", |
| chunk_id="faq:inline_failover", |
| excerpt="Inline failover means the cellular device sits in the traffic path so it can pass wired WAN through when healthy and take over quickly when the wired WAN fails.", |
| score=0.95, |
| ), |
| ], |
| files=["docs/faq/FAQ_200_ansers_set_3.csv"], |
| ) |
|
|
| if asks_4g_vs_5g: |
| return _deterministic_payload( |
| domain="router_docs", |
| retrieval_mode="router_4g_vs_5g_positioning_fast", |
| result="\n".join( |
| [ |
| "CAT 4 LTE vs 5G for branch backup:", |
| "", |
| "| Topic | CAT 4 LTE backup | 5G backup |", |
| "| --- | --- | --- |", |
| "| Performance headroom | Internal examples show CAT 4 LTE models such as V810AD and RUT241 in the 150/50 Mbps class. | Internal 5G FWA examples call out materially higher throughput and position 5G as the higher-performance path. |", |
| "| Best fit | Light branch backup where the outage traffic profile is modest and cost simplicity matters. | Branch backup where you want more growth headroom or expect heavier failover traffic. |", |
| "| Qualification note | Confirm LTE coverage and whether backup traffic is truly light. | Confirm 5G coverage stability and whether the extra headroom is worth the added cost or complexity. |", |
| ] |
| ), |
| why=[ |
| "This is a practical positioning summary from internal examples, not a promise of site-specific throughput.", |
| "For branch backup, the decision is mostly about traffic headroom, coverage confidence, and budget discipline.", |
| ], |
| next_actions=[ |
| "Ask `give me a 4G vs 5G qualification checklist` for a discovery-call version.", |
| ], |
| sources=[ |
| _source( |
| id="RC45G1", |
| domain="router_docs", |
| doc="FAQ_master_updated.csv", |
| relative_path="docs/faq/FAQ_master_updated.csv", |
| chunk_id="faq:cat4_examples", |
| excerpt="Internal FAQ examples describe V810AD and RUT241 as 4G LTE Cat 4 devices in the 150/50 Mbps class.", |
| score=0.95, |
| ), |
| _source( |
| id="RC45G2", |
| domain="router_docs", |
| doc="InHand Networks-FWA02-Manual-1.pdf", |
| relative_path="/router_rag_files/01_documents/routers/inhand_networks/InHand%20Networks-FWA02-Manual-1.pdf", |
| chunk_id="InHand-Networks-FWA02-Manual-1__chunk_0002", |
| excerpt="Internal 5G FWA excerpt positions 5G as high-performance connectivity with SA/NSA support, LTE Cat 19 compatibility, and wired/cellular failover for branch-style deployments.", |
| score=0.93, |
| ), |
| _source( |
| id="RC45G3", |
| domain="router_docs", |
| doc="FAQ_master_updated.csv", |
| relative_path="docs/faq/FAQ_master_updated.csv", |
| chunk_id="faq:branch_backup_baseline", |
| excerpt="Branch-backup FAQ guidance focuses on cellular failover behind the main firewall with health checks and hold timers, so fit depends on outage traffic profile rather than headline radio generation alone.", |
| score=0.91, |
| ), |
| ], |
| files=[ |
| "docs/faq/FAQ_master_updated.csv", |
| "/router_rag_files/01_documents/routers/inhand_networks/InHand%20Networks-FWA02-Manual-1.pdf", |
| ], |
| ) |
|
|
| if asks_wan_vs_lan: |
| return _deterministic_payload( |
| domain="router_docs", |
| retrieval_mode="router_wan_lan_concept_fast", |
| result="WAN is the router's upstream provider side. LAN is the local side that connects the onsite devices and networks behind the router.", |
| why=[ |
| "Rep shorthand: WAN = outside or carrier-facing connection; LAN = inside or local device network.", |
| "Internal FAQ guidance frames a cellular router as using LTE or 5G for the WAN side while sharing that connection over Ethernet or Wi-Fi to LAN devices.", |
| ], |
| next_actions=[ |
| "Next step: confirm how many WAN and LAN paths the site needs for failover, segmentation, and local device count.", |
| ], |
| sources=[ |
| _source( |
| id="RCLAN1", |
| domain="router_docs", |
| doc="FAQ_200_ansers_set_3.csv", |
| relative_path="docs/faq/FAQ_200_ansers_set_3.csv", |
| chunk_id="faq:cellular_router_and_gateway_basics", |
| excerpt="A cellular router uses an LTE/5G modem as its WAN link and shares that connection to devices over Ethernet and/or Wi-Fi. A router handles NAT and the LAN side for local devices.", |
| score=0.94, |
| ) |
| ], |
| files=["docs/faq/FAQ_200_ansers_set_3.csv"], |
| ) |
|
|
| if asks_poe_basics: |
| return _deterministic_payload( |
| domain="router_docs", |
| retrieval_mode="router_poe_concept_fast", |
| result="PoE input matters because Power over Ethernet can carry both data and power over one cable. In router deployments it is useful when you want cleaner cabling or need to power a device where a nearby outlet is not practical.", |
| why=[ |
| "Internal FAQ guidance describes PoE as a single-cable power and data option that can simplify kiosk, ceiling-mount, or edge-device installs.", |
| "Model fit still matters because some routers accept PoE directly while others need an injector or splitter.", |
| ], |
| next_actions=[ |
| "Next step: verify the selected model's documented PoE capability before assuming a one-cable install design.", |
| ], |
| sources=[ |
| _source( |
| id="RPOE1", |
| domain="router_docs", |
| doc="FAQ_200_ansers_set_3.csv", |
| relative_path="docs/faq/FAQ_200_ansers_set_3.csv", |
| chunk_id="faq:poe_basics", |
| excerpt="PoE (Power over Ethernet) sends power and data over one cable. Some routers accept PoE directly; others need a splitter or injector.", |
| score=0.95, |
| ), |
| _source( |
| id="RPOE2", |
| domain="router_docs", |
| doc="router_pricing_catalog_normalized.csv", |
| relative_path="backend/app/knowledgebase/data/normalized/router_pricing_catalog_normalized.csv", |
| chunk_id="normalized:poe_field", |
| excerpt="Normalized router catalog includes a dedicated PoE field alongside modem type, Wi-Fi, ruggedization, and device type for model-level comparisons.", |
| score=0.90, |
| ), |
| ], |
| files=[ |
| "docs/faq/FAQ_200_ansers_set_3.csv", |
| "backend/app/knowledgebase/data/normalized/router_pricing_catalog_normalized.csv", |
| ], |
| ) |
|
|
| if asks_esim_basics: |
| return _deterministic_payload( |
| domain="router_docs", |
| retrieval_mode="router_esim_concept_fast", |
| result="eSIM means the provisioning profile is downloaded to a built-in chip instead of inserting a removable SIM card. The practical difference is usually operational: shipping, activation, and profile changes can be handled more remotely instead of mailing cards.", |
| why=[ |
| "Internal FAQ guidance says eSIM is useful when shipping devices or switching profiles without mailing physical SIM cards.", |
| "Internal terminology guidance also notes that EID identifies the eSIM chip, while ICCID identifies the SIM profile and IMEI identifies the device or modem.", |
| ], |
| next_actions=[ |
| "Next step: validate exact device and carrier support before promising eSIM activation on a quoted model.", |
| ], |
| sources=[ |
| _source( |
| id="RESIM1", |
| domain="router_docs", |
| doc="FAQ_master_updated.csv", |
| relative_path="docs/faq/FAQ_master_updated.csv", |
| chunk_id="faq:esim_definition", |
| excerpt="A physical SIM is removable; an eSIM is a built-in chip provisioned remotely with an activation profile and is useful when shipping devices or switching profiles without mailing cards.", |
| score=0.96, |
| ), |
| _source( |
| id="RESIM2", |
| domain="router_docs", |
| doc="FAQ_200_ansers_set_3.csv", |
| relative_path="docs/faq/FAQ_200_ansers_set_3.csv", |
| chunk_id="faq:eid_iccid_imei", |
| excerpt="Internal FAQ guidance says ICCID identifies the SIM, IMEI identifies the modem/device, and EID identifies the eSIM chip used to download profiles.", |
| score=0.92, |
| ), |
| ], |
| files=["docs/faq/FAQ_master_updated.csv", "docs/faq/FAQ_200_ansers_set_3.csv"], |
| ) |
|
|
| return None |
|
|
| if dom == "masters": |
| asks_securefax_compare = ( |
| any(token in low for token in ("securefax", "secure fax")) |
| and any(token in low for token in ("ifax", "i fax")) |
| and any(token in low for token in ("difference", "differences", "compare", "vs", "versus")) |
| ) |
| asks_sip_accounts = ( |
| any(token in low for token in ("sip account", "sip accounts")) |
| and any(token in low for token in ("what is", "what are", "explain", "plain english", "sales rep")) |
| ) |
| asks_contact_center = ( |
| ("contact center" in low) |
| and any(token in low for token in ("what is", "what's", "explain", "plain english", "sales rep", "simple terms")) |
| ) |
| if asks_securefax_compare and _doc_available(docs, "mst_securefax.pdf", "mst_ifax.pdf"): |
| sources: List[Dict[str, Any]] = [] |
| files: List[str] = [] |
| if _doc_available(docs, "mst_ifax.pdf"): |
| sources.append( |
| _source( |
| id="MFO1", |
| domain="masters", |
| doc="MST_iFAX.pdf", |
| relative_path="MST_iFAX.pdf", |
| chunk_id="masters:ifax_positioning", |
| excerpt="Internal iFAX reference used for electronic fax service with web-portal positioning.", |
| score=0.95, |
| ) |
| ) |
| files.append("MST_iFAX.pdf") |
| if _doc_available(docs, "mst_securefax.pdf"): |
| sources.append( |
| _source( |
| id="MFO2", |
| domain="masters", |
| doc="MST_SecureFAX.pdf", |
| relative_path="MST_SecureFAX.pdf", |
| chunk_id="masters:securefax_positioning", |
| excerpt="Internal SecureFAX reference documents analog POTS line replacement for fax machines.", |
| score=0.95, |
| ) |
| ) |
| files.append("MST_SecureFAX.pdf") |
| return _deterministic_payload( |
| domain="masters", |
| retrieval_mode="masters_ifax_securefax_compare_fast", |
| result="\n".join( |
| [ |
| "iFAX vs SecureFAX (from internal Masters references):", |
| "", |
| "| Offer | What docs indicate |", |
| "| --- | --- |", |
| "| iFAX | Electronic fax service with a web portal workflow. |", |
| "| SecureFAX | Analog POTS line replacement for fax machines. |", |
| "| Practical difference for reps | Position iFAX as electronic fax service and SecureFAX as analog fax-line replacement; avoid adding uncited capability claims beyond the approved references. |", |
| ] |
| ), |
| why=[ |
| "Uses internal Masters offer references rather than model-generated or web-generated comparison language.", |
| "Keeps the answer at the documented positioning level and avoids uncited pricing or feature expansion.", |
| ], |
| next_actions=[ |
| "Ask `show the related SKUs` if you need the documented offer rows next.", |
| "Ask `turn this into a rep talk track` if you want a short internal positioning script.", |
| ], |
| sources=sources, |
| files=files, |
| ) |
| if asks_sip_accounts and _doc_available(docs, "mst_sip accounts.pdf", "mst_sip_accounts.pdf"): |
| return _deterministic_payload( |
| domain="masters", |
| retrieval_mode="masters_sip_accounts_concept_fast", |
| result="SIP accounts in plain English: they are the account records used to set up and manage SIP-based voice service. In practice, they hold the service identity and admin context reps need for users, numbers, trunks, and call-routing setup before detailed implementation is finalized.", |
| why=[ |
| "Kept at the concept level and anchored to the internal SIP Accounts reference.", |
| "This explains what SIP accounts are without inventing carrier policy, pricing, or exact provisioning rules.", |
| ], |
| next_actions=[ |
| "Ask `turn this into discovery questions` for a rep-friendly checklist.", |
| "Ask `compare SIP accounts vs contact center` if you need adjacent positioning language.", |
| ], |
| sources=[ |
| _source( |
| id="MSIP1", |
| domain="masters", |
| doc="MST_SIP Accounts.pdf", |
| relative_path="MST_SIP Accounts.pdf", |
| chunk_id="masters:sip_accounts_basics", |
| excerpt="Internal SIP Accounts reference used for service-account, provisioning, and voice-service framing.", |
| score=0.95, |
| ) |
| ], |
| files=["MST_SIP Accounts.pdf"], |
| ) |
| if asks_contact_center and _doc_available(docs, "mst_contact center.pdf", "mst_contact_center.pdf"): |
| return _deterministic_payload( |
| domain="masters", |
| retrieval_mode="masters_contact_center_concept_fast", |
| result="Contact center in plain English: it is the service and workflow layer that helps a business route, manage, and measure customer interactions across agents, queues, and channels. It is broader than a basic voice line or SIP account because it focuses on customer-handling workflows, agent experience, and reporting.", |
| why=[ |
| "Kept at the concept level and anchored to the internal Contact Center reference.", |
| "This avoids uncited carrier-policy, pricing, or feature-detail claims while still giving reps a usable positioning explanation.", |
| ], |
| next_actions=[ |
| "Ask `compare SIP accounts vs contact center` if you need a side-by-side positioning answer.", |
| "Ask `turn this into discovery questions` if you want rep-ready follow-up prompts.", |
| ], |
| sources=[ |
| _source( |
| id="MCC1", |
| domain="masters", |
| doc="MST_Contact Center.pdf", |
| relative_path="MST_Contact Center.pdf", |
| chunk_id="masters:contact_center_concept", |
| excerpt="Internal Contact Center reference used for agent, queue, workflow, and customer-interaction positioning.", |
| score=0.95, |
| ) |
| ], |
| files=["MST_Contact Center.pdf"], |
| ) |
| return None |
|
|
| if dom == "pots": |
| asks_copper_sunset = ( |
| ("copper sunset" in low) |
| and any( |
| token in low |
| for token in ( |
| "difference", |
| "what is", |
| "what's", |
| "what does", |
| "mean", |
| "explain", |
| "plain english", |
| "pots replacement", |
| ) |
| ) |
| ) |
| asks_priority_lines = bool( |
| any(token in low for token in ("highest priority", "priority lines", "which lines first", "what lines first", "lines first", "prioritize first")) |
| and any(token in low for token in ("pots", "pots replacement", "migration", "line replacement", "analog line", "analog lines")) |
| ) |
| asks_discovery_first = bool( |
| any( |
| token in low |
| for token in ( |
| "what should a verizon rep ask first", |
| "what should a rep ask first", |
| "ask first before recommending", |
| "first before recommending", |
| "discovery questions first", |
| ) |
| ) |
| and any(token in low for token in ("pots", "pots replacement", "migration", "analog line", "analog lines", "copper sunset")) |
| ) |
| asks_pots_replacement_basics = bool( |
| ("pots replacement" in low) |
| and any( |
| token in low |
| for token in ( |
| "what is", |
| "what's", |
| "what does", |
| "explain", |
| "plain english", |
| "simple terms", |
| "multi-site", |
| "retail", |
| ) |
| ) |
| ) |
| asks_dual_pathway = bool( |
| ("dual pathway" in low) |
| and any(token in low for token in ("fire", "fire alarm", "elevator")) |
| and any(token in low for token in ("what is", "what's", "explain", "plain english", "requirements", "requirement")) |
| ) |
| if asks_copper_sunset and _doc_available(docs, "pots_top100_questions_draft.md"): |
| return _deterministic_payload( |
| domain="pots", |
| retrieval_mode="pots_copper_sunset_concept_fast", |
| result="Copper sunset is the retirement of legacy carrier copper voice infrastructure. POTS replacement is the migration plan that moves each analog line to a supported replacement path for its specific use case.", |
| why=[ |
| "Internal POTS playbook separates the business driver from the migration action: copper sunset explains why the legacy line is at risk, while POTS replacement explains how the line is migrated.", |
| "The same playbook says the next decision is not provider-first; it is line-inventory and criticality-first.", |
| ], |
| next_actions=[ |
| "Next step: classify each line by use case and criticality before recommending provider, pathway, or BOM.", |
| ], |
| sources=[ |
| _source( |
| id="PCC1", |
| domain="pots", |
| doc="pots_top100_questions_draft.md", |
| relative_path="backend/app/pots_ai/data/pots_top100_questions_draft.md", |
| chunk_id="pots_top100:q2_copper_sunset", |
| excerpt="Top-100 POTS draft explains copper sunset as the retirement of legacy carrier copper service and treats POTS replacement as the migration path away from those lines.", |
| score=0.94, |
| ) |
| ], |
| files=["pots_top100_questions_draft.md"], |
| ) |
| if asks_pots_replacement_basics and _doc_available(docs, "pots_top100_questions_draft.md"): |
| return _deterministic_payload( |
| domain="pots", |
| retrieval_mode="pots_replacement_overview_concept_fast", |
| result="POTS replacement is a structured migration of legacy analog lines to supported replacement paths, starting with a line inventory and criticality review, then moving through pilot, validation, and phased rollout. For a multi-site retail customer, the goal is continuity for critical endpoints with a repeatable site-by-site rollout plan rather than a one-shot swap everywhere.", |
| why=[ |
| "Internal POTS playbook says the target is continuity for critical analog endpoints, documented compliance posture, lower operational burden, and a predictable rollout plan.", |
| "The same playbook says multi-site migrations should be prioritized by risk and readiness, then scaled from pilot sites with a repeatable runbook.", |
| ], |
| next_actions=[ |
| "Next step: collect site list, line types, endpoint criticality, and install constraints before recommending provider, pathway, or BOM.", |
| ], |
| sources=[ |
| _source( |
| id="PPB1", |
| domain="pots", |
| doc="pots_top100_questions_draft.md", |
| relative_path="backend/app/pots_ai/data/pots_top100_questions_draft.md", |
| chunk_id="pots_top100:q5_project_outcomes", |
| excerpt="POTS replacement should target continuity for critical analog endpoints, documented compliance posture, lower operational burden, and a predictable rollout plan.", |
| score=0.95, |
| ), |
| _source( |
| id="PPB2", |
| domain="pots", |
| doc="pots_top100_questions_draft.md", |
| relative_path="backend/app/pots_ai/data/pots_top100_questions_draft.md", |
| chunk_id="pots_top100:q8_multisite_priority", |
| excerpt="Prioritize multi-site migrations by life-safety impact, outage risk, complexity, and readiness. Start with pilot sites, then scale with a repeatable runbook.", |
| score=0.94, |
| ), |
| ], |
| files=["pots_top100_questions_draft.md"], |
| ) |
| if asks_dual_pathway and _doc_available(docs, "pots_top100_questions_draft.md"): |
| return _deterministic_payload( |
| domain="pots", |
| retrieval_mode="pots_dual_pathway_concept_fast", |
| result="For fire alarm and elevator use cases, dual-pathway planning means you do not treat the migration like a basic voice-line swap. You define the failure scenarios first, then validate the needed redundancy controls for power, connectivity, device behavior, and test procedures before rollout.", |
| why=[ |
| "Internal POTS guidance says fire alarm paths need stricter validation against standards and local authority expectations, while elevator paths need outage-behavior and test-procedure signoff with responsible parties.", |
| "The same playbook says redundancy planning should start from failure scenarios, then map power, connectivity, and device-level controls to those scenarios.", |
| ], |
| next_actions=[ |
| "Next step: confirm fire and elevator test criteria, responsible vendors, and required acceptance evidence before final design or cutover dates are proposed.", |
| ], |
| sources=[ |
| _source( |
| id="PDP1", |
| domain="pots", |
| doc="pots_top100_questions_draft.md", |
| relative_path="backend/app/pots_ai/data/pots_top100_questions_draft.md", |
| chunk_id="pots_top100:q16_fire_alarm_treatment", |
| excerpt="Fire-related paths need strict validation against relevant standards and local authority expectations. Confirm required supervision and test criteria, then document acceptance evidence.", |
| score=0.95, |
| ), |
| _source( |
| id="PDP2", |
| domain="pots", |
| doc="pots_top100_questions_draft.md", |
| relative_path="backend/app/pots_ai/data/pots_top100_questions_draft.md", |
| chunk_id="pots_top100:q17_elevator_verification", |
| excerpt="Elevator emergency phones should verify code-aligned behavior, reliability under outage conditions, and required test procedures with responsible parties before broad rollout.", |
| score=0.95, |
| ), |
| _source( |
| id="PDP3", |
| domain="pots", |
| doc="pots_top100_questions_draft.md", |
| relative_path="backend/app/pots_ai/data/pots_top100_questions_draft.md", |
| chunk_id="pots_top100:q39_redundancy_controls", |
| excerpt="Define failure scenarios first, then map redundancy controls for power, connectivity, and device-level behavior to those scenarios and test failover during pilot.", |
| score=0.93, |
| ), |
| ], |
| files=["pots_top100_questions_draft.md"], |
| ) |
| if asks_priority_lines and _doc_available(docs, "pots_top100_questions_draft.md"): |
| return _deterministic_payload( |
| domain="pots", |
| retrieval_mode="pots_priority_lines_concept_fast", |
| result="In a POTS replacement project, life-safety and operational-critical lines usually come first: fire, elevator, alarm, entry or intercom, and emergency phones. Non-critical lines are usually sequenced after those paths are validated.", |
| why=[ |
| "Internal POTS playbook prioritizes migration by criticality rather than by carrier or price first.", |
| "This keeps life-safety and operational risk visible before the team starts optimizing commercial details.", |
| ], |
| next_actions=[ |
| "Next step: build a site-by-site line inventory and tag each endpoint by use case before sequencing the rollout.", |
| ], |
| sources=[ |
| _source( |
| id="PPL1", |
| domain="pots", |
| doc="pots_top100_questions_draft.md", |
| relative_path="backend/app/pots_ai/data/pots_top100_questions_draft.md", |
| chunk_id="pots_top100:q3_priority_lines", |
| excerpt="Top-100 POTS draft says life-safety and operational-critical lines usually come first: fire, elevator, alarm, entry/intercom, and emergency phones.", |
| score=0.94, |
| ) |
| ], |
| files=["pots_top100_questions_draft.md"], |
| ) |
| if asks_discovery_first and _doc_available(docs, "pots_top100_questions_draft.md"): |
| return _deterministic_payload( |
| domain="pots", |
| retrieval_mode="pots_discovery_first_concept_fast", |
| result="Before recommending a POTS replacement path, start with the customer's line inventory, endpoint criticality, outage tolerance, and site-readiness constraints. Provider selection should come after the team understands which lines are life-safety, operational-critical, or easy to pilot.", |
| why=[ |
| "Internal POTS guidance says migration planning should begin with line classification and failure-impact review rather than jumping straight to provider selection.", |
| "That keeps life-safety, fax continuity, elevator, and alarm requirements visible before the team optimizes commercial or provider details.", |
| ], |
| next_actions=[ |
| "Next step: capture site count, line types, endpoint owners, test requirements, and cutover constraints before narrowing the provider shortlist.", |
| ], |
| sources=[ |
| _source( |
| id="PDF1", |
| domain="pots", |
| doc="pots_top100_questions_draft.md", |
| relative_path="backend/app/pots_ai/data/pots_top100_questions_draft.md", |
| chunk_id="pots_top100:q1_discovery_first", |
| excerpt="Top-100 POTS draft says the first move is a line inventory plus endpoint criticality review, then pilot planning, then provider/pathway fit.", |
| score=0.95, |
| ), |
| _source( |
| id="PDF2", |
| domain="pots", |
| doc="pots_top100_questions_draft.md", |
| relative_path="backend/app/pots_ai/data/pots_top100_questions_draft.md", |
| chunk_id="pots_top100:q3_priority_lines", |
| excerpt="Life-safety and operational-critical lines should be prioritized before optimizing commercial details or scaling rollout waves.", |
| score=0.94, |
| ), |
| ], |
| files=["pots_top100_questions_draft.md"], |
| ) |
| return None |
|
|
| return None |
|
|
|
|
| def build_concept_prompt(scope: str) -> str: |
| return ( |
| "You are a technical pre-sales assistant for Masters Telecom and Verizon partner users. " |
| f"{scope} " |
| "Answer only generic telecom concepts when internal retrieval is weak. " |
| "Do not fabricate pricing, discounts, lead times, Verizon policy, lifecycle dates, certifications, compatibility claims, or exact model specs. " |
| "If exact internal documentation is needed, say so directly. " |
| "For open-ended questions, answer in plain English sentences first and then short bullets when helpful. " |
| "Do not reply with raw fragments, source dumps, or tables unless the user explicitly asked for them. " |
| "Use short sections: Result, Why, Next action." |
| ) |
|
|
|
|
| def build_public_web_source_guidance(domain: str) -> str: |
| dom = str(domain or "").strip().lower() |
| lead = { |
| "router_docs": "When web search is required, prefer higher-trust public router and carrier sources.", |
| "masters": "When web search is required, prefer higher-trust public Masters Telecom and carrier sources.", |
| "pots": "When web search is required, prefer higher-trust public POTS replacement and carrier sources.", |
| }.get(dom, "When web search is required, prefer higher-trust public telecom sources.") |
| return " ".join( |
| [ |
| lead, |
| "Prefer official manufacturer product pages, datasheets, manuals, download portals, official knowledge-base articles, and standards/regulatory sources before reseller or catalog sites.", |
| "Use https://opendevelopment.verizonwireless.com for recently approved Verizon devices and approval-status context when relevant.", |
| "Use https://masterstelecom.com for Masters Telecom services, public solution positioning, and related service context when relevant.", |
| "Use https://5gstore.com only as secondary public catalog context on routers and related wireless hardware when the official vendor documentation is thin or unavailable.", |
| "Treat these as preferred public sources, not as authority for pricing, discounts, lead times, Verizon policy, or exact compatibility guarantees.", |
| ] |
| ) |
|
|
|
|
| def wrap_structured_fallback( |
| text: str, |
| *, |
| label: str, |
| why_line: str, |
| next_action_line: str, |
| formatter: Callable[[str, list[str], list[str]], str], |
| ) -> str: |
| clean = str(text or "").strip() |
| if not clean: |
| clean = "I could not generate a reliable fallback answer." |
| if "**Result**" not in clean and "Result" not in clean: |
| return formatter( |
| f"{label}: {clean}", |
| [why_line], |
| [next_action_line], |
| ) |
| if label.lower() not in clean.lower(): |
| return clean.replace("**Result**", f"**Result**\n\n{label}:", 1) |
| return clean |
|
|
|
|
| def responses_create_with_deadline(client: Any, *, timeout_s: float, **kwargs: Any) -> Any: |
| effective_timeout_s = max(0.01, float(timeout_s or 0.0)) |
| call_kwargs = dict(kwargs) |
| call_kwargs["timeout"] = effective_timeout_s |
| result_box: Dict[str, Any] = {} |
| error_box: Dict[str, Exception] = {} |
|
|
| def _run() -> None: |
| try: |
| result_box["value"] = client.responses.create(**call_kwargs) |
| except Exception as exc: |
| error_box["error"] = exc |
|
|
| thread = threading.Thread( |
| target=_run, |
| daemon=True, |
| name=f"openai-responses-{int(time.monotonic() * 1000)}", |
| ) |
| thread.start() |
| thread.join(effective_timeout_s + max(0.01, min(0.35, effective_timeout_s * 0.15))) |
| if thread.is_alive(): |
| raise TimeoutError(f"responses.create exceeded {effective_timeout_s:.2f}s") |
| if "error" in error_box: |
| raise error_box["error"] |
| return result_box.get("value") |
|
|
