"""axiom.regenerate — regenerating a narrative variant.
Replays turn `turn_id` with the same player message to produce a **new
variant** of the narrative text (without re-evaluating rules or stats), then
appends it to the turn's multiverse payload in the `Event_Log`
(`{"active": idx, "variants": [...]}`).
Zero Qt dependency. Streaming is surfaced through the `on_token` callback.
"""
from __future__ import annotations
import json
from typing import Callable
from axiom.backends.base import LLMBackend
from axiom.prompts import build_narrative_prompt
from axiom.schema import get_connection
# Mapping verbosité → plafond de tokens (aligné sur l'arbitrator).
_VERBOSITY_TO_TOKENS = {"short": 150, "balanced": 400, "talkative": 1024}
[docs]
def history_to_messages(history: list[dict]) -> list[dict]:
"""Convert the event-sourced history (user_input / narrative_text) into LLM
messages (the active variant is authoritative for the narrative).
"""
messages: list[dict] = []
for h in history:
payload = h.get("payload", "")
if h.get("event_type") == "user_input":
text = payload.get("text", str(payload)) if isinstance(payload, dict) else str(payload)
messages.append({"role": "user", "content": text})
elif h.get("event_type") == "narrative_text":
if isinstance(payload, dict) and "variants" in payload:
text = payload["variants"][payload["active"]]
else:
text = str(payload)
messages.append({"role": "assistant", "content": text})
return messages
[docs]
def regenerate_variant(
llm: LLMBackend,
db_path: str,
save_id: str,
turn_id: int,
history: list[dict],
system_prompt: str,
user_message: str,
temperature: float = 0.7,
top_p: float = 1.0,
verbosity_level: str = "balanced",
player_id: str = "player_1",
on_token: Callable[[str], None] | None = None,
) -> str:
"""Generate an alternative variant of turn `turn_id` and record it.
The new variant is appended to the turn's `narrative_text` payload and
becomes the **active** variant. Returns the generated text.
"""
llm_history = history_to_messages(history)
prompt = build_narrative_prompt(
universe_system_prompt=system_prompt,
entity_stats_block="", # pas de stats : on ne réévalue pas les règles
rag_chunks=[],
history=llm_history,
intents={player_id: user_message},
verbosity_level=verbosity_level,
)
# Pas de tool-call sur une régénération : on ne veut que du texte.
for msg in prompt:
if msg["role"] == "system":
msg["content"] = msg["content"].replace(
"You MUST end your response with a JSON block",
"You are generating a new variant. Do NOT output any JSON tool calls.",
)
stops = ["\nUser:", "\nPlayer:", "\n[User]", "<|eot_id|>",
f"\n{player_id}:", f"\n[{player_id}]"]
max_tokens = _VERBOSITY_TO_TOKENS.get(verbosity_level.lower(), 400)
narrative_text = ""
for token in llm.stream_tokens(
prompt,
temperature=temperature,
top_p=top_p,
stop_sequences=stops,
max_tokens=max_tokens,
):
narrative_text += token
if on_token is not None:
on_token(token)
append_variant(db_path, save_id, turn_id, narrative_text.strip())
return narrative_text
[docs]
def append_variant(db_path: str, save_id: str, turn_id: int, text: str) -> bool:
"""Append `text` as the active variant of a turn's `narrative_text`.
A historical non-multiverse payload is converted on the way. Returns False
when the turn has no narrative event (nothing is written).
"""
with get_connection(db_path) as conn:
row = conn.execute(
"SELECT payload FROM Event_Log WHERE save_id = ? AND turn_id = ? "
"AND event_type = 'narrative_text';",
(save_id, turn_id),
).fetchone()
if not row:
return False
payload = json.loads(row[0])
if not isinstance(payload, dict) or "variants" not in payload:
old = payload.get("text", "") if isinstance(payload, dict) else str(payload)
payload = {"active": 0, "variants": [old]}
payload["variants"].append(text)
payload["active"] = len(payload["variants"]) - 1
conn.execute(
"UPDATE Event_Log SET payload = ? WHERE save_id = ? AND turn_id = ? "
"AND event_type = 'narrative_text';",
(json.dumps(payload), save_id, turn_id),
)
conn.commit()
return True