Spaces:
Sleeping
Sleeping
File size: 7,051 Bytes
2e818da | 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 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 | from __future__ import annotations
import asyncio
import logging
import os
from app.schemas.visual_lesson import EvidenceClaim, EvidenceSource, VisualizationRequest
from app.services.visual_evidence import VisualEvidenceResolver, _clean
CANONICAL_NUCLEIC_SOURCES = [
EvidenceSource(
source_id="builtin:nucleic-chemistry",
origin="builtin",
title="NCBI Bookshelf: Molecular Biology of the Cell — DNA structure",
url="https://www.ncbi.nlm.nih.gov/books/NBK26821/",
excerpt="Canonical reference metadata for nucleotides, antiparallel strands, complementary base pairing, and the sugar-phosphate backbone.",
authority="canonical",
),
EvidenceSource(
source_id="builtin:nucleic-forms",
origin="builtin",
title="RCSB PDB-101: Nucleic acid structure",
url="https://pdb101.rcsb.org/learn/guide-to-understanding-pdb-data/introduction",
excerpt="Canonical reference metadata for archived nucleic-acid structures and common DNA/RNA structural representations.",
authority="canonical",
),
]
logger = logging.getLogger(__name__)
class NucleicEvidenceResolver:
"""Resolve only context that can affect the requested nucleic visualization."""
def __init__(self, base: VisualEvidenceResolver | None = None) -> None:
self.base = base or VisualEvidenceResolver()
@staticmethod
def _needs_context_research(request: VisualizationRequest) -> bool:
prompt_text = request.prompt.lower()
explicit_context_request = any(
term in prompt_text
for term in ("paper", "selected", "according to", "reported", "study", "figure", "this molecule", "this structure")
)
# Surrounding PDF text is transport context, not an instruction to use
# the active paper. Named canonical molecules must not fan out into an
# unrelated project merely because a document is open.
named_canonical = any(
term in prompt_text
for term in ("adenine", "cytosine", "guanine", "thymine", "uracil")
)
if named_canonical and not explicit_context_request:
return False
return bool(
explicit_context_request
or request.selection_text
or request.selection_snippets
or request.selection_image_base64
)
@staticmethod
def _dedupe(sources: list[EvidenceSource]) -> list[EvidenceSource]:
output: list[EvidenceSource] = []
seen: set[tuple[str, str]] = set()
limits = {"selection": 4, "project": 4, "web": 3, "builtin": 2, "database": 3}
counts: dict[str, int] = {}
for source in sources:
key = (source.source_id, source.excerpt[:240])
if key in seen or counts.get(source.origin, 0) >= limits[source.origin]:
continue
seen.add(key)
counts[source.origin] = counts.get(source.origin, 0) + 1
output.append(source)
return output
async def resolve_context(self, project_id: str, request: VisualizationRequest) -> tuple[list[EvidenceSource], list[EvidenceClaim], list[str]]:
warnings: list[str] = []
sources = self.base._selection_sources(request)
if self._needs_context_research(request):
queries = list(dict.fromkeys(filter(None, [
_clean(request.prompt),
_clean(request.selection_text),
"DNA RNA nucleotide nucleobase molecular structure",
])))
logger.info("Nucleic evidence contextual fan-out request=%s queries=%d", request.request_id, len(queries))
project_task = (
asyncio.create_task(self.base._project_search(project_id, queries, request.active_document_ids))
if request.project_context_enabled else None
)
image_task = asyncio.create_task(self.base._image_source(request))
web_rows: list[dict] = []
if os.getenv("TAVILY_API_KEY"):
try:
search = self.base._web_search or self.base._default_web_search
batches = await asyncio.gather(*(search(query) for query in queries[:2]), return_exceptions=True)
for batch in batches:
if isinstance(batch, list):
web_rows.extend(batch)
except Exception:
warnings.append("Live contextual research failed; canonical and official database evidence were still used.")
if project_task is not None:
try:
project_rows = await project_task
sources.extend(self.base._candidate_sources(project_rows, "project"))
except Exception:
warnings.append("Project context retrieval failed; canonical and official database evidence were still used.")
image_source = await image_task
if image_source:
sources.append(image_source)
sources.extend(self.base._candidate_sources(web_rows, "web"))
else:
logger.info(
"Nucleic evidence canonical-only request=%s active_documents=%d",
request.request_id,
len(request.active_document_ids),
)
sources.extend(CANONICAL_NUCLEIC_SOURCES)
sources = self._dedupe(sources)
claims = [
EvidenceClaim(
claim_id="nucleic-backbone",
text="Nucleic-acid strands have a repeating sugar-phosphate backbone and directional 5-prime and 3-prime ends.",
claim_type="standard_definition",
support_level="canonical",
source_ids=["builtin:nucleic-chemistry"],
),
EvidenceClaim(
claim_id="nucleic-pairing",
text="The visualization uses canonical A–T and G–C DNA pairing, or A–U and G–C RNA pairing.",
claim_type="standard_definition",
support_level="canonical",
source_ids=["builtin:nucleic-chemistry"],
),
EvidenceClaim(
claim_id="nucleic-forms",
text="A-, B-, and Z-DNA are shown with distinct canonical helix parameters; Z-DNA is left-handed.",
claim_type="standard_definition",
support_level="canonical",
source_ids=["builtin:nucleic-forms"],
),
]
# Retrieved passages remain planning context only. A literal selection
# anchor proves provenance, not that the passage supports a displayed
# molecular fact, so it is never promoted to a claim automatically.
logger.info(
"Nucleic evidence resolved request=%s sources=%d claims=%d warnings=%d",
request.request_id,
len(sources),
len(claims),
len(warnings),
)
return sources, claims, warnings
|