File size: 3,508 Bytes
1d9bd9b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """Stable, public fingerprint for persisted research-answer compatibility."""
from __future__ import annotations
import hashlib
import inspect
import json
from pathlib import Path
from typing import Any
def _hash(value: Any) -> str:
if not isinstance(value, str):
value = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(value.encode("utf-8")).hexdigest()[:20]
def _source(*values: Any) -> str:
parts = []
for value in values:
if isinstance(value, str):
parts.append(value)
continue
if inspect.ismodule(value):
try:
path = inspect.getsourcefile(value)
parts.append(Path(path).read_text(encoding="utf-8") if path else repr(value))
except (OSError, TypeError):
parts.append(repr(value))
continue
try:
parts.append(inspect.getsource(value))
except (OSError, TypeError):
parts.append(repr(value))
return "\n\n".join(parts)
def build_research_release(corpus: Any, agent_module: Any) -> dict[str, Any]:
"""Fingerprint every input that can make a saved research answer stale.
Prompt hashes are derived from their actual text and algorithm hashes from
their function source, so ordinary prompt/retrieval edits invalidate saved
answers without a developer remembering to bump a browser-session number.
"""
coverage = corpus.coverage()
manifest = getattr(corpus, "manifest", {}) or {}
corpus_descriptor = {
"release_version": coverage.get("release_version") or manifest.get("release_version") or "legacy",
"accepted_judgments": coverage.get("accepted_judgments"),
"units": coverage.get("units"),
"paragraphs": coverage.get("paragraphs"),
"artifacts": manifest.get("artifacts") or {},
"model": manifest.get("model") or {},
}
corpus_type = type(corpus)
# Full source digests deliberately over-invalidate rather than risk serving
# an answer produced by changed ranking or verification helpers that were
# not listed individually here.
agent_source = _source(agent_module)
corpus_source = _source(corpus_type)
components = {
"corpus": _hash(corpus_descriptor),
"query_router": _hash(_source(
getattr(agent_module, "QUERY_ROUTER_VERSION", ""),
getattr(agent_module, "QUERY_BRIEF_SYS", ""),
getattr(agent_module, "query_brief", None),
)),
"retrieval": _hash(_source(
getattr(agent_module, "RETRIEVAL_VERSION", ""),
agent_source,
corpus_source,
getattr(agent_module, "structured_search_stream", None),
getattr(agent_module, "judge", None),
getattr(corpus_type, "identity_hits", None),
getattr(corpus_type, "search_lanes", None),
)),
"answer": _hash(_source(
getattr(agent_module, "ANSWER_PROMPT_VERSION", ""),
agent_source,
getattr(agent_module, "_GROUND_SYS", ""),
getattr(agent_module, "ground", None),
getattr(agent_module, "verify_claims", None),
getattr(agent_module, "_grounded_answer_text", None),
)),
}
return {
"fingerprint": _hash(components),
"components": components,
"corpus_release": corpus_descriptor["release_version"],
}
__all__ = ["build_research_release"]
|