Coverage for astrocyte/pipeline/preference_compile.py: 87%
83 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-10-29 22:10 +0000
1"""M14.6: consolidated user-preference compile.
3Distils raw preference-type facts into structured consolidated
4preferences, stored as :class:`MentalModel` rows with
5``kind="preference"``. Matches Hindsight's mental-models pattern for
6consolidated knowledge (their ``mental_models.subtype`` column).
8Why a separate compile pass rather than enriching raw extraction:
9- Hindsight tried the dedicated ``opinion`` fact_type and **removed
10 it** (migration g2h3i4j5k6l7 — April 2026). Their evidence: schema
11 rigidity at ingest didn't pan out; preferences are better handled
12 at the compiled-knowledge layer downstream of fact extraction.
13- Raw preference facts capture individual statements but lose the
14 conditional context (when/why/under-what-conditions) the LME
15 single-session-preference category probes. A consolidation pass
16 with an LLM-judged synthesis can preserve that context.
18Algorithm (one LLM call per document):
201. Pull all PageIndexFacts with ``fact_type='preference'`` for the doc.
212. If fewer than 2 preference facts, skip (no consolidation signal).
223. Send to LLM with a prompt asking for up to N consolidated
23 preferences, each with title + content (the content preserves
24 qualifier / condition / sentiment inline) + the source fact ids.
254. Save each as a :class:`MentalModel` with ``kind='preference'``,
26 ``scope=f'document:{document_id}'``. Idempotent — skip if the
27 document already has preference models (M14.6 v1; M14.7 may add
28 incremental refresh).
30The bench's existing :meth:`AstrocyteClient.get_user_profile` lists
31mental models by ``scope`` and serialises them into the answerer's
32``## User Profile`` block — preference-kind models surface there
33automatically. No retrieval-path change needed for v1.
35Cost: ~1 LLM call per document (gpt-4o-mini), ~$0.01-0.02. Marginal.
37See:
38- ``docs/_design/m13-m14-roadmap.md`` §4 (M14.6 placement)
39- ``astrocyte.pipeline.mental_model_compile`` — sibling that produces
40 ``kind='general'`` profile statements
41"""
43from __future__ import annotations
45import json
46import logging
47import re
48from datetime import datetime, timezone
49from typing import TYPE_CHECKING
51from astrocyte.pipeline._json_tolerant import tolerant_json_loads_or_raise
52from astrocyte.types import MentalModel, Message
54if TYPE_CHECKING:
55 from astrocyte.provider import LLMProvider, MentalModelStore
56 from astrocyte.types import PageIndexFact
58_logger = logging.getLogger("astrocyte.pipeline.preference_compile")
61_COMPILE_PROMPT = """\
62You are consolidating a user's STATED PREFERENCES from a conversation \
63transcript. You will be given the raw preference statements extracted \
64from the conversation. Your job: produce up to {max_prefs} consolidated \
65preferences that capture WHAT the user prefers, plus the CONTEXT \
66(when, why, under what conditions, sentiment strength).
68Output a JSON object with one key, ``preferences``, containing an array \
69of preference objects. Each preference has:
70- "title": 3-7 word noun phrase naming the preference dimension \
71 (e.g. "Breakfast preference", "Coffee in morning", "Reading material")
72- "content": ONE declarative sentence stating the user's preference \
73 with FULL CONTEXT inline. Capture: WHAT they prefer (the object), \
74 WHEN/WHERE (qualifier), WHY (if stated), and STRENGTH (strong / mild \
75 / context-dependent). Examples:
76 - "User prefers oatmeal with berries for breakfast on busy mornings, \
77 citing time efficiency."
78 - "User strongly prefers Dr. Patel's clinic over their previous \
79 clinic because the staff was more attentive."
80 - "User prefers Sony cameras for travel photography over Nikon, \
81 mainly for sensor performance in low light."
82- "source_fact_ids": array of strings — the raw fact_id values from \
83 the input that contributed to this consolidated preference.
85Rules:
86- DO NOT invent preferences not stated in the input. If the user only \
87 said "I like X", that's a preference; if they merely mentioned X, \
88 that's NOT a preference.
89- Merge multiple raw mentions of the SAME preference into one \
90 consolidated row. Cite all contributing fact_ids.
91- If two raw preferences CONTRADICT (e.g. "I used to prefer X, now I \
92 prefer Y"), produce ONE row reflecting the LATEST state, with \
93 "(previously X)" inline.
94- Skip vague / fleeting statements ("kind of liked it"). Only \
95 consolidate STABLE preferences.
96- If the input has fewer than 2 distinct preferences, return \
97 ``{{"preferences": []}}``.
99Input preferences (raw extractions):
100{prefs_block}
102OUTPUT MUST BE VALID JSON. No prose around it.
103"""
106def _slugify(text: str) -> str:
107 """3-7 word title → URL-safe slug. Same shape as section_compile."""
108 s = text.lower().strip()
109 s = re.sub(r"[^a-z0-9\s-]", "", s)
110 s = re.sub(r"\s+", "-", s)
111 return s[:60] or "pref"
114def _format_pref_for_prompt(fact: PageIndexFact) -> str:
115 """Render one preference fact for the consolidation prompt."""
116 parts = [f"id={fact.id}"]
117 if fact.entities:
118 parts.append(f"entities={','.join(fact.entities[:6])}")
119 return f"[{', '.join(parts)}] {fact.text}"
122async def compile_preferences_for_document(
123 *,
124 mental_model_store: MentalModelStore,
125 bank_id: str,
126 document_id: str,
127 facts: list[PageIndexFact],
128 provider: LLMProvider,
129 model: str | None = None,
130 max_preferences: int = 12,
131) -> list[str]:
132 """Consolidate preference-type facts → preference-kind MentalModels.
134 Args:
135 mental_model_store: where to persist the consolidated models.
136 bank_id: tenant scope.
137 document_id: scope for the resulting models
138 (``scope='document:<id>'``).
139 facts: ALL facts extracted for this document — the function
140 filters internally for ``fact_type=='preference'``. Passed
141 from the caller's retain path to avoid an extra round trip
142 to the store.
143 provider: LLM provider for the consolidation call.
144 model: LLM model name (defaults to provider's default —
145 typically gpt-4o-mini for our bench).
146 max_preferences: cap on consolidated rows produced. Same cap
147 shape as ``mental_model_compile``.
149 Returns:
150 List of ``model_id`` values for the rows actually persisted.
151 Empty list when there were insufficient preference facts or
152 the LLM produced no usable output.
154 Idempotent: if preference-kind models already exist for this
155 ``(bank_id, document_id)`` scope, skip the compile. Re-running on a
156 cached bank is therefore a no-op — fresh banks pay the LLM call.
157 """
158 pref_facts = [f for f in facts if f.fact_type == "preference"]
159 if len(pref_facts) < 2:
160 _logger.debug(
161 "preference_compile: doc=%s has %d preference facts (<2), skip",
162 document_id,
163 len(pref_facts),
164 )
165 return []
167 # Idempotency check — skip if we already have preference models for this scope.
168 try:
169 existing = await mental_model_store.list(
170 bank_id,
171 scope=f"document:{document_id}",
172 kind="preference",
173 )
174 except TypeError:
175 # Older MentalModelStore SPI without `kind` kwarg — fall back to
176 # listing all and filtering client-side. Lets this module work
177 # against not-yet-migrated stores (some tests use older fakes).
178 all_for_scope = await mental_model_store.list(
179 bank_id,
180 scope=f"document:{document_id}",
181 )
182 existing = [m for m in all_for_scope if m.kind == "preference"]
183 if existing:
184 _logger.debug(
185 "preference_compile: doc=%s already has %d preference models — skip",
186 document_id,
187 len(existing),
188 )
189 return [m.model_id for m in existing]
191 prefs_block = "\n".join(_format_pref_for_prompt(f) for f in pref_facts)
192 msg = _COMPILE_PROMPT.format(
193 max_prefs=max_preferences,
194 prefs_block=prefs_block,
195 )
197 try:
198 # max_tokens=800 (M14.6) truncated JSON output mid-string on ~90% of
199 # LME documents (M14.7-b1.1 diagnostic 2026-05-13). At
200 # max_preferences=12, each preference body runs ~150-200 tokens of
201 # structured JSON, so 800 was undersized by ~2.5×. The truncation
202 # caused entire documents to land with 0 preference-models, which
203 # silently neutered the B-1 anchor pool for downstream questions.
204 # Bumped to 3000 with margin for ``source_fact_ids`` arrays.
205 completion = await provider.complete(
206 [Message(role="user", content=msg)],
207 model=model,
208 max_tokens=3000,
209 temperature=0.0,
210 response_format={"type": "json_object"},
211 )
212 except Exception as exc: # noqa: BLE001
213 _logger.warning(
214 "preference_compile.llm: call failed for doc=%s (%s)",
215 document_id,
216 exc,
217 )
218 return []
220 try:
221 data = tolerant_json_loads_or_raise(completion.text)
222 except (json.JSONDecodeError, AttributeError) as exc:
223 _logger.warning(
224 "preference_compile.parse: bad JSON for doc=%s (%s) text=%r",
225 document_id,
226 exc,
227 getattr(completion, "text", "")[:200],
228 )
229 return []
231 items = data.get("preferences") or []
232 if not isinstance(items, list):
233 return []
235 now = datetime.now(tz=timezone.utc)
236 scope = f"document:{document_id}"
237 saved: list[str] = []
238 seen_ids: set[str] = set()
240 # M40 — index source facts by fact_id so we can pull per-source
241 # evidence timestamps when building each preference's MentalModel.
242 # Preference order: mentioned_at (when the user said it) > occurred_start
243 # (when the event occurred) > now (last resort — only for facts that
244 # carry no temporal info at all, which would trend NEW from any
245 # reference_date in the conversation timeline).
246 _fact_by_id = {getattr(f, "fact_id", None): f for f in pref_facts}
248 def _ts_for_fact_id(fid: str) -> datetime:
249 f = _fact_by_id.get(fid)
250 if f is None:
251 return now
252 return getattr(f, "mentioned_at", None) or getattr(f, "occurred_start", None) or now
254 for raw in items[:max_preferences]:
255 if not isinstance(raw, dict):
256 continue
257 title = str(raw.get("title", "")).strip()
258 content = str(raw.get("content", "")).strip()
259 if not title or not content:
260 continue
261 source_fact_ids = raw.get("source_fact_ids") or []
262 if not isinstance(source_fact_ids, list):
263 source_fact_ids = []
264 source_fact_ids = [str(s) for s in source_fact_ids if isinstance(s, (str, int))]
265 # Stable model_id: pref:<doc_short>:<title-slug>
266 slug = _slugify(title)
267 model_id = f"pref:{document_id[:8]}:{slug}"
268 if model_id in seen_ids:
269 continue
270 seen_ids.add(model_id)
272 # M40 — build positionally-aligned timestamp list (one per source_id).
273 source_timestamps = [_ts_for_fact_id(sid) for sid in source_fact_ids]
275 mm = MentalModel(
276 model_id=model_id,
277 bank_id=bank_id,
278 title=title,
279 content=content,
280 scope=scope,
281 source_ids=source_fact_ids,
282 revision=1,
283 refreshed_at=now,
284 kind="preference",
285 source_timestamps=source_timestamps,
286 )
287 try:
288 await mental_model_store.upsert(mm, bank_id)
289 saved.append(model_id)
290 except Exception as exc: # noqa: BLE001
291 _logger.warning(
292 "preference_compile.save: doc=%s id=%s failed (%s)",
293 document_id,
294 model_id,
295 exc,
296 )
298 _logger.info(
299 "preference_compile: doc=%s consolidated %d preferences from %d raw facts",
300 document_id,
301 len(saved),
302 len(pref_facts),
303 )
304 return saved