saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
328 lines
13 KiB
Python
328 lines
13 KiB
Python
"""Deterministic mock provider - the default (`LLM_PROVIDER=mock`) and what
|
|
every automated test runs against. Never calls a network or costs money.
|
|
|
|
Rather than a generic reflection-based filler, each of the six analysis
|
|
tasks gets a purpose-built, deterministic builder that reads the same
|
|
evidence block a real model would see (see app/prompts/base.py) and
|
|
produces genuinely useful output from it - real counts, real titles, real
|
|
severities - never fabricated facts. This is what makes the fixture demo
|
|
(Phase 9) work end-to-end without a paid API key.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from app.prompts.alert_summarization import AlertSummary
|
|
from app.prompts.base import extract_evidence_block
|
|
from app.prompts.change_significance import ChangeSignificanceAssessment
|
|
from app.prompts.company_profile import CompanyProfileExtraction
|
|
from app.prompts.extraction import ExtractionResult
|
|
from app.prompts.relevance import RelevanceAssessment
|
|
from app.prompts.report_generation import ReportContent
|
|
from app.prompts.synthesis import SynthesisResult
|
|
|
|
|
|
def _confidence_label(score: float) -> str:
|
|
if score >= 0.85:
|
|
return "confirmed"
|
|
if score >= 0.65:
|
|
return "strongly_indicated"
|
|
if score >= 0.45:
|
|
return "likely"
|
|
if score >= 0.25:
|
|
return "possible"
|
|
if score > 0:
|
|
return "unconfirmed"
|
|
return "insufficient_evidence"
|
|
|
|
|
|
def _build_relevance(evidence: dict[str, Any]) -> dict[str, Any]:
|
|
text = (evidence.get("document_text") or "").lower()
|
|
focus = (evidence.get("monitoring_focus") or "").lower()
|
|
focus_words = [w for w in focus.split() if len(w) > 4]
|
|
matches_focus = any(w in text for w in focus_words) if focus_words else False
|
|
return {
|
|
"is_relevant": True,
|
|
"matches_focus": matches_focus,
|
|
"topic_categories": [],
|
|
"source_reliability": 0.7,
|
|
"reasoning": "Mock provider: keyword-based heuristic (no live LLM configured).",
|
|
}
|
|
|
|
|
|
def _build_extraction(evidence: dict[str, Any]) -> dict[str, Any]:
|
|
text = (evidence.get("document_text") or "").strip()
|
|
if not text:
|
|
return {"signals": []}
|
|
first_sentence = text.split(".")[0][:200].strip()
|
|
if not first_sentence:
|
|
return {"signals": []}
|
|
return {
|
|
"signals": [
|
|
{
|
|
"signal_type": "event",
|
|
"description": first_sentence,
|
|
"supporting_passage": first_sentence,
|
|
"date": None,
|
|
"entities": [],
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
def _build_synthesis(evidence: dict[str, Any]) -> dict[str, Any]:
|
|
signals = evidence.get("signals") or []
|
|
if len(signals) < 2:
|
|
return {"conclusions": []}
|
|
return {
|
|
"conclusions": [
|
|
{
|
|
"conclusion": (
|
|
"Multiple related signals were detected together; the mock provider "
|
|
"does not attempt fine-grained synthesis - configure a live LLM provider "
|
|
"for a specific conclusion."
|
|
),
|
|
"evidence_summary": [s.get("description", "") for s in signals[:5]],
|
|
"source_count": len(signals),
|
|
"confidence": 0.3,
|
|
"alternative_explanations": [
|
|
"A live LLM provider would assess this more precisely."
|
|
],
|
|
"missing_information": [],
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
def _build_report(evidence: dict[str, Any]) -> dict[str, Any]:
|
|
profile = evidence.get("company_profile") or {}
|
|
company_name = profile.get("name") or "The company"
|
|
documents = evidence.get("source_documents") or []
|
|
changes = evidence.get("detected_changes") or []
|
|
failed = evidence.get("sources_that_failed_to_collect") or []
|
|
|
|
# The discovered profile (real, from onboarding) is genuine evidence even
|
|
# when no monitoring run has collected source_documents/detected_changes
|
|
# yet - build company_overview/market_positioning from it honestly
|
|
# rather than defaulting straight to "insufficient evidence".
|
|
overview_parts = []
|
|
if profile.get("description"):
|
|
overview_parts.append(profile["description"])
|
|
facts = []
|
|
if profile.get("industry"):
|
|
facts.append(f"industry: {profile['industry']}")
|
|
if profile.get("headquarters"):
|
|
facts.append(f"headquartered in {profile['headquarters']}")
|
|
elif profile.get("country") or profile.get("region"):
|
|
facts.append(
|
|
f"based in {', '.join(f for f in (profile.get('country'), profile.get('region')) if f)}"
|
|
)
|
|
if profile.get("aliases"):
|
|
facts.append(f"also known as {', '.join(profile['aliases'])}")
|
|
if facts:
|
|
overview_parts.append(f"{company_name} ({'; '.join(facts)}).")
|
|
company_overview = " ".join(overview_parts) or f"No description on file for {company_name}."
|
|
|
|
if profile.get("competitors"):
|
|
market_positioning = (
|
|
f"{company_name} operates in a space that includes "
|
|
f"{', '.join(profile['competitors'])} as named competitors, per the discovered "
|
|
"company profile. No comparative data (pricing, features, market share) is "
|
|
"available to assess relative positioning."
|
|
)
|
|
else:
|
|
market_positioning = "Insufficient evidence to assess market positioning."
|
|
|
|
executive_summary = (
|
|
f"Mock analysis (LLM_PROVIDER=mock) based on {len(documents)} collected document(s) "
|
|
f"and {len(changes)} detected change(s) for {company_name}."
|
|
)
|
|
if failed:
|
|
executive_summary += (
|
|
f" {len(failed)} source(s) failed to collect this run and are excluded below."
|
|
)
|
|
|
|
recent_developments = [
|
|
{
|
|
"headline": change.get("summary") or "Change detected",
|
|
"summary": (
|
|
f"{(change.get('change_type') or 'change').replace('_', ' ')} detected "
|
|
f"with {change.get('severity') or 'unknown'} severity."
|
|
),
|
|
"evidence": [
|
|
{
|
|
"detected_change_id": change.get("id"),
|
|
"description": change.get("summary") or "",
|
|
}
|
|
],
|
|
"confidence": _confidence_label(change.get("confidence_score") or 0.5),
|
|
"category": change.get("change_type"),
|
|
"date": change.get("created_at"),
|
|
}
|
|
for change in changes[:10]
|
|
]
|
|
|
|
hiring_signals = [
|
|
{
|
|
"headline": doc.get("title") or doc.get("url") or "Job posting",
|
|
"summary": (doc.get("excerpt") or "")[:280],
|
|
"evidence": [
|
|
{
|
|
"source_document_id": doc.get("id"),
|
|
"url": doc.get("url"),
|
|
"description": "Collected source document",
|
|
}
|
|
],
|
|
"confidence": "confirmed",
|
|
"category": "job_posting",
|
|
"date": doc.get("retrieved_date"),
|
|
}
|
|
for doc in documents
|
|
if doc.get("source_type") == "job_posting"
|
|
]
|
|
|
|
unknowns = (
|
|
[f"{len(failed)} source(s) failed to collect this run: {', '.join(failed[:5])}"]
|
|
if failed
|
|
else []
|
|
)
|
|
|
|
return {
|
|
"executive_summary": executive_summary,
|
|
"company_overview": company_overview,
|
|
"products_and_services": [],
|
|
"market_positioning": market_positioning,
|
|
"recent_developments": recent_developments,
|
|
"strategic_initiatives": [],
|
|
"key_inferred_projects": [],
|
|
"leadership_changes": [
|
|
d for d in recent_developments if d["category"] == "leadership_change"
|
|
],
|
|
"hiring_signals": hiring_signals,
|
|
"technology_signals": [],
|
|
"patent_signals": [],
|
|
"manufacturing_and_expansion_signals": [],
|
|
"partnerships_and_acquisitions": [],
|
|
"financial_signals": [d for d in recent_developments if d["category"] == "filing_new"],
|
|
"regulatory_and_legal_signals": [],
|
|
"customer_sentiment": "Insufficient evidence to assess customer sentiment.",
|
|
"competitor_comparison": "Insufficient evidence to compare against competitors.",
|
|
"swot": {"strengths": [], "weaknesses": [], "opportunities": [], "threats": []},
|
|
"risks": [],
|
|
"opportunities": [],
|
|
"unknowns_and_missing_data": unknowns,
|
|
"monitoring_recommendations": ["Continue monitoring configured sources on schedule."],
|
|
"methodology": (
|
|
f"Generated by the mock LLM provider from {len(documents)} stored source "
|
|
f"document(s) and {len(changes)} deterministic change-detection result(s). No "
|
|
"external model was called."
|
|
),
|
|
"limitations": (
|
|
"Generated by the deterministic mock provider, not a live LLM. Set "
|
|
"LLM_PROVIDER=anthropic or LLM_PROVIDER=ollama for narrative synthesis."
|
|
),
|
|
}
|
|
|
|
|
|
def _build_change_significance(evidence: dict[str, Any]) -> dict[str, Any]:
|
|
severity = evidence.get("deterministic_severity") or "low"
|
|
confidence = evidence.get("deterministic_confidence") or 0.5
|
|
is_meaningful = severity in ("critical", "high", "medium")
|
|
change_type = (evidence.get("change_type") or "change").replace("_", " ")
|
|
return {
|
|
"is_real_change": True,
|
|
"is_meaningful": is_meaningful,
|
|
"why_it_matters": (
|
|
f"Deterministic scoring classified this {change_type} as {severity} severity "
|
|
f"with {confidence:.0%} confidence."
|
|
),
|
|
"confidence": confidence,
|
|
"should_notify": is_meaningful,
|
|
}
|
|
|
|
|
|
def _build_alert_summary(evidence: dict[str, Any]) -> dict[str, Any]:
|
|
company = evidence.get("company_name") or "The company"
|
|
change_type = (evidence.get("change_type") or "change").replace("_", " ")
|
|
severity = evidence.get("severity") or "medium"
|
|
confidence = evidence.get("confidence") or 0.5
|
|
return {
|
|
"title": f"{company}: {change_type} detected"[:100],
|
|
"summary": evidence.get("change_summary") or f"A {change_type} was detected for {company}.",
|
|
"why_it_matters": f"Classified as {severity} severity with {confidence:.0%} confidence.",
|
|
}
|
|
|
|
|
|
_HQ_RE = re.compile(r"(?:headquartered|based) in ([A-Z][\w\s,]{2,60}?)(?:[.\n]|$)", re.IGNORECASE)
|
|
_FORMERLY_RE = re.compile(
|
|
r"formerly (?:known as|named) ([A-Z][\w&\s]{2,60}?)(?:[.,\n]|$)", re.IGNORECASE
|
|
)
|
|
|
|
|
|
def _build_company_profile(evidence: dict[str, Any]) -> dict[str, Any]:
|
|
"""Mirrors app/change_detection/extractors.py's philosophy: cheap,
|
|
deterministic regex heuristics over real evidence text, never a
|
|
fabricated guess. Most fields (industry/country/region/public
|
|
identifiers) stay empty since a name-only mock search has no real
|
|
signal for them - see search/mock.py's docstring for why that's
|
|
intentional, not a gap."""
|
|
homepage_text = evidence.get("homepage_text") or ""
|
|
search_results = evidence.get("search_results") or []
|
|
combined_text = homepage_text + "\n" + "\n".join(r.get("snippet", "") for r in search_results)
|
|
|
|
hq_match = _HQ_RE.search(combined_text)
|
|
headquarters = hq_match.group(1).strip() if hq_match else None
|
|
|
|
alias_match = _FORMERLY_RE.search(combined_text)
|
|
aliases = [alias_match.group(1).strip()] if alias_match else []
|
|
|
|
# First real sentence of the fetched homepage, if any - an honest,
|
|
# evidence-derived summary rather than a fabricated one.
|
|
first_sentence = re.split(r"(?<=[.!?])\s", homepage_text.strip(), maxsplit=1)[0].strip()
|
|
description = first_sentence[:280] if first_sentence and len(first_sentence) > 15 else None
|
|
|
|
return {
|
|
"description": description,
|
|
"industry": None,
|
|
"country": None,
|
|
"region": None,
|
|
"headquarters": headquarters,
|
|
"aliases": aliases,
|
|
"competitors": [],
|
|
"public_identifiers": [],
|
|
}
|
|
|
|
|
|
_BUILDERS = {
|
|
RelevanceAssessment: _build_relevance,
|
|
ExtractionResult: _build_extraction,
|
|
SynthesisResult: _build_synthesis,
|
|
ReportContent: _build_report,
|
|
ChangeSignificanceAssessment: _build_change_significance,
|
|
AlertSummary: _build_alert_summary,
|
|
CompanyProfileExtraction: _build_company_profile,
|
|
}
|
|
|
|
|
|
class MockLLMProvider:
|
|
provider_name = "mock"
|
|
|
|
async def generate_structured[T: BaseModel](
|
|
self, system_prompt: str, user_prompt: str, response_model: type[T]
|
|
) -> T:
|
|
evidence = extract_evidence_block(user_prompt)
|
|
builder = _BUILDERS.get(response_model)
|
|
data = builder(evidence) if builder is not None else {}
|
|
return response_model.model_validate(data)
|
|
|
|
async def generate_text(self, system_prompt: str, user_prompt: str) -> str:
|
|
evidence = extract_evidence_block(user_prompt)
|
|
return (
|
|
"[mock provider] No live LLM configured. "
|
|
f"{len(evidence)} evidence field(s) were provided for this request."
|
|
)
|