Compare commits
7 Commits
194e3e64b9
...
bfb81428ab
| Author | SHA1 | Date | |
|---|---|---|---|
| bfb81428ab | |||
| d7e2fce694 | |||
| 34392e4097 | |||
| aae95bfa6c | |||
| 30185f3fd8 | |||
| ecf0b852f9 | |||
| 071522ea33 |
@@ -2,6 +2,10 @@
|
|||||||
LOCAL_BASE_URL=http://localhost:11434
|
LOCAL_BASE_URL=http://localhost:11434
|
||||||
LOCAL_MODEL=qwen2.5:7b-instruct
|
LOCAL_MODEL=qwen2.5:7b-instruct
|
||||||
|
|
||||||
|
# MI50 backend — OpenAI-compatible llama.cpp server on the home-lab GPU box (CT202).
|
||||||
|
MI50_BASE_URL=http://10.0.0.42:8080/v1
|
||||||
|
MI50_MODEL=local-gpu
|
||||||
|
|
||||||
# Cloud backend (OpenAI) — higher quality, costs money.
|
# Cloud backend (OpenAI) — higher quality, costs money.
|
||||||
OPENAI_API_KEY=
|
OPENAI_API_KEY=
|
||||||
CLOUD_MODEL=gpt-4o-mini
|
CLOUD_MODEL=gpt-4o-mini
|
||||||
|
|||||||
+18
-1
@@ -39,6 +39,21 @@ def build_messages(session_id: str, user_msg: str) -> list[Message]:
|
|||||||
"""Assemble the full, tiered message list for one turn."""
|
"""Assemble the full, tiered message list for one turn."""
|
||||||
messages: list[Message] = [{"role": "system", "content": persona.system_prompt()}]
|
messages: list[Message] = [{"role": "system", "content": persona.system_prompt()}]
|
||||||
|
|
||||||
|
# Semantic memory: the distilled profile (who Brian is) — answers identity
|
||||||
|
# questions that raw recall can't. Always in context when it exists.
|
||||||
|
profile = memory.get_profile()
|
||||||
|
if profile:
|
||||||
|
messages.append(
|
||||||
|
{"role": "system", "content": "What you know about Brian:\n" + profile}
|
||||||
|
)
|
||||||
|
|
||||||
|
# Time-aware memory: the current narrative (recent arc, trends, callbacks).
|
||||||
|
narrative = memory.get_narrative()
|
||||||
|
if narrative:
|
||||||
|
messages.append(
|
||||||
|
{"role": "system", "content": "What's going on with Brian lately:\n" + narrative}
|
||||||
|
)
|
||||||
|
|
||||||
recent = memory.recent(session_id, n=RECENT_N)
|
recent = memory.recent(session_id, n=RECENT_N)
|
||||||
recent_ids = {ex.id for ex in recent}
|
recent_ids = {ex.id for ex in recent}
|
||||||
|
|
||||||
@@ -73,7 +88,9 @@ def build_messages(session_id: str, user_msg: str) -> list[Message]:
|
|||||||
def respond(session_id: str, user_msg: str, backend: Backend = "cloud") -> str:
|
def respond(session_id: str, user_msg: str, backend: Backend = "cloud") -> str:
|
||||||
"""Produce Lyra's reply to a single user message and persist the exchange."""
|
"""Produce Lyra's reply to a single user message and persist the exchange."""
|
||||||
cfg = config.load()
|
cfg = config.load()
|
||||||
model = cfg.local_model if backend == "local" else cfg.cloud_model
|
model = {"local": cfg.local_model, "cloud": cfg.cloud_model, "mi50": cfg.mi50_model}.get(
|
||||||
|
backend, backend
|
||||||
|
)
|
||||||
logbus.log(
|
logbus.log(
|
||||||
"info", "chat request", session=session_id, backend=backend,
|
"info", "chat request", session=session_id, backend=backend,
|
||||||
model=model, embed=cfg.embed_backend,
|
model=model, embed=cfg.embed_backend,
|
||||||
|
|||||||
@@ -14,6 +14,8 @@ load_dotenv()
|
|||||||
class Config:
|
class Config:
|
||||||
local_base_url: str
|
local_base_url: str
|
||||||
local_model: str
|
local_model: str
|
||||||
|
mi50_base_url: str # OpenAI-compatible llama.cpp server on the MI50 box
|
||||||
|
mi50_model: str
|
||||||
openai_api_key: str
|
openai_api_key: str
|
||||||
cloud_model: str
|
cloud_model: str
|
||||||
embed_backend: str # "cloud" (OpenAI) or "local" (Ollama)
|
embed_backend: str # "cloud" (OpenAI) or "local" (Ollama)
|
||||||
@@ -27,6 +29,8 @@ def load() -> Config:
|
|||||||
return Config(
|
return Config(
|
||||||
local_base_url=os.getenv("LOCAL_BASE_URL", "http://localhost:11434"),
|
local_base_url=os.getenv("LOCAL_BASE_URL", "http://localhost:11434"),
|
||||||
local_model=os.getenv("LOCAL_MODEL", "qwen2.5:7b-instruct"),
|
local_model=os.getenv("LOCAL_MODEL", "qwen2.5:7b-instruct"),
|
||||||
|
mi50_base_url=os.getenv("MI50_BASE_URL", "http://10.0.0.42:8080/v1"),
|
||||||
|
mi50_model=os.getenv("MI50_MODEL", "local-gpu"),
|
||||||
openai_api_key=os.getenv("OPENAI_API_KEY", ""),
|
openai_api_key=os.getenv("OPENAI_API_KEY", ""),
|
||||||
cloud_model=os.getenv("CLOUD_MODEL", "gpt-4o-mini"),
|
cloud_model=os.getenv("CLOUD_MODEL", "gpt-4o-mini"),
|
||||||
embed_backend=os.getenv("EMBED_BACKEND", "cloud").lower(),
|
embed_backend=os.getenv("EMBED_BACKEND", "cloud").lower(),
|
||||||
|
|||||||
+83
@@ -0,0 +1,83 @@
|
|||||||
|
"""Era rollups: per-month "what was happening" digests (consolidation step 3).
|
||||||
|
|
||||||
|
Groups session gists by the calendar month the session occurred (from real
|
||||||
|
exchange timestamps) and map-reduces each month into one digest. These are the
|
||||||
|
temporal memory tier — they answer "what was going on last December" and feed
|
||||||
|
the narrative engine. Runs on the consolidation backend (MI50 in steady state).
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
|
||||||
|
from lyra import config, llm, logbus, memory
|
||||||
|
from lyra.llm import Backend, Message
|
||||||
|
|
||||||
|
BATCH_CHARS = 18000
|
||||||
|
|
||||||
|
_PROMPT = """You are writing a monthly memory digest about Brian from the session \
|
||||||
|
summaries below (all from the same month). Capture: what he was focused on (poker \
|
||||||
|
and otherwise), notable events/results/decisions, recurring themes, and his mood \
|
||||||
|
and arc across the month. Third person, referring to him as "Brian". 5-10 \
|
||||||
|
sentences. This is a memory record, not a reply. No preamble."""
|
||||||
|
|
||||||
|
_MERGE_PROMPT = """Merge these partial monthly digests (same month) into one \
|
||||||
|
coherent digest about Brian for that month. Keep it tight, 5-10 sentences, no \
|
||||||
|
repetition. Third person."""
|
||||||
|
|
||||||
|
|
||||||
|
def _batch_texts(texts: list[str], budget: int) -> list[str]:
|
||||||
|
blocks, buf, size = [], [], 0
|
||||||
|
for t in texts:
|
||||||
|
if size + len(t) > budget and buf:
|
||||||
|
blocks.append("\n\n".join(buf))
|
||||||
|
buf, size = [], 0
|
||||||
|
buf.append(t)
|
||||||
|
size += len(t)
|
||||||
|
if buf:
|
||||||
|
blocks.append("\n\n".join(buf))
|
||||||
|
return blocks
|
||||||
|
|
||||||
|
|
||||||
|
def _call(prompt: str, body: str, backend: Backend) -> str:
|
||||||
|
messages: list[Message] = [
|
||||||
|
{"role": "system", "content": prompt},
|
||||||
|
{"role": "user", "content": body},
|
||||||
|
]
|
||||||
|
return llm.complete(messages, backend=backend)
|
||||||
|
|
||||||
|
|
||||||
|
def _digest_month(gists: list[str], backend: Backend) -> str:
|
||||||
|
"""Map-reduce a month's session gists into one digest."""
|
||||||
|
blocks = _batch_texts(gists, BATCH_CHARS)
|
||||||
|
partials = [_call(_PROMPT, b, backend) for b in blocks]
|
||||||
|
while len(partials) > 1:
|
||||||
|
partials = [_call(_MERGE_PROMPT, g, backend) for g in _batch_texts(partials, BATCH_CHARS)]
|
||||||
|
return partials[0]
|
||||||
|
|
||||||
|
|
||||||
|
def rebuild_eras(backend: Backend | None = None) -> dict:
|
||||||
|
"""(Re)build a digest for every month that has session gists."""
|
||||||
|
backend = backend or config.load().summary_backend
|
||||||
|
by_month = memory.summaries_by_month()
|
||||||
|
months = 0
|
||||||
|
for month in sorted(by_month):
|
||||||
|
digest = _digest_month(by_month[month], backend)
|
||||||
|
memory.store_era(month, digest, len(by_month[month]))
|
||||||
|
months += 1
|
||||||
|
logbus.log("info", "era built", month=month, sessions=len(by_month[month]))
|
||||||
|
report = {"months": months}
|
||||||
|
logbus.log("info", "eras complete", **report)
|
||||||
|
return report
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> int:
|
||||||
|
report = rebuild_eras()
|
||||||
|
if not report["months"]:
|
||||||
|
print("No summaries yet — run lyra-summarize first.")
|
||||||
|
return 1
|
||||||
|
for era in memory.list_eras():
|
||||||
|
print(f"\n## {era.month} ({era.session_count} sessions)\n{era.content}")
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
raise SystemExit(main())
|
||||||
+7
-1
@@ -14,7 +14,7 @@ class Message(TypedDict):
|
|||||||
content: str
|
content: str
|
||||||
|
|
||||||
|
|
||||||
Backend = Literal["local", "cloud"]
|
Backend = Literal["local", "cloud", "mi50"]
|
||||||
|
|
||||||
|
|
||||||
def complete(messages: list[Message], backend: Backend = "local") -> str:
|
def complete(messages: list[Message], backend: Backend = "local") -> str:
|
||||||
@@ -26,6 +26,12 @@ def complete(messages: list[Message], backend: Backend = "local") -> str:
|
|||||||
resp = client.chat.completions.create(model=cfg.cloud_model, messages=messages)
|
resp = client.chat.completions.create(model=cfg.cloud_model, messages=messages)
|
||||||
return resp.choices[0].message.content or ""
|
return resp.choices[0].message.content or ""
|
||||||
|
|
||||||
|
if backend == "mi50":
|
||||||
|
# MI50 box runs an OpenAI-compatible llama.cpp server; key is unused.
|
||||||
|
client = OpenAI(api_key="not-needed", base_url=cfg.mi50_base_url)
|
||||||
|
resp = client.chat.completions.create(model=cfg.mi50_model, messages=messages)
|
||||||
|
return resp.choices[0].message.content or ""
|
||||||
|
|
||||||
resp = httpx.post(
|
resp = httpx.post(
|
||||||
f"{cfg.local_base_url}/api/chat",
|
f"{cfg.local_base_url}/api/chat",
|
||||||
json={"model": cfg.local_model, "messages": messages, "stream": False},
|
json={"model": cfg.local_model, "messages": messages, "stream": False},
|
||||||
|
|||||||
+166
@@ -43,6 +43,33 @@ CREATE TABLE IF NOT EXISTS summaries (
|
|||||||
last_exchange_id INTEGER NOT NULL,
|
last_exchange_id INTEGER NOT NULL,
|
||||||
created_at TEXT NOT NULL
|
created_at TEXT NOT NULL
|
||||||
);
|
);
|
||||||
|
|
||||||
|
-- Derived semantic memory: standing facts about the user, distilled from the
|
||||||
|
-- session gists by the consolidation pass. Single row (id='self').
|
||||||
|
CREATE TABLE IF NOT EXISTS profile (
|
||||||
|
id TEXT PRIMARY KEY,
|
||||||
|
content TEXT NOT NULL,
|
||||||
|
sessions_covered INTEGER NOT NULL,
|
||||||
|
updated_at TEXT NOT NULL
|
||||||
|
);
|
||||||
|
|
||||||
|
-- Temporal memory: one "what was happening" digest per calendar month, rolled
|
||||||
|
-- up from that month's session gists. month is "YYYY-MM".
|
||||||
|
CREATE TABLE IF NOT EXISTS eras (
|
||||||
|
month TEXT PRIMARY KEY,
|
||||||
|
content TEXT NOT NULL,
|
||||||
|
embedding BLOB NOT NULL,
|
||||||
|
session_count INTEGER NOT NULL,
|
||||||
|
created_at TEXT NOT NULL
|
||||||
|
);
|
||||||
|
|
||||||
|
-- The current narrative: time-aware arc/trends/callbacks (vs the timeless
|
||||||
|
-- profile). Distilled from profile + recent eras. Single row (id='current').
|
||||||
|
CREATE TABLE IF NOT EXISTS narrative (
|
||||||
|
id TEXT PRIMARY KEY,
|
||||||
|
content TEXT NOT NULL,
|
||||||
|
updated_at TEXT NOT NULL
|
||||||
|
);
|
||||||
"""
|
"""
|
||||||
|
|
||||||
_conn: sqlite3.Connection | None = None
|
_conn: sqlite3.Connection | None = None
|
||||||
@@ -86,6 +113,15 @@ class Summary:
|
|||||||
score: float | None = None
|
score: float | None = None
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Era:
|
||||||
|
month: str # "YYYY-MM"
|
||||||
|
content: str
|
||||||
|
session_count: int
|
||||||
|
created_at: str
|
||||||
|
score: float | None = None
|
||||||
|
|
||||||
|
|
||||||
def _to_blob(vec: list[float]) -> bytes:
|
def _to_blob(vec: list[float]) -> bytes:
|
||||||
return np.asarray(vec, dtype=np.float32).tobytes()
|
return np.asarray(vec, dtype=np.float32).tobytes()
|
||||||
|
|
||||||
@@ -290,6 +326,136 @@ def unsummarized_count(session_id: str) -> int:
|
|||||||
return int(r["n"])
|
return int(r["n"])
|
||||||
|
|
||||||
|
|
||||||
|
def list_summaries() -> list[Summary]:
|
||||||
|
"""Every session gist (for the profile/era consolidation passes)."""
|
||||||
|
conn = _connection()
|
||||||
|
rows = conn.execute(
|
||||||
|
"SELECT session_id, content, last_exchange_id, created_at FROM summaries "
|
||||||
|
"ORDER BY created_at ASC"
|
||||||
|
).fetchall()
|
||||||
|
return [
|
||||||
|
Summary(
|
||||||
|
session_id=r["session_id"],
|
||||||
|
content=r["content"],
|
||||||
|
last_exchange_id=r["last_exchange_id"],
|
||||||
|
created_at=r["created_at"],
|
||||||
|
)
|
||||||
|
for r in rows
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def set_profile(content: str, sessions_covered: int, profile_id: str = "self") -> None:
|
||||||
|
"""Store/replace the derived semantic profile."""
|
||||||
|
now = datetime.now(timezone.utc).isoformat()
|
||||||
|
conn = _connection()
|
||||||
|
with conn:
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO profile (id, content, sessions_covered, updated_at) "
|
||||||
|
"VALUES (?, ?, ?, ?) "
|
||||||
|
"ON CONFLICT(id) DO UPDATE SET content=excluded.content, "
|
||||||
|
"sessions_covered=excluded.sessions_covered, updated_at=excluded.updated_at",
|
||||||
|
(profile_id, content, sessions_covered, now),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def get_profile(profile_id: str = "self") -> str | None:
|
||||||
|
conn = _connection()
|
||||||
|
r = conn.execute("SELECT content FROM profile WHERE id = ?", (profile_id,)).fetchone()
|
||||||
|
return r["content"] if r else None
|
||||||
|
|
||||||
|
|
||||||
|
# --- Era tier (per-month temporal rollups) ---
|
||||||
|
|
||||||
|
|
||||||
|
def summaries_by_month() -> dict[str, list[str]]:
|
||||||
|
"""Map "YYYY-MM" -> list of session gists for sessions that occurred that month.
|
||||||
|
|
||||||
|
A session's month comes from its earliest exchange timestamp (real ChatGPT
|
||||||
|
dates for imported sessions), not when it was summarized.
|
||||||
|
"""
|
||||||
|
conn = _connection()
|
||||||
|
rows = conn.execute(
|
||||||
|
"""
|
||||||
|
SELECT substr(MIN(e.created_at), 1, 7) AS month, s.content AS content
|
||||||
|
FROM summaries s JOIN exchanges e ON e.session_id = s.session_id
|
||||||
|
GROUP BY s.session_id
|
||||||
|
"""
|
||||||
|
).fetchall()
|
||||||
|
out: dict[str, list[str]] = {}
|
||||||
|
for r in rows:
|
||||||
|
out.setdefault(r["month"], []).append(r["content"])
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def store_era(month: str, content: str, session_count: int) -> None:
|
||||||
|
"""Embed and persist a month's digest, replacing any prior one."""
|
||||||
|
[embedding] = llm.embed([content])
|
||||||
|
now = datetime.now(timezone.utc).isoformat()
|
||||||
|
conn = _connection()
|
||||||
|
with conn:
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO eras (month, content, embedding, session_count, created_at) "
|
||||||
|
"VALUES (?, ?, ?, ?, ?) "
|
||||||
|
"ON CONFLICT(month) DO UPDATE SET content=excluded.content, "
|
||||||
|
"embedding=excluded.embedding, session_count=excluded.session_count, "
|
||||||
|
"created_at=excluded.created_at",
|
||||||
|
(month, content, _to_blob(embedding), session_count, now),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def list_eras() -> list[Era]:
|
||||||
|
"""All month digests, chronological."""
|
||||||
|
conn = _connection()
|
||||||
|
rows = conn.execute(
|
||||||
|
"SELECT month, content, session_count, created_at FROM eras ORDER BY month ASC"
|
||||||
|
).fetchall()
|
||||||
|
return [
|
||||||
|
Era(month=r["month"], content=r["content"],
|
||||||
|
session_count=r["session_count"], created_at=r["created_at"])
|
||||||
|
for r in rows
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def set_narrative(content: str, narrative_id: str = "current") -> None:
|
||||||
|
"""Store/replace the current narrative."""
|
||||||
|
now = datetime.now(timezone.utc).isoformat()
|
||||||
|
conn = _connection()
|
||||||
|
with conn:
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO narrative (id, content, updated_at) VALUES (?, ?, ?) "
|
||||||
|
"ON CONFLICT(id) DO UPDATE SET content=excluded.content, updated_at=excluded.updated_at",
|
||||||
|
(narrative_id, content, now),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def get_narrative(narrative_id: str = "current") -> str | None:
|
||||||
|
conn = _connection()
|
||||||
|
r = conn.execute("SELECT content FROM narrative WHERE id = ?", (narrative_id,)).fetchone()
|
||||||
|
return r["content"] if r else None
|
||||||
|
|
||||||
|
|
||||||
|
def recall_eras(query: str, k: int = 2) -> list[Era]:
|
||||||
|
"""Top-k month digests most similar to `query` (time-based context)."""
|
||||||
|
[q_vec] = llm.embed([query])
|
||||||
|
q = np.asarray(q_vec, dtype=np.float32)
|
||||||
|
conn = _connection()
|
||||||
|
rows = conn.execute(
|
||||||
|
"SELECT month, content, embedding, session_count, created_at FROM eras"
|
||||||
|
).fetchall()
|
||||||
|
if not rows:
|
||||||
|
return []
|
||||||
|
matrix = np.stack([_from_blob(r["embedding"]) for r in rows])
|
||||||
|
norms = np.linalg.norm(matrix, axis=1)
|
||||||
|
scores = (matrix @ q) / (norms * np.linalg.norm(q) + 1e-9)
|
||||||
|
top_idx = np.argsort(scores)[::-1][:k]
|
||||||
|
return [
|
||||||
|
Era(month=rows[i]["month"], content=rows[i]["content"],
|
||||||
|
session_count=rows[i]["session_count"], created_at=rows[i]["created_at"],
|
||||||
|
score=float(scores[i]))
|
||||||
|
for i in top_idx
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
def recall_summaries(query: str, k: int = 3, exclude_session: str | None = None) -> list[Summary]:
|
def recall_summaries(query: str, k: int = 3, exclude_session: str | None = None) -> list[Summary]:
|
||||||
"""Top-k session summaries most similar to `query` (the long-term gist tier)."""
|
"""Top-k session summaries most similar to `query` (the long-term gist tier)."""
|
||||||
[q_vec] = llm.embed([query])
|
[q_vec] = llm.embed([query])
|
||||||
|
|||||||
@@ -0,0 +1,66 @@
|
|||||||
|
"""Narrative engine (consolidation step 4): the current arc, trends, callbacks.
|
||||||
|
|
||||||
|
Where the profile is timeless ("who Brian is"), the narrative is time-aware
|
||||||
|
("what's going on lately, where things are trending"). It distills the profile
|
||||||
|
plus the most recent monthly era digests into the current story — recent focus,
|
||||||
|
notable trends or changes, mood/arc, and a few specific callbacks worth
|
||||||
|
referencing. Injected into chat so Lyra follows along like a friend who's been
|
||||||
|
paying attention. Runs on the consolidation backend (MI50 in steady state).
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
|
||||||
|
from lyra import config, llm, logbus, memory
|
||||||
|
from lyra.llm import Backend, Message
|
||||||
|
|
||||||
|
RECENT_ERAS = 4
|
||||||
|
|
||||||
|
_PROMPT = """You are distilling the CURRENT narrative about Brian — what a close \
|
||||||
|
friend who has been following along would keep in mind right now. From his profile \
|
||||||
|
and recent monthly digests below, write: what he's been focused on lately, any \
|
||||||
|
notable trends or changes (improving, slipping, new patterns), his current arc and \
|
||||||
|
mood, and 2-4 specific things worth referencing back to him ("remember when…"). \
|
||||||
|
Third person, referring to him as "Brian". 6-10 sentences. This is a memory note, \
|
||||||
|
not a reply. No preamble."""
|
||||||
|
|
||||||
|
|
||||||
|
def rebuild_narrative(backend: Backend | None = None) -> str | None:
|
||||||
|
"""(Re)derive the current narrative from the profile + recent era digests."""
|
||||||
|
backend = backend or config.load().summary_backend
|
||||||
|
profile = memory.get_profile()
|
||||||
|
eras = memory.list_eras()
|
||||||
|
if not profile and not eras:
|
||||||
|
return None
|
||||||
|
|
||||||
|
parts = []
|
||||||
|
if profile:
|
||||||
|
parts.append("PROFILE (timeless):\n" + profile)
|
||||||
|
recent = eras[-RECENT_ERAS:]
|
||||||
|
if recent:
|
||||||
|
parts.append(
|
||||||
|
"RECENT MONTHS (oldest first):\n"
|
||||||
|
+ "\n\n".join(f"[{e.month}]\n{e.content}" for e in recent)
|
||||||
|
)
|
||||||
|
body = "\n\n".join(parts)
|
||||||
|
|
||||||
|
messages: list[Message] = [
|
||||||
|
{"role": "system", "content": _PROMPT},
|
||||||
|
{"role": "user", "content": body},
|
||||||
|
]
|
||||||
|
narrative = llm.complete(messages, backend=backend)
|
||||||
|
memory.set_narrative(narrative)
|
||||||
|
logbus.log("info", "narrative rebuilt", chars=len(narrative), eras=len(recent))
|
||||||
|
return narrative
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> int:
|
||||||
|
narrative = rebuild_narrative()
|
||||||
|
if narrative is None:
|
||||||
|
print("Need a profile and/or eras first — run lyra-profile and lyra-era.")
|
||||||
|
return 1
|
||||||
|
print(narrative)
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
raise SystemExit(main())
|
||||||
@@ -0,0 +1,84 @@
|
|||||||
|
"""Profile derivation: distill standing facts about the user (semantic memory).
|
||||||
|
|
||||||
|
This is consolidation step 2. It reads every session gist and map-reduces them
|
||||||
|
into one profile document — who Brian is as a player and person — which is then
|
||||||
|
injected into every prompt. This is what answers identity/abstract questions
|
||||||
|
("what kind of player am I", "what are my leaks") that raw recall handles badly,
|
||||||
|
because those are patterns across many sessions, not facts in any single message.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
|
||||||
|
from lyra import config, llm, logbus, memory
|
||||||
|
from lyra.llm import Backend, Message
|
||||||
|
|
||||||
|
BATCH_CHARS = 18000
|
||||||
|
|
||||||
|
_MAP_PROMPT = """From these session summaries, extract durable facts about Brian \
|
||||||
|
— things that are stably true, not one-off events. Cover, where present: poker \
|
||||||
|
games/formats/stakes he plays, his playing style and strengths, recurring leaks \
|
||||||
|
and tendencies, mental-game patterns (tilt triggers, scared money, fatigue), \
|
||||||
|
relevant personal context, and how he likes to be coached. Terse bullet points. \
|
||||||
|
Omit anything not supported by the summaries."""
|
||||||
|
|
||||||
|
_REDUCE_PROMPT = """Merge these fact lists into one deduplicated profile of Brian. \
|
||||||
|
Organize under these headings: Poker Style, Leaks & Tendencies, Mental Game, \
|
||||||
|
Personal Context, Working With Brian. Keep it tight — bullets, no fluff, no \
|
||||||
|
repetition. Resolve contradictions toward the more recent/frequent signal."""
|
||||||
|
|
||||||
|
|
||||||
|
def _batch_texts(texts: list[str], budget: int) -> list[str]:
|
||||||
|
"""Group texts into joined blocks under `budget` chars."""
|
||||||
|
blocks, buf, size = [], [], 0
|
||||||
|
for t in texts:
|
||||||
|
if size + len(t) > budget and buf:
|
||||||
|
blocks.append("\n\n".join(buf))
|
||||||
|
buf, size = [], 0
|
||||||
|
buf.append(t)
|
||||||
|
size += len(t)
|
||||||
|
if buf:
|
||||||
|
blocks.append("\n\n".join(buf))
|
||||||
|
return blocks
|
||||||
|
|
||||||
|
|
||||||
|
def _call(prompt: str, body: str, backend: Backend) -> str:
|
||||||
|
messages: list[Message] = [
|
||||||
|
{"role": "system", "content": prompt},
|
||||||
|
{"role": "user", "content": body},
|
||||||
|
]
|
||||||
|
return llm.complete(messages, backend=backend)
|
||||||
|
|
||||||
|
|
||||||
|
def rebuild_profile(backend: Backend | None = None) -> str | None:
|
||||||
|
"""Re-derive the profile from all current session gists and store it."""
|
||||||
|
backend = backend or config.load().summary_backend
|
||||||
|
summaries = memory.list_summaries()
|
||||||
|
if not summaries:
|
||||||
|
return None
|
||||||
|
|
||||||
|
# MAP: extract facts from batches of gists.
|
||||||
|
blocks = _batch_texts([s.content for s in summaries], BATCH_CHARS)
|
||||||
|
partials = [_call(_MAP_PROMPT, b, backend) for b in blocks]
|
||||||
|
logbus.log("info", "profile map done", batches=len(partials), sessions=len(summaries))
|
||||||
|
|
||||||
|
# REDUCE: fold partials together until one remains.
|
||||||
|
while len(partials) > 1:
|
||||||
|
partials = [_call(_REDUCE_PROMPT, g, backend) for g in _batch_texts(partials, BATCH_CHARS)]
|
||||||
|
profile = partials[0]
|
||||||
|
|
||||||
|
memory.set_profile(profile, len(summaries))
|
||||||
|
logbus.log("info", "profile rebuilt", sessions=len(summaries), chars=len(profile))
|
||||||
|
return profile
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> int:
|
||||||
|
profile = rebuild_profile()
|
||||||
|
if profile is None:
|
||||||
|
print("No summaries yet — run lyra-summarize first.")
|
||||||
|
return 1
|
||||||
|
print(profile)
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
raise SystemExit(main())
|
||||||
+121
-25
@@ -1,17 +1,27 @@
|
|||||||
"""Session summarization: compact a session's raw exchanges into a stored gist.
|
"""Session summarization: compact a session's raw exchanges into a stored gist.
|
||||||
|
|
||||||
This is the compaction half of the tiered memory. Raw exchanges stay for detail
|
This is the first consolidation stage. Raw exchanges stay for detail recall; the
|
||||||
recall; the summary is what surfaces when an *older* session is recalled later —
|
summary is what surfaces when an *older* session is recalled, and it's the input
|
||||||
"a month ago is a general idea," per the design.
|
to the profile (semantic memory) and era-rollup tiers.
|
||||||
|
|
||||||
|
Long sessions are summarized in chunks, then the partial gists are merged, so a
|
||||||
|
big imported conversation doesn't blow the local model's context window.
|
||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from lyra import config, llm, logbus, memory
|
import sys
|
||||||
from lyra.llm import Backend
|
import time
|
||||||
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||||
|
|
||||||
# Re-summarize a session once it has accumulated this many new raw exchanges
|
from lyra import config, llm, logbus, memory
|
||||||
# beyond what its current summary covers.
|
from lyra.llm import Backend, Message
|
||||||
|
|
||||||
|
_RETRIES = 4
|
||||||
|
|
||||||
|
# Re-summarize a session once it has accumulated this many new raw exchanges.
|
||||||
SUMMARIZE_AFTER = 20
|
SUMMARIZE_AFTER = 20
|
||||||
|
# Transcript budget per LLM call; longer sessions are chunked + merged.
|
||||||
|
MAX_TRANSCRIPT_CHARS = 24000
|
||||||
|
|
||||||
_PROMPT = """You are compacting a conversation into a long-term memory record \
|
_PROMPT = """You are compacting a conversation into a long-term memory record \
|
||||||
(not replying to anyone). Write a concise gist of the session below: what was \
|
(not replying to anyone). Write a concise gist of the session below: what was \
|
||||||
@@ -24,29 +34,54 @@ def _transcript(exchanges: list[memory.Exchange]) -> str:
|
|||||||
return "\n".join(f"{ex.role}: {ex.content}" for ex in exchanges)
|
return "\n".join(f"{ex.role}: {ex.content}" for ex in exchanges)
|
||||||
|
|
||||||
|
|
||||||
def summarize_session(session_id: str, backend: Backend | None = None) -> str | None:
|
def _chunk(text: str, budget: int) -> list[str]:
|
||||||
"""(Re)generate and store the gist for a session. Returns the summary text.
|
"""Split on line boundaries into pieces under `budget` chars."""
|
||||||
|
chunks, buf, size = [], [], 0
|
||||||
|
for line in text.splitlines(keepends=True):
|
||||||
|
if size + len(line) > budget and buf:
|
||||||
|
chunks.append("".join(buf))
|
||||||
|
buf, size = [], 0
|
||||||
|
buf.append(line)
|
||||||
|
size += len(line)
|
||||||
|
if buf:
|
||||||
|
chunks.append("".join(buf))
|
||||||
|
return chunks
|
||||||
|
|
||||||
Returns None if the session has no exchanges. The summarizer defaults to the
|
|
||||||
local backend so routine compaction stays free.
|
def _summarize_text(text: str, backend: Backend) -> str:
|
||||||
"""
|
messages: list[Message] = [
|
||||||
|
{"role": "system", "content": _PROMPT},
|
||||||
|
{"role": "user", "content": text},
|
||||||
|
]
|
||||||
|
# Retry transient backend errors (e.g. the GPU server restarting) with backoff.
|
||||||
|
for attempt in range(_RETRIES):
|
||||||
|
try:
|
||||||
|
return llm.complete(messages, backend=backend)
|
||||||
|
except Exception as exc:
|
||||||
|
if attempt == _RETRIES - 1:
|
||||||
|
raise
|
||||||
|
logbus.log("debug", "summary retry", attempt=attempt + 1, error=str(exc)[:80])
|
||||||
|
time.sleep(5 * (attempt + 1))
|
||||||
|
raise RuntimeError("unreachable")
|
||||||
|
|
||||||
|
|
||||||
|
def _summarize_transcript(transcript: str, backend: Backend) -> str:
|
||||||
|
"""Transcript -> gist (LLM only, no DB). Chunks + merges if oversized."""
|
||||||
|
if len(transcript) <= MAX_TRANSCRIPT_CHARS:
|
||||||
|
return _summarize_text(transcript, backend)
|
||||||
|
partials = [_summarize_text(c, backend) for c in _chunk(transcript, MAX_TRANSCRIPT_CHARS)]
|
||||||
|
return _summarize_text("Partial summaries to merge:\n\n" + "\n\n".join(partials), backend)
|
||||||
|
|
||||||
|
|
||||||
|
def summarize_session(session_id: str, backend: Backend | None = None) -> str | None:
|
||||||
|
"""(Re)generate and store the gist for a session. Returns the summary text."""
|
||||||
exchanges = memory.history(session_id)
|
exchanges = memory.history(session_id)
|
||||||
if not exchanges:
|
if not exchanges:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
backend = backend or config.load().summary_backend
|
backend = backend or config.load().summary_backend
|
||||||
messages = [
|
gist = _summarize_transcript(_transcript(exchanges), backend)
|
||||||
{"role": "system", "content": _PROMPT},
|
memory.store_summary(session_id, gist, exchanges[-1].id)
|
||||||
{"role": "user", "content": _transcript(exchanges)},
|
logbus.log("info", "summarized session", session=session_id, exchanges=len(exchanges))
|
||||||
]
|
|
||||||
gist = llm.complete(messages, backend=backend)
|
|
||||||
|
|
||||||
last_id = exchanges[-1].id
|
|
||||||
memory.store_summary(session_id, gist, last_id)
|
|
||||||
logbus.log(
|
|
||||||
"info", "summarized session", session=session_id,
|
|
||||||
exchanges=len(exchanges), backend=backend,
|
|
||||||
)
|
|
||||||
return gist
|
return gist
|
||||||
|
|
||||||
|
|
||||||
@@ -54,3 +89,64 @@ def maybe_summarize(session_id: str, backend: Backend | None = None) -> None:
|
|||||||
"""Summarize the session if enough new turns have accumulated since last time."""
|
"""Summarize the session if enough new turns have accumulated since last time."""
|
||||||
if memory.unsummarized_count(session_id) >= SUMMARIZE_AFTER:
|
if memory.unsummarized_count(session_id) >= SUMMARIZE_AFTER:
|
||||||
summarize_session(session_id, backend=backend)
|
summarize_session(session_id, backend=backend)
|
||||||
|
|
||||||
|
|
||||||
|
def summarize_all(
|
||||||
|
backend: Backend | None = None, limit: int | None = None, workers: int = 8
|
||||||
|
) -> dict:
|
||||||
|
"""Summarize every session that needs it. Idempotent and resumable.
|
||||||
|
|
||||||
|
LLM summarization runs concurrently across `workers` threads (great for a
|
||||||
|
cloud backend). DB reads (loading transcripts) and writes (store_summary,
|
||||||
|
which also embeds) happen on the main thread, so the single SQLite
|
||||||
|
connection is never touched from multiple threads.
|
||||||
|
"""
|
||||||
|
backend = backend or config.load().summary_backend
|
||||||
|
|
||||||
|
# Main thread: collect the work (transcripts) for sessions needing a summary.
|
||||||
|
todo: list[tuple[str, str, int]] = []
|
||||||
|
for s in memory.list_sessions():
|
||||||
|
sid = s["id"]
|
||||||
|
if memory.get_summary(sid) and memory.unsummarized_count(sid) == 0:
|
||||||
|
continue
|
||||||
|
exchanges = memory.history(sid)
|
||||||
|
if not exchanges:
|
||||||
|
continue
|
||||||
|
todo.append((sid, _transcript(exchanges), exchanges[-1].id))
|
||||||
|
if limit is not None and len(todo) >= limit:
|
||||||
|
break
|
||||||
|
|
||||||
|
done, failed = 0, 0
|
||||||
|
logbus.log("info", "summarize-all starting", todo=len(todo), backend=backend, workers=workers)
|
||||||
|
|
||||||
|
def work(item: tuple[str, str, int]) -> tuple[str, str, int]:
|
||||||
|
sid, transcript, last_id = item
|
||||||
|
return sid, _summarize_transcript(transcript, backend), last_id
|
||||||
|
|
||||||
|
with ThreadPoolExecutor(max_workers=workers) as pool:
|
||||||
|
futures = {pool.submit(work, item): item for item in todo}
|
||||||
|
for fut in as_completed(futures):
|
||||||
|
sid = futures[fut][0]
|
||||||
|
try:
|
||||||
|
_, gist, last_id = fut.result()
|
||||||
|
memory.store_summary(sid, gist, last_id) # main thread: embed + write
|
||||||
|
done += 1
|
||||||
|
except Exception as exc:
|
||||||
|
failed += 1
|
||||||
|
logbus.log("error", "summarize failed", session=sid, error=str(exc)[:120])
|
||||||
|
if (done + failed) % 25 == 0:
|
||||||
|
logbus.log("info", "summarize-all progress", done=done, failed=failed, total=len(todo))
|
||||||
|
|
||||||
|
report = {"summarized": done, "failed": failed, "total": len(todo)}
|
||||||
|
logbus.log("info", "summarize-all complete", **report)
|
||||||
|
return report
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> int:
|
||||||
|
limit = int(sys.argv[1]) if len(sys.argv) > 1 else None
|
||||||
|
print(summarize_all(limit=limit))
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
raise SystemExit(main())
|
||||||
|
|||||||
+4
-1
@@ -32,7 +32,10 @@ _CLOUD = {"OPENAI", "cloud", "custom"}
|
|||||||
|
|
||||||
|
|
||||||
def _backend_for(label: str | None) -> Backend:
|
def _backend_for(label: str | None) -> Backend:
|
||||||
if label and label.upper() in {"PRIMARY", "SECONDARY", "FALLBACK", "LOCAL"}:
|
key = (label or "").lower()
|
||||||
|
if key == "mi50":
|
||||||
|
return "mi50"
|
||||||
|
if key in {"local", "primary", "secondary", "fallback"}:
|
||||||
return "local"
|
return "local"
|
||||||
return "cloud"
|
return "cloud"
|
||||||
|
|
||||||
|
|||||||
@@ -123,6 +123,11 @@
|
|||||||
<span>Local — Ollama</span>
|
<span>Local — Ollama</span>
|
||||||
<small>Free, private, runs on your home lab (LOCAL_MODEL)</small>
|
<small>Free, private, runs on your home lab (LOCAL_MODEL)</small>
|
||||||
</label>
|
</label>
|
||||||
|
<label class="radio-label">
|
||||||
|
<input type="radio" name="backend" value="mi50">
|
||||||
|
<span>MI50 — local GPU</span>
|
||||||
|
<small>Free, llama.cpp on the MI50 box (MI50_BASE_URL)</small>
|
||||||
|
</label>
|
||||||
<label class="radio-label">
|
<label class="radio-label">
|
||||||
<input type="radio" name="backend" value="cloud">
|
<input type="radio" name="backend" value="cloud">
|
||||||
<span>Cloud — OpenAI</span>
|
<span>Cloud — OpenAI</span>
|
||||||
|
|||||||
@@ -17,6 +17,10 @@ dependencies = [
|
|||||||
lyra = "lyra.__main__:main"
|
lyra = "lyra.__main__:main"
|
||||||
lyra-web = "lyra.web.server:serve"
|
lyra-web = "lyra.web.server:serve"
|
||||||
lyra-import = "lyra.ingest:main"
|
lyra-import = "lyra.ingest:main"
|
||||||
|
lyra-summarize = "lyra.summary:main"
|
||||||
|
lyra-profile = "lyra.profile:main"
|
||||||
|
lyra-era = "lyra.era:main"
|
||||||
|
lyra-narrative = "lyra.narrative:main"
|
||||||
|
|
||||||
[dependency-groups]
|
[dependency-groups]
|
||||||
dev = [
|
dev = [
|
||||||
|
|||||||
Reference in New Issue
Block a user