add evidence-grounded ai assistance

apps/api/app/ai_assistance.py

@@ -0,0 +1,93 @@
+"""Evidence-bounded AI assistance primitives for Phase 13.
+
+This module deliberately has no network or model dependency.  The local provider is
+an auditable formatter over stored records; other providers are reported as
+not_configured rather than guessed at.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+from typing import Any
+
+MAX_INPUT_ITEMS = 100
+MAX_FIELD_CHARS = 500
+MAX_OUTPUT_CHARS = 12_000
+SUPPORTED_KINDS = ("summary", "qualification_explanation", "missing_data_questions", "research_note")
+_SECRET_RE = re.compile(r"(?i)(password|passwd|secret|token|api[_-]?key|authorization|private[_-]?key|credential)\s*[:=]\s*[^\s,;]+")
+
+
+def _text(value: Any, limit: int = MAX_FIELD_CHARS) -> str:
+    value = "" if value is None else str(value)
+    value = _SECRET_RE.sub(r"\1: [REDACTED]", value)
+    return value[:limit]
+
+
+def redact(value: Any) -> Any:
+    if isinstance(value, dict):
+        return {str(k)[:80]: ("[REDACTED]" if re.search(r"(?i)(password|passwd|secret|token|api[_-]?key|authorization|private[_-]?key|credential)", str(k)) else redact(v)) for k, v in list(value.items())[:100]}
+    if isinstance(value, list):
+        return [redact(v) for v in value[:MAX_INPUT_ITEMS]]
+    if isinstance(value, str):
+        return _text(value)
+    return value
+
+
+def evidence_hashes(evidence: list[dict[str, Any]]) -> list[str]:
+    return [hashlib.sha256(json.dumps(redact(item), sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest() for item in evidence]
+
+
+def _citation(item: dict[str, Any]) -> dict[str, Any]:
+    return {"evidence_id": int(item["id"]), "kind": _text(item.get("kind", "evidence"), 80), "url": _text(item.get("url", ""), 500)}
+
+
+def _claim(item: dict[str, Any]) -> str:
+    return _text(item.get("claim", ""), MAX_FIELD_CHARS).strip()
+
+
+def build_local_suggestions(business: dict[str, Any], scans: list[dict[str, Any]], contacts: list[dict[str, Any]], evidence: list[dict[str, Any]], score_history: list[dict[str, Any]] | None = None) -> dict[str, Any]:
+    """Create deterministic suggestions using only supplied stored data.
+
+    Every claim-bearing item cites one or more rows from ``evidence``.  No
+    contact details are emitted, and contacts are used only as aggregate counts.
+    """
+    evidence = [redact(x) for x in evidence[:MAX_INPUT_ITEMS] if _claim(x)]
+    citations = [_citation(x) for x in evidence]
+    claims = [_claim(x) for x in evidence]
+    suggestions: list[dict[str, Any]] = []
+    name = _text(business.get("name", "this business"), 200)
+    if claims:
+        joined = " ".join(f"{claim} [evidence:{item['id']}]" for claim, item in zip(claims[:5], evidence[:5]))
+        suggestions.append({"type": "summary", "text": f"Stored evidence for {name}: {joined}", "citations": citations[:5]})
+        score = business.get("score")
+        if score is not None:
+            suggestions.append({"type": "qualification_explanation", "text": f"The stored qualification score is {_text(score, 30)}; review the cited evidence before relying on it. [evidence:{evidence[0]['id']}]", "citations": citations[:1]})
+        missing = []
+        if not _text(business.get("website", "")).strip(): missing.append("official website")
+        if not contacts: missing.append("public contact evidence")
+        if missing:
+            suggestions.append({"type": "missing_data_questions", "text": "Confirm whether the following data is available: " + ", ".join(missing) + f". [evidence:{evidence[0]['id']}]", "citations": citations[:1]})
+        suggestions.append({"type": "research_note", "text": f"Draft note: independently verify the stored claims for {name}; do not infer facts beyond the cited records. [evidence:{evidence[0]['id']}]", "citations": citations[:1]})
+    else:
+        # No claim is fabricated.  A question is safe but has no citation, so
+        # return no suggestions and let the caller expose the missing-data state.
+        suggestions = []
+    output = {"provider": "local", "version": "deterministic-v1", "suggestions": suggestions, "grounded": True, "claim_policy": "stored_evidence_only"}
+    encoded = json.dumps(output, sort_keys=True, ensure_ascii=False)
+    return json.loads(encoded[:MAX_OUTPUT_CHARS]) if len(encoded) <= MAX_OUTPUT_CHARS else {"provider": "local", "version": "deterministic-v1", "suggestions": suggestions[:1], "grounded": True, "claim_policy": "stored_evidence_only"}
+
+
+def provider_name() -> str | None:
+    value = os.environ.get("AI_PROVIDER", "").strip().lower()
+    return value or None
+
+
+def generate(business: dict[str, Any], scans: list[dict[str, Any]], contacts: list[dict[str, Any]], evidence: list[dict[str, Any]], score_history: list[dict[str, Any]] | None = None) -> tuple[str, str, str, dict[str, Any]]:
+    provider = provider_name()
+    hashes = evidence_hashes(evidence)
+    metadata = {"input_counts": {"business": 1, "scans": min(len(scans), MAX_INPUT_ITEMS), "contacts": min(len(contacts), MAX_INPUT_ITEMS), "evidence": min(len(evidence), MAX_INPUT_ITEMS)}, "redacted": True, "max_input_items": MAX_INPUT_ITEMS, "max_field_chars": MAX_FIELD_CHARS, "evidence_hashes": hashes}
+    if provider not in {"local", "deterministic"}:
+        return "not_configured", provider or "", "", metadata
+    return "succeeded", "local", "deterministic-v1", {**metadata, "output": build_local_suggestions(business, scans, contacts, evidence, score_history)}