Spaces:
Running
Running
File size: 4,147 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 | """Adapters from CommitAdaptationReview proposals to the real, strict
MemoryCandidate / ProjectMemoryRecord models. The LLM never authors
candidate_id, recurrence_key, novelty_key, or authoritative attribution —
all of that is derived here in application code. See
docs/commit-memory-adaptation-architecture.md for the full rationale.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Literal
from app.rag.models import normalize_evidence_text
from app.rag.project_memory import ProjectMemoryRecord, make_project_memory_id
from app.schemas.commit_adaptation import MemoryProposal
from app.services.commit_signal_patterns import is_explicit_global_signal, is_explicit_student_signal
from app.services.memory_candidates import MemoryCandidate, make_candidate_id
@dataclass(frozen=True)
class StudentInteraction:
interaction_id: str
timestamp: float
actor: Literal["student", "assistant"]
source: str
agent_id: str
text: str
def _normalize_for_quote_match(text: str) -> str:
# normalize_evidence_text() does NFKC + whitespace collapse, but not
# casefold or smart-quote mapping — layer those on top rather than
# duplicating the whole normalizer.
normalized = normalize_evidence_text(text)
normalized = normalized.replace("‘", "'").replace("’", "'")
normalized = normalized.replace("“", '"').replace("”", '"')
return normalized.casefold()
def proposal_to_memory_candidate(
proposal: MemoryProposal,
*,
project_id: str,
interactions_by_id: dict[str, StudentInteraction],
) -> MemoryCandidate:
verified_quotes = []
for evidence in proposal.evidence_quotes:
interaction = interactions_by_id.get(evidence.interaction_id)
if interaction is None or interaction.actor != "student":
continue
if _normalize_for_quote_match(evidence.quote) not in _normalize_for_quote_match(interaction.text):
continue
verified_quotes.append(evidence)
verified_interaction_ids = sorted({evidence.interaction_id for evidence in verified_quotes})
explicitly_global = any(is_explicit_global_signal(evidence.quote) for evidence in verified_quotes)
# A global-scoped signal ("I generally prefer...") is a stronger form of
# explicit wording, not a separate category — it must also count toward
# `explicit`, or global-but-not-matching-the-narrower-explicit-patterns
# statements fall through to the weakest, unattributed bucket.
explicit = explicitly_global or any(is_explicit_student_signal(evidence.quote) for evidence in verified_quotes)
destination = proposal.destination
if destination == "student" and explicit and not explicitly_global:
destination = "project"
if explicit:
attribution = "explicit_student"
elif destination == "project":
attribution = "idea_observer_interaction"
else:
attribution = "idea_observer_profile_proposal"
return MemoryCandidate(
candidate_id=make_candidate_id(destination, project_id, proposal.kind, proposal.statement),
destination=destination,
project_id=project_id,
kind=proposal.kind,
statement=proposal.statement,
attribution=attribution,
confidence=proposal.confidence,
interaction_ids=verified_interaction_ids,
evidence_ids=[],
)
def candidate_to_project_memory_record(candidate: MemoryCandidate) -> ProjectMemoryRecord:
if candidate.destination != "project":
raise ValueError(f"Expected destination='project', got {candidate.destination!r}")
return ProjectMemoryRecord(
memory_id=make_project_memory_id(
candidate.project_id, candidate.kind, candidate.recurrence_key, candidate.statement,
),
project_id=candidate.project_id,
kind=candidate.kind,
statement=candidate.statement,
novelty_key=candidate.recurrence_key,
attribution=candidate.attribution,
confidence=candidate.confidence,
evidence_ids=candidate.evidence_ids,
interaction_ids=candidate.interaction_ids,
)
|