Spaces:
Sleeping
Sleeping
File size: 7,283 Bytes
ce45eb0 | 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 155 156 157 158 | """HomePilot adapter (MVP) — the first real consumer.
HomePilot keeps two distinctions that already work, and this adapter preserves
both rather than replacing HomePilot's memory:
* **Profile vs. learned memory.** Profile fields (persona identity, preferences,
device facts) map to the ``profile`` expert and are *always injectable* — they
ride along regardless of the routing decision. Persona memories map to the
``semantic`` / ``episodic`` experts and are recalled by routing.
* **Two engines.** ``Mode.BASIC`` is deterministic: a TTL on each memory and a
per-scope cap (newest-wins). ``Mode.ADAPTIVE`` leans on the engine's recency
decay and importance weighting (the assembler already scores both);
consolidation is deferred to the v1 lifecycle module.
Compact-injection discipline is respected: :meth:`build_pack` defaults to a low
hundreds-of-tokens budget. Device / persona live state is exposed through a
``live_state`` expert seam that activates when the live-state expert lands in v1.
The framing is that HomePilot treats Matrix Context as an externalized,
inspectable context plane it can hand off to other agents or reach over MCP —
not a black box that replaces what already works. Every selection is explainable
via :meth:`explain` (``inspect()``).
"""
from __future__ import annotations
from enum import Enum
from typing import Dict, Iterable, List
from ..manager import ContextManager
from ..schema.pack import ContextPack
# Profile is always injectable; live_state is the v1 seam for device/persona state.
PINNED_EXPERTS = ("profile",)
LIVE_STATE_EXPERT = "live_state" # [v1] activates when the live-state expert lands
# Profile-shaped persona fields are identity, not learned memory.
PROFILE_FIELDS = ("id", "label", "category", "psychology_approach",
"name", "timezone", "location", "language")
class Mode(str, Enum):
BASIC = "basic" # deterministic: TTL + per-scope cap
ADAPTIVE = "adaptive" # recency decay + importance weighting
class HomePilotAdapter:
def __init__(self, ctx: ContextManager, scope: str = "/homepilot",
mode: Mode = Mode.ADAPTIVE,
basic_ttl_seconds: float = 7 * 86400.0,
basic_cap: int = 200):
self.ctx = ctx
self.scope = scope
self.mode = Mode(mode)
self.basic_ttl_seconds = basic_ttl_seconds
self.basic_cap = basic_cap
# ----------------------------------------------------------------- scopes
def _scope(self, kind: str) -> str:
return f"{self.scope}/{kind}"
# -------------------------------------------------------------- profile in
def load_profile(self, profile: Dict[str, str]) -> int:
"""Map profile fields -> profile expert (always-injectable identity)."""
n = 0
for key, value in profile.items():
if value is None:
continue
self.ctx.remember(f"{key}: {value}", expert="profile",
scope=self._scope("profile"), importance=0.9)
n += 1
return n
def load_persona(self, persona: Dict) -> Dict[str, int]:
"""Split a HomePilot persona definition into profile + learned memory.
Profile-shaped fields become always-injectable profile facts; the rest
of the persona's descriptive text (system prompt, techniques, behaviors)
becomes semantic memory recalled by routing.
"""
prof = {k: persona[k] for k in PROFILE_FIELDS if persona.get(k)}
counts = {"profile": self.load_profile(prof), "memory": 0}
memories: List[str] = []
if persona.get("system_prompt"):
memories.append(f"persona style: {persona['system_prompt']}")
for k in ("key_techniques", "unique_behaviors", "affirmations"):
for v in persona.get(k, []) or []:
memories.append(f"{k[:-1] if k.endswith('s') else k}: {v}")
counts["memory"] = self.load_memories(memories, expert="semantic")
return counts
# --------------------------------------------------------------- memory in
def load_memories(self, memories: Iterable[str], expert: str = "semantic",
importance: float = 0.5) -> int:
"""Map persona memories -> semantic/episodic experts (routed at recall)."""
ttl = self.basic_ttl_seconds if self.mode is Mode.BASIC else None
n = 0
for m in memories:
self.ctx.remember(m, expert=expert, scope=self._scope("memory"),
importance=importance, ttl=ttl)
n += 1
if self.mode is Mode.BASIC:
self._enforce_cap(self._scope("memory"))
return n
def remember_turn(self, text: str, expert: str = "episodic",
importance: float = 0.5) -> None:
"""Record one conversational turn (Basic applies TTL + cap)."""
ttl = self.basic_ttl_seconds if self.mode is Mode.BASIC else None
self.ctx.remember(text, expert=expert, scope=self._scope("memory"),
importance=importance, ttl=ttl)
if self.mode is Mode.BASIC:
self._enforce_cap(self._scope("memory"))
def _enforce_cap(self, scope: str) -> None:
"""Basic-engine cap: keep only the newest ``basic_cap`` items in scope."""
items = [it for it in self.ctx.store.all_items()
if it.scope == scope]
if len(items) <= self.basic_cap:
return
items.sort(key=lambda it: it.created_at, reverse=True)
for it in items[self.basic_cap:]:
self.ctx.store.delete(it.id)
# ------------------------------------------------------------------ recall
def build_pack(self, query: str, max_tokens: int = 220,
top_experts: int = 2) -> ContextPack:
"""Compact, profile-pinned context pack for one turn.
Profile is always injectable; learned memory is recalled by routing. The
default budget keeps per-turn injection in the low hundreds of tokens.
"""
return self.ctx.build_pack(query, scope=self.scope, top_experts=top_experts,
max_tokens=max_tokens, pin_experts=PINNED_EXPERTS)
def explain(self, query: str, max_tokens: int = 220,
top_experts: int = 2) -> str:
"""Inspect why each item was selected (externalized & inspectable)."""
return self.ctx.inspect(query, scope=self.scope, top_experts=top_experts,
max_tokens=max_tokens, pin_experts=PINNED_EXPERTS)
# ------------------------------------------------------- live state (v1 seam)
def load_live_state(self, state: Dict[str, str]) -> int:
"""Device / persona live state -> live_state expert (always injectable).
The live-state expert is a v1 component; until then these are stored and
pinned like profile so device facts ride along with each pack.
"""
n = 0
for key, value in state.items():
if value is None:
continue
self.ctx.remember(f"{key}: {value}", expert=LIVE_STATE_EXPERT,
scope=self._scope("state"), importance=0.8)
n += 1
return n
|