Coverage for astrocyte/pipeline/directive_compile.py: 21%
90 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-10-29 22:10 +0000
1"""M17 follow-up — directive-style preference compile.
3Distils raw preference-type facts into a small set of imperative
4directives stored as :class:`MentalModel` rows with ``kind="directive"``.
5Hindsight-parity: their ``mental_models.subtype='directive'`` rows are
6user-curated hard rules that the answerer treats as authoritative;
7auto-extracted preferences (subtype='preference') are advisory and
8rarely surfaced verbatim in the system prompt.
10Why a separate "directive" tier rather than re-enabling the M14.6
11preference surface that M14.7 reverted:
13- M14.6 surfaced every consolidated preference (typically 8-12 rows
14 per document) on EVERY question. That bloated the system prompt and
15 caused cross-category regression on LME (single-session-user dropped
16 5/5 → 1/5 as the answerer was overwhelmed by personalization context
17 on non-preference questions).
18- Hindsight's fix is to keep the consolidation pass (subtype='preference',
19 advisory) AND add a tight "directive" tier capped at 3-5 rows of the
20 most-confident, most-stable preferences. Only the directive tier
21 surfaces verbatim. Result: the answerer gets sparse, high-confidence
22 hard rules, and the verbose preference tier stays in the store for
23 on-demand recall.
25Algorithm (one LLM call per document):
271. Pull all ``fact_type='preference'`` PageIndexFacts for the document.
282. If fewer than 2 preference facts, skip (no consolidation signal —
29 single-mention preferences are too noisy to promote to directives).
303. Send to an LLM with a prompt asking for **up to 5** imperative
31 directives, each with title + content phrased as a short rule the
32 answerer can follow verbatim. The prompt explicitly rejects vague
33 or single-mention preferences and asks the model to abstain when no
34 stable signal exists.
354. Save each as a :class:`MentalModel` with ``kind='directive'``,
36 ``scope=f'document:{document_id}'``. Idempotent — skip when the
37 document already has directive models.
39The bench's :meth:`AstrocyteClient.get_user_profile` surfaces both
40``kind='general'`` and ``kind='directive'`` models in the
41``## User Profile`` block. ``kind='preference'`` rows remain in the
42store but are NOT surfaced in the profile (M14.7 revert preserved).
44Cost: 1 LLM call per document (gpt-4o-mini), ~$0.01. Marginal.
46See:
47- docs/_design/m17-pageindex-ingestion.md (Conversation-Engine bench)
48- ``astrocyte.pipeline.preference_compile`` — sibling that produces the
49 larger ``kind='preference'`` pool this directive pass distils from
50"""
52from __future__ import annotations
54import json
55import logging
56import re
57from datetime import datetime, timezone
58from typing import TYPE_CHECKING
60from astrocyte.pipeline._json_tolerant import tolerant_json_loads_or_raise
61from astrocyte.types import MentalModel, Message
63if TYPE_CHECKING:
64 from astrocyte.provider import LLMProvider, MentalModelStore
65 from astrocyte.types import PageIndexFact
67_logger = logging.getLogger("astrocyte.pipeline.directive_compile")
70_COMPILE_PROMPT = """\
71You are extracting a SMALL SET of HARD-RULE DIRECTIVES from a user's \
72stated preferences in a conversation transcript. The directives will be \
73injected verbatim into an answering assistant's system prompt — treat \
74them as authoritative rules the assistant must honour when answering \
75follow-up questions.
77Output a JSON object with one key, ``directives``, containing an array \
78of AT MOST 5 directive objects. Each directive has:
79- "title": 2-5 word noun phrase naming the directive (e.g. \
80"Breakfast preference", "Dance studio")
81- "content": ONE imperative sentence the assistant can follow. \
82Use the form "Prefer X over Y because Z." or "Avoid X; the user dislikes \
83it." or "Recommend X for context Z." Always include the specific entity \
84the user named. Keep under 25 words.
85- "source_fact_ids": array of strings — the contributing raw fact_ids.
87STRICT RULES:
88- Emit AT MOST 5 directives. Quality over quantity. Sparse is better \
89than verbose: surface only the most CONFIDENT, STABLE preferences.
90- {min_facts_rule}
91- Phrase content imperatively (do/don't, prefer/avoid). The assistant \
92will read it as a rule.
93- DO NOT include fleeting / weak statements ("kind of liked it"). \
94Only stable preferences with explicit positive/negative sentiment.
95- {empty_input_rule}
97Input preferences (raw extractions):
98{prefs_block}
100OUTPUT MUST BE VALID JSON. No prose around it.
101"""
103_MIN_FACTS_RULE_MULTI = (
104 "A directive must be backed by AT LEAST 2 distinct raw preference "
105 "facts in the input. Drop single-mention preferences entirely."
106)
107_MIN_FACTS_RULE_SINGLE = (
108 "This document has only ONE session — single-mention preferences "
109 "are admissible because no repeat signal is possible. Emit a "
110 "directive for any clearly-stated preference even if it appears "
111 "only once, as long as it carries explicit positive/negative "
112 'sentiment (e.g. "no screens after 9:30pm", "avoid Y", '
113 '"prefer X"). Skip only vague / fleeting statements.'
114)
115_EMPTY_RULE_MULTI = (
116 "If the input has fewer than 2 distinct stable preferences, "
117 'return ``{{"directives": []}}``. Returning zero is correct '
118 "when the signal is absent."
119)
120_EMPTY_RULE_SINGLE = (
121 "If the input has no clearly-stated preferences, return "
122 '``{{"directives": []}}``. Returning zero is correct when the '
123 "signal is absent."
124)
127_MAX_DIRECTIVES = 5
128_MIN_FACTS_TO_COMPILE = 2
131def _slugify(text: str) -> str:
132 s = text.lower().strip()
133 s = re.sub(r"[^a-z0-9\s-]", "", s)
134 s = re.sub(r"\s+", "-", s)
135 return s[:60] or "dir"
138def _format_pref_for_prompt(fact: PageIndexFact) -> str:
139 parts = [f"id={fact.id}"]
140 if fact.entities:
141 parts.append(f"entities={','.join(fact.entities[:6])}")
142 return f"[{', '.join(parts)}] {fact.text}"
145async def compile_directives_for_document(
146 *,
147 mental_model_store: MentalModelStore,
148 bank_id: str,
149 document_id: str,
150 facts: list[PageIndexFact],
151 provider: LLMProvider,
152 model: str | None = None,
153 max_directives: int = _MAX_DIRECTIVES,
154 n_sessions: int | None = None,
155) -> list[str]:
156 """Consolidate preference facts → at most ``max_directives`` directive
157 mental models. Idempotent: skips when directive models already exist
158 for the document.
160 ``n_sessions`` is the document's session count (caller-supplied so
161 we don't need a store round-trip). When ``n_sessions == 1`` we lower
162 the minimum-facts threshold to 1: in a single-session document, the
163 ≥2-distinct-facts heuristic silently drops every single-mention
164 preference (e.g. "no screens after 9:30pm" said once), which is the
165 only signal we have for that document. The cross-document corpus
166 that motivated the ≥2 threshold doesn't apply when there IS only
167 one session.
169 Returns the list of ``model_id`` values for rows persisted (or
170 already-present on repeat call).
171 """
172 pref_facts = [f for f in facts if f.fact_type == "preference"]
173 # Fix 2 (conv-run-4): single-session docs lower the threshold to 1
174 # so single-mention preferences make it to a directive instead of
175 # being dropped silently. The ≥2 threshold is a noise-reduction
176 # heuristic for multi-session corpora — irrelevant when only one
177 # session's worth of preferences exists for this document.
178 is_single_session = n_sessions is not None and n_sessions <= 1
179 min_threshold = 1 if is_single_session else _MIN_FACTS_TO_COMPILE
180 if len(pref_facts) < min_threshold:
181 _logger.debug(
182 "directive_compile: doc=%s has %d preference facts (<%d, single_session=%s), skip",
183 document_id,
184 len(pref_facts),
185 min_threshold,
186 is_single_session,
187 )
188 return []
190 # Idempotency — skip if directives already exist for this document scope.
191 try:
192 existing = await mental_model_store.list(
193 bank_id,
194 scope=f"document:{document_id}",
195 kind="directive",
196 )
197 except TypeError:
198 all_for_scope = await mental_model_store.list(
199 bank_id,
200 scope=f"document:{document_id}",
201 )
202 existing = [m for m in all_for_scope if getattr(m, "kind", None) == "directive"]
203 if existing:
204 return [m.model_id for m in existing]
206 prefs_block = "\n".join(_format_pref_for_prompt(f) for f in pref_facts)
207 msg = _COMPILE_PROMPT.format(
208 prefs_block=prefs_block,
209 min_facts_rule=(_MIN_FACTS_RULE_SINGLE if is_single_session else _MIN_FACTS_RULE_MULTI),
210 empty_input_rule=(_EMPTY_RULE_SINGLE if is_single_session else _EMPTY_RULE_MULTI),
211 )
213 try:
214 completion = await provider.complete(
215 [Message(role="user", content=msg)],
216 model=model,
217 max_tokens=1200,
218 temperature=0.0,
219 response_format={"type": "json_object"},
220 )
221 except Exception as exc: # noqa: BLE001
222 _logger.warning(
223 "directive_compile.llm: call failed doc=%s (%s)",
224 document_id,
225 exc,
226 )
227 return []
229 try:
230 data = tolerant_json_loads_or_raise(completion.text)
231 except (json.JSONDecodeError, AttributeError) as exc:
232 _logger.warning(
233 "directive_compile.parse: bad JSON doc=%s (%s) text=%r",
234 document_id,
235 exc,
236 getattr(completion, "text", "")[:200],
237 )
238 return []
240 items = data.get("directives") or []
241 if not isinstance(items, list):
242 return []
244 now = datetime.now(tz=timezone.utc)
245 scope = f"document:{document_id}"
246 saved: list[str] = []
247 seen_ids: set[str] = set()
249 # M40 — index source facts so MM construction can attach per-source
250 # evidence timestamps in conversation time (mirrors preference_compile).
251 _fact_by_id = {getattr(f, "fact_id", None): f for f in pref_facts}
253 def _ts_for_fact_id(fid: str) -> datetime:
254 f = _fact_by_id.get(fid)
255 if f is None:
256 return now
257 return getattr(f, "mentioned_at", None) or getattr(f, "occurred_start", None) or now
259 for raw in items[:max_directives]:
260 if not isinstance(raw, dict):
261 continue
262 title = str(raw.get("title", "")).strip()
263 content = str(raw.get("content", "")).strip()
264 if not title or not content:
265 continue
266 source_fact_ids = raw.get("source_fact_ids") or []
267 if not isinstance(source_fact_ids, list):
268 source_fact_ids = []
269 source_fact_ids = [str(s) for s in source_fact_ids if isinstance(s, (str, int))]
270 slug = _slugify(title)
271 model_id = f"dir:{document_id[:8]}:{slug}"
272 if model_id in seen_ids:
273 continue
274 seen_ids.add(model_id)
276 source_timestamps = [_ts_for_fact_id(sid) for sid in source_fact_ids]
278 mm = MentalModel(
279 model_id=model_id,
280 bank_id=bank_id,
281 title=title,
282 content=content,
283 scope=scope,
284 source_ids=source_fact_ids,
285 revision=1,
286 refreshed_at=now,
287 kind="directive",
288 source_timestamps=source_timestamps,
289 )
290 try:
291 await mental_model_store.upsert(mm, bank_id)
292 saved.append(model_id)
293 except Exception as exc: # noqa: BLE001
294 _logger.warning(
295 "directive_compile.save: doc=%s id=%s failed (%s)",
296 document_id,
297 model_id,
298 exc,
299 )
301 _logger.info(
302 "directive_compile: doc=%s produced %d directives from %d preference facts",
303 document_id,
304 len(saved),
305 len(pref_facts),
306 )
307 return saved