feat(P2): perceive (read the moment) + route nudges register on charged turns
The control plane gains senses — cheap, deterministic, no LLM:
- lyra/perceive.py: lexicon+signal heuristic → {sentiment, intensity, tilt, kind:
emotional|strategic|meta|build|casual}. Good at the action-relevant signal,
especially tilt (the mental-game core). Word-boundary matching so 'line' doesn't
fire inside 'pipeline'.
- mind: _perceive fills ctx.moment; _route keeps the manual mode as the dominant
frame but, on a genuinely charged moment, adds a per-turn register nudge — tilt →
"meet him there, warm and steady, don't clip into logging"; up/energized → "match
his energy." Neutral turns get nothing (don't over-narrate). Injected via
build_messages(moment=...). Logged to /logs for observability.
- tests: perceive read (tilt/strategy/up/build/casual) + route nudge on/off.
Suite 92 green, ruff clean.
Complements modes (manual frame) — perceive refines register within it, doesn't
override. Model routing (mind/mouth) is P3.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
+34
-7
@@ -16,7 +16,7 @@ from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from lyra import clock, config, llm, logbus, memory, modes, persona, self_state, thoughts
|
||||
from lyra import clock, config, llm, logbus, memory, modes, perceive, persona, self_state, thoughts
|
||||
from lyra.llm import Backend, Message
|
||||
|
||||
RECALL_K = 3 # raw cross-session "sharp detail" hits
|
||||
@@ -91,7 +91,7 @@ def _render(messages: list[Message]) -> str:
|
||||
|
||||
|
||||
def build_messages(session_id: str, user_msg: str,
|
||||
mode: modes.Mode | None = None) -> list[Message]:
|
||||
mode: modes.Mode | None = None, moment: dict | None = None) -> list[Message]:
|
||||
"""Assemble the full, tiered message list for one turn."""
|
||||
messages: list[Message] = [{"role": "system", "content": persona.system_prompt()}]
|
||||
|
||||
@@ -114,6 +114,11 @@ def build_messages(session_id: str, user_msg: str,
|
||||
if state_note:
|
||||
messages.append({"role": "system", "content": state_note})
|
||||
|
||||
# Read of the moment (from perceive/route) — a per-turn register nudge, e.g. "he
|
||||
# sounds tilted, meet him there." Only present when the moment is genuinely charged.
|
||||
if moment and moment.get("note"):
|
||||
messages.append({"role": "system", "content": moment["note"]})
|
||||
|
||||
# When she is: current time + the gap since Brian last spoke (she has no clock).
|
||||
messages.append(_now_note())
|
||||
|
||||
@@ -230,25 +235,47 @@ class TurnContext:
|
||||
backend: Backend
|
||||
model: str | None = None
|
||||
mode: modes.Mode | None = None
|
||||
moment: dict = field(default_factory=dict) # perceive fills this in (P2)
|
||||
moment: dict = field(default_factory=dict) # perceive fills this in
|
||||
register: str | None = None # route's per-turn register nudge
|
||||
messages: list[Message] = field(default_factory=list)
|
||||
|
||||
|
||||
def _perceive(ctx: TurnContext) -> TurnContext:
|
||||
"""Read the moment (sentiment / kind / tilt). Stub for now — P2 fills it in."""
|
||||
ctx.moment = {}
|
||||
"""Read the moment from what he just said — cheap heuristics (perceive.read)."""
|
||||
ctx.moment = perceive.read(ctx.user_msg)
|
||||
return ctx
|
||||
|
||||
|
||||
# How charged a moment must be before we nudge her register (avoid narrating every turn).
|
||||
_TILT_BAR = 0.5
|
||||
_UP_BAR = 0.6
|
||||
|
||||
|
||||
def _route(ctx: TurnContext) -> TurnContext:
|
||||
"""Pick how she shows up. Manual for now: the mode chosen for this session."""
|
||||
"""Decide how she shows up. The manual mode is the dominant frame; on top of it,
|
||||
a charged emotional moment adds a per-turn register nudge (deterministic). Most
|
||||
turns are neutral and get no note — that's the point (don't over-narrate)."""
|
||||
ctx.mode = modes.get(memory.get_session_mode(ctx.session_id))
|
||||
m = ctx.moment or {}
|
||||
note = None
|
||||
if m.get("tilt", 0) >= _TILT_BAR:
|
||||
ctx.register = "steady"
|
||||
note = ("Read of the moment: Brian sounds frustrated / on tilt right now. Meet him "
|
||||
"there first — warm, steady, present. Don't clip into logging-shorthand or "
|
||||
"bury him in analysis; settle him, then help. (Still log any facts he hands you.)")
|
||||
elif m.get("sentiment", 0) >= _UP_BAR and m.get("intensity", 0) >= 0.4:
|
||||
ctx.register = "hype"
|
||||
note = "Read of the moment: he's up / energized — match his energy, don't flatten it."
|
||||
if note:
|
||||
m["note"] = note
|
||||
logbus.log("info", "perceived", session=ctx.session_id, kind=m.get("kind"),
|
||||
tilt=m.get("tilt"), sentiment=m.get("sentiment"), register=ctx.register)
|
||||
return ctx
|
||||
|
||||
|
||||
def _compose(ctx: TurnContext) -> TurnContext:
|
||||
"""Assemble the tiered prompt for the voice model."""
|
||||
ctx.messages = build_messages(ctx.session_id, ctx.user_msg, ctx.mode)
|
||||
ctx.messages = build_messages(ctx.session_id, ctx.user_msg, ctx.mode, moment=ctx.moment)
|
||||
return ctx
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,97 @@
|
||||
"""Perceive: read the moment from what Brian just said — cheap, deterministic, no LLM.
|
||||
|
||||
The control plane's senses. A lexicon + signal heuristic that estimates emotional
|
||||
charge (sentiment, intensity, tilt) and the kind of turn (emotional / strategic /
|
||||
meta / build / casual). It's rough on purpose — the point of the society-of-parts
|
||||
design is that *most* parts are free heuristics and the LLM is the exception.
|
||||
|
||||
What it's GOOD at: catching the obvious, action-relevant signal — especially tilt
|
||||
(the mental-game core of her job). What it's NOT: nuanced understanding (that's the
|
||||
LLM's job downstream). `route` turns this read into a per-turn register nudge.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
# Negative / tilt charge — frustration, downswing, mental-game trouble.
|
||||
_NEG = (
|
||||
"tilt", "tilted", "steaming", "steam", "frustrated", "pissed", "angry", "annoyed",
|
||||
"hate", "sick of", "fed up", "card dead", "carddead", "cold deck", "brutal", "cooler",
|
||||
"punt", "punted", "spew", "spewing", "stuck", "losing", "bad beat", "badbeat",
|
||||
"unlucky", "rigged", "sigh", "ugh", "fml", "can't win", "cant win", "miserable",
|
||||
"over it", "fuck this", "hate this", "can't catch", "cant catch",
|
||||
)
|
||||
# Positive / up charge — running good, energized.
|
||||
_POS = (
|
||||
"great", "awesome", "love", "crushing", "running good", "rungood", "hell yeah",
|
||||
"let's go", "lets go", "stoked", "pumped", "feeling good", "on fire", "dialed",
|
||||
"killing it", "in the zone", "so good", "amazing",
|
||||
)
|
||||
_PROFANITY = ("fuck", "fucking", "shit", "damn", "bullshit", "fml")
|
||||
# Strategic / poker-analysis cues.
|
||||
_STRATEGY = (
|
||||
"fold", "call", "raise", "3bet", "three-bet", "range", "equity", "gto", "bluff",
|
||||
"value", "river", "turn", "flop", "preflop", "pot odds", "outs", "should i",
|
||||
"what would you", "sizing", "check-raise", "overbet", "line",
|
||||
)
|
||||
# Meta / about-her cues.
|
||||
_META = (
|
||||
"do you", "are you", "yourself", "conscious", "sentient", "you feel", "you exist",
|
||||
"your thoughts", "your mind", "who are you", "what are you", "your own",
|
||||
)
|
||||
# Building / technical cues.
|
||||
_BUILD = (
|
||||
"code", "function", "bug", "build", "implement", "refactor", "architecture",
|
||||
"prompt", "python", "commit", "deploy", "pipeline", "algorithm", "repo", "api",
|
||||
"schema", "module", "wire it", "the model",
|
||||
)
|
||||
|
||||
|
||||
def _clamp(x: float, lo: float = 0.0, hi: float = 1.0) -> float:
|
||||
return max(lo, min(hi, x))
|
||||
|
||||
|
||||
def _hits(text: str, lexicon: tuple[str, ...]) -> int:
|
||||
"""Count lexicon matches. Multi-token terms match as substrings ('card dead');
|
||||
single words match on word boundaries so 'line' doesn't fire inside 'pipeline'."""
|
||||
n = 0
|
||||
for term in lexicon:
|
||||
if " " in term or "-" in term or "'" in term:
|
||||
n += 1 if term in text else 0
|
||||
else:
|
||||
n += 1 if re.search(rf"\b{re.escape(term)}\b", text) else 0
|
||||
return n
|
||||
|
||||
|
||||
def read(user_msg: str) -> dict:
|
||||
"""Estimate the emotional charge + kind of this turn. Returns
|
||||
{sentiment: -1..1, intensity: 0..1, tilt: 0..1, kind: str}."""
|
||||
t = (user_msg or "").lower()
|
||||
words = re.findall(r"[a-z']+", t)
|
||||
|
||||
neg = _hits(t, _NEG)
|
||||
pos = _hits(t, _POS)
|
||||
prof = _hits(t, _PROFANITY)
|
||||
exclam = user_msg.count("!")
|
||||
caps = sum(1 for w in re.findall(r"[A-Za-z]{2,}", user_msg) if w.isupper())
|
||||
short_and_hot = len(words) <= 6 and (neg or exclam or prof)
|
||||
|
||||
intensity = _clamp(0.2 * exclam + 0.25 * caps + 0.3 * prof + (0.2 if short_and_hot else 0))
|
||||
sentiment = _clamp((pos - neg) * 0.5, -1.0, 1.0)
|
||||
tilt = _clamp(0.35 * neg + 0.5 * intensity) if (neg or prof) else 0.0
|
||||
|
||||
if tilt >= 0.4 or (neg and sentiment < 0):
|
||||
kind = "emotional"
|
||||
elif _hits(t, _STRATEGY):
|
||||
kind = "strategic"
|
||||
elif _hits(t, _META):
|
||||
kind = "meta"
|
||||
elif _hits(t, _BUILD):
|
||||
kind = "build"
|
||||
elif pos and intensity >= 0.3:
|
||||
kind = "emotional" # up/energized still wants an emotional read
|
||||
else:
|
||||
kind = "casual"
|
||||
|
||||
return {"sentiment": round(sentiment, 2), "intensity": round(intensity, 2),
|
||||
"tilt": round(tilt, 2), "kind": kind}
|
||||
@@ -0,0 +1,56 @@
|
||||
"""Perceive: cheap heuristic read of the moment, and route turning it into a nudge."""
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
|
||||
import pytest
|
||||
|
||||
from lyra import perceive
|
||||
|
||||
|
||||
def test_reads_tilt():
|
||||
m = perceive.read("I'm so fucking tilted, card dead all night, this is brutal!!")
|
||||
assert m["tilt"] >= 0.5 and m["sentiment"] < 0 and m["kind"] == "emotional"
|
||||
|
||||
|
||||
def test_reads_strategy_calm():
|
||||
m = perceive.read("Should I fold the river here given his range and the board?")
|
||||
assert m["kind"] == "strategic" and m["tilt"] < 0.4
|
||||
|
||||
|
||||
def test_reads_up_energy():
|
||||
m = perceive.read("Let's go!! crushing it tonight, feeling so good!")
|
||||
assert m["sentiment"] > 0 and m["kind"] == "emotional"
|
||||
|
||||
|
||||
def test_reads_build_and_casual():
|
||||
assert perceive.read("let's refactor the cognition pipeline module").get("kind") == "build"
|
||||
assert perceive.read("ok sounds good to me").get("kind") == "casual"
|
||||
assert perceive.read("ok sounds good to me")["tilt"] == 0.0
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mind(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("LYRA_DB_PATH", str(tmp_path / "test.db"))
|
||||
monkeypatch.setenv("CHAT_DELIBERATE", "false")
|
||||
from lyra import llm
|
||||
monkeypatch.setattr(llm, "embed", lambda texts: [[0.1, 0.2, 0.3] for _ in texts])
|
||||
import lyra.memory as memory
|
||||
importlib.reload(memory)
|
||||
import lyra.mind as mind
|
||||
importlib.reload(mind)
|
||||
memory.ensure_session("s1")
|
||||
return mind
|
||||
|
||||
|
||||
def test_route_injects_tilt_nudge(mind):
|
||||
turn = mind.assemble("s1", "ugh I'm steaming, fucking coolered again!!", "cloud", None)
|
||||
assert turn.register == "steady"
|
||||
sys_blob = " ".join(m["content"] for m in turn.messages if m["role"] == "system")
|
||||
assert "on tilt" in sys_blob.lower() or "frustrated" in sys_blob.lower()
|
||||
|
||||
|
||||
def test_route_quiet_on_neutral_turn(mind):
|
||||
turn = mind.assemble("s1", "what did we decide about the schema yesterday?", "cloud", None)
|
||||
assert turn.register is None # neutral -> no nudge
|
||||
assert not (turn.moment or {}).get("note")
|
||||
Reference in New Issue
Block a user