| """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) |
| |
| |
| |
| 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"] |
|
|
