Joseph Doherty
73 lines
2.2 KiB
Python
73 lines
2.2 KiB
Python
"""Event-lifecycle detection (T52).
|
|
|
|
After each turn, classify whether any active events transitioned
|
|
(started, completed, cancelled). Conservative bias — most turns
|
|
return empty. T61 turn flow appends one event_started/completed/
|
|
cancelled per transition via append_and_apply.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
from pydantic import BaseModel, Field
|
|
|
|
from chat.llm.classify import classify
|
|
from chat.llm.client import LLMClient
|
|
|
|
|
|
class EventTransition(BaseModel):
|
|
event_id: str
|
|
new_status: str # "active" | "completed" | "cancelled"
|
|
reason: str = ""
|
|
|
|
|
|
class EventLifecycleDecision(BaseModel):
|
|
transitions: list[EventTransition] = Field(default_factory=list)
|
|
|
|
|
|
_SYSTEM = (
|
|
"You decide whether any active events transitioned this turn. "
|
|
"STRONGLY default to empty transitions — most turns do NOT resolve "
|
|
"or start a known event. Output only transitions that the narrative "
|
|
"text clearly resolves or starts. Each transition MUST reference an "
|
|
"event_id from the active_events list. new_status is one of "
|
|
"'active' (planned -> active), 'completed', or 'cancelled'. "
|
|
"Output strict JSON matching the schema."
|
|
)
|
|
|
|
|
|
async def detect_event_transitions(
|
|
client: LLMClient,
|
|
*,
|
|
classifier_model: str,
|
|
narrative_text: str,
|
|
active_events: list[dict], # [{event_id, kind, status, props}, ...]
|
|
timeout_s: float = 30.0,
|
|
) -> EventLifecycleDecision:
|
|
"""Classify event transitions for the latest turn. Empty active_events
|
|
short-circuits without an LLM call."""
|
|
if not active_events:
|
|
return EventLifecycleDecision()
|
|
|
|
user_lines = ["Active events:"]
|
|
for ev in active_events:
|
|
user_lines.append(
|
|
f"- event_id={ev['event_id']} kind={ev['kind']} "
|
|
f"status={ev['status']} props={ev.get('props', {})}"
|
|
)
|
|
user_lines.append("")
|
|
user_lines.append("Latest narrative:")
|
|
user_lines.append(narrative_text.strip())
|
|
user = "\n".join(user_lines)
|
|
|
|
return await classify(
|
|
client,
|
|
model=classifier_model,
|
|
system=_SYSTEM,
|
|
user=user,
|
|
schema=EventLifecycleDecision,
|
|
default=EventLifecycleDecision(),
|
|
timeout_s=timeout_s,
|
|
)
|
|
|
|
|
|
__all__ = ["EventTransition", "EventLifecycleDecision", "detect_event_transitions"]
|