study-buddy / app /services /commit_candidate_adapters.py
GitHub Actions
deploy d092bea3608b7a29952f16357fda39b7a29e399b
2e818da
Raw
History Blame Contribute Delete
4.15 kB
"""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,
)