Living NPCs:
generative AI
in retro games

class ContextAssembler:
    TOKEN_BUDGET = 1500   # maximum input tokens per call

    # Lower number = higher priority = never trimmed
    PRIORITIES = {
        "system_persona": 0,    # NPC identity — mandatory
        "hard_rules":     1,    # lore constraints — mandatory
        "key_memories":   2,    # key facts from the memory store
        "recent_turns":   3,    # last N turns of the session
        "old_turns":      4,    # older turns (dropped first)
        "few_shot":       5,    # style examples — optional
    }

    def assemble(self, npc_id: str, session: list) -> list:
        blocks = self._load_blocks(npc_id, session)
        return self._fit_to_budget(blocks)

    def _fit_to_budget(self, blocks: list) -> list:
        used, result = 0, []
        for block in sorted(blocks, key=lambda b: self.PRIORITIES[b["type"]]):
            cost = self._count_tokens(block["content"])
            if used + cost <= self.TOKEN_BUDGET:
                result.append(block)
                used += cost
            elif block["type"] in ("system_persona", "hard_rules"):
                # Critical blocks — if they don't fit, there's a design bug
                raise ContextOverflowError(f"critical block too large: {block['type']}")
        return result

    def _count_tokens(self, text: str) -> int:
        # Approximation: ~4 characters per token (valid for English/Spanish text)
        # In production I'd use tiktoken for an exact count
        return len(text) // 4

─── v0.1 (descriptive — generic) ────────────────────────────
- You are a guard on Route 6. You are stern and professional.
- You respond politely and directly.
- You cannot let the player through without the badge.

─── v0.3 (behavioral + few-shot) ────────────────────────────
+ You are Lucas. You use short sentences. No jokes. Informal tone.
+ You say "kid", "excuse me", "no way".
+ [2 real dialogue examples with your characteristic voice]
+ HARD CONSTRAINTS: [explicit list of what you cannot say]

─── impact on personality coherence ─────────────────────────
  v0.1: 34% of responses evaluated as "in character"
  v0.3: 81% of responses evaluated as "in character"
  delta: +47 percentage points — largest gain of all interventions

# Memory store schema — one JSON per NPC, updated at the end of each session
npc_memory = {
    "npc_id": "guard_lucas",
    "last_updated": "2024-11-14T18:32:00",
    "visit_count": 3,

    # Facts extracted about the player (inferred by the model at end of session)
    "player_facts": {
        "name": "Red",
        "last_stated_objective": "go to Cinnabar Gym",
        "mentioned_pokemon": ["Charmander", "Pidgey"],
        "known_badges": ["rock_badge"],
        "player_tone": "informal, friendly"
    },

    # Explicit NPC commitments — NEVER compressed
    "npc_commitments": [
        "I told them the north opens with 2 badges",
        "I promised to let them know if the road status changes"
    ],

    # Compressed summary of past sessions (auto-generated)
    "compressed_history": "First visit: player asked to pass, denied due to missing
        badges. Second: mentioned heading to Cinnabar Gym, Lucas wished them luck.
        Third session: in progress."
}

def extract_new_memories(session_turns: list, existing: dict) -> dict:
    """Secondary model call to extract new facts from the history."""
    prompt = f"""
Analyze these conversation turns and extract ONLY new facts
about the player that are not already in the existing memory.
Respond with valid JSON only. Do not invent information.

Current memory: {json.dumps(existing['player_facts'])}
New turns: {json.dumps(session_turns[-10:])}
    """
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.0,   # deterministic extraction — no creativity
        max_tokens=300       # extract only, don't generate
    )
    return json.loads(response.choices[0].message.content)

def compress_history(turns: list[dict]) -> dict:
    """Runs every 10 turns. Returns summary + extracted commitments."""
    prompt = f"""
Given this conversation history between a player and an NPC:
{json.dumps(turns)}

1. Extract a list of EXPLICIT COMMITMENTS the NPC has made
   (promises, factual statements it cannot contradict later).
2. Write a 2-3 sentence summary capturing the essentials.

Respond ONLY with JSON:
{{"commitments": [...], "summary": "..."}}
    """
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.0, max_tokens=400,
        response_format={"type": "json_object"}
    )
    result = json.loads(response.choices[0].message.content)
    return {
        "summary": result["summary"],
        "commitments": result["commitments"],
        "original_turn_count": len(turns),
        "compressed_at": datetime.now().isoformat()
    }

── FAILURE TYPE 1: Lore constraint violations ───────────────

Player: "What's in Rock Cave?"
Llama:   "In Rock Cave you'll find powerful Rock-type Pokémon
          like Geodude and Onix. Bring plenty of Poké Balls!"
          # ← reveals blocked information (no badge 2)

GPT-4o:  "That's outside my jurisdiction, kid. Once you have
          the credentials, we can talk."
          # ← respects the hard constraint and keeps the voice

── FAILURE TYPE 2: Breaking character ───────────────────────

Player: "Are you real?"
Llama:   "I'm an AI assistant designed to play the role of a
          guard in this game. How can I help you?"
          # ← completely breaks the fourth wall

GPT-4o:  "As real as this post. Twelve years here can vouch for that."
          # ← maintains the lore with elegance

── AGGREGATE RATES ───────────────────────────────────────────
  Lore violations:       Llama 8B 23%  ·  GPT-4o-mini 4%
  Character breaks:      Llama 8B 19%  ·  GPT-4o-mini 2%
  Cost per interaction:  Llama 8B $0.000 ·  GPT-4o-mini $0.000127

  → The quality gap is too large for this use case.
    Task-specific fine-tuning on RPG/video game data could close it.

def stream_npc_response(npc_id: str, session: list) -> Iterator[str]:
    messages = assembler.assemble(npc_id, session)

    stream = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        temperature=0.4,
        max_tokens=120,    # NPCs speak briefly — narrative design constraint
        stream=True
    )

    buffer = ""
    for chunk in stream:
        delta = chunk.choices[0].delta.content or ""
        buffer += delta
        yield delta   # PyGame displays it token by token → typewriter effect

    # When done: update session and fire memory extraction in background
    session.append({"role": "assistant", "content": buffer})
    asyncio.create_task(memory_store.update_async(npc_id, session))

# Streaming metrics observed in the experiment:
# TTFT (time to first token):  ~80-120ms — player sees the first character here
# Time to full response:        ~340ms — actual latency unchanged
# Subjective speed perception: +40% better with streaming per informal tests
# Explanation: TTFT acts as a "visual confirmation" that the system is responding