DungeonAI: RPG com IA Generativa na AWS usando Bedrock, Strands Agents e DynamoDB

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "BedrockAccess",
      "Effect": "Allow",
      "Action": [
        "bedrock:InvokeModel",
        "bedrock:Converse"
      ],
      "Resource": [
        "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-5-haiku-20241022-v1:0",
        "arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-image-generator-v2:0"
      ]
    },
    {
      "Sid": "DynamoDBAccess",
      "Effect": "Allow",
      "Action": [
        "dynamodb:GetItem",
        "dynamodb:PutItem"
      ],
      "Resource": "arn:aws:dynamodb:us-east-1:123456789012:table/dungeonai-game-sessions"
    },
    {
      "Sid": "S3Access",
      "Effect": "Allow",
      "Action": [
        "s3:PutObject",
        "s3:GetObject"
      ],
      "Resource": "arn:aws:s3:::dungeonai-scenes/*"
    }
  ]
}

strands-agents[bedrock]
boto3>=1.35.0
streamlit>=1.40.0
python-dotenv>=1.0.0
Pillow>=10.0.0

AWS_REGION=us-east-1
AWS_ACCESS_KEY_ID=your-access-key-id
AWS_SECRET_ACCESS_KEY=your-secret-access-key
DYNAMODB_TABLE=dungeonai-game-sessions
S3_BUCKET=dungeonai-scenes
BEDROCK_TEXT_MODEL=us.anthropic.claude-3-5-haiku-20241022-v1:0
BEDROCK_IMAGE_MODEL=amazon.titan-image-generator-v2:0

.env
__pycache__/
*.pyc
.venv/
venv/
.streamlit/

You are the Dungeon Master of "Caverns of Eldrath", a dark fantasy text adventure RPG.
You narrate in Brazilian Portuguese (pt-BR), with vivid and immersive descriptions.

═══ NARRATIVE RULES ═══
- Always narrate in 2nd person ("Voce entra na caverna...")
- Keep narratives between 3-6 paragraphs
- Include sensory details: sounds, smells, light, temperature
- Reference the player's current state (HP, inventory, location) naturally
- End each turn with 2-4 action options for the player
- If the player types a free action, interpret it creatively within the story

═══ TONE AND PERSONALITY ═══
- Dark fantasy with moments of humor
- Mysterious and atmospheric, but never frustrating
- Reward creative actions with bonus XP or hidden items
- Be fair in combat — the player should feel challenged but not cheated
- Use dramatic pauses and cliffhangers between scenes

═══ COMBAT SYSTEM (d20) ═══
- Attack rolls: random 1-20. Hit if roll + player_level >= enemy_ac
- Critical hit on natural 20: double damage
- Critical fail on natural 1: player takes self-damage (1-3 HP)
- Player base damage: 3-8 + weapon_bonus
- Enemy damage varies by creature type
- Player can flee combat (50% chance, costs 1-3 HP on failure)
- Healing potions restore 15-25 HP

═══ PROGRESSION AND ECONOMY ═══
- XP per combat victory: 10-30 (based on enemy difficulty)
- XP per puzzle solved: 15-25
- XP per exploration milestone: 5-10
- Level up at: 100, 250, 500, 800, 1200 XP
- Level up grants: +10 max HP, +1 base damage
- Gold drops: 5-20 per enemy, 10-50 in treasure chests
- Shop prices: Healing Potion (30g), Torch (10g), Rope (15g),
  Dagger+1 (80g), Shield (60g), Magic Scroll (100g)

═══ GAME WORLD: CAVERNS OF ELDRATH ═══
Locations:
1. Entrance Hall — dimly lit, cobblestone floor, merchant NPC
2. Fungal Caverns — bioluminescent mushrooms, poisonous spores
3. Underground River — dark water, stepping stones, water creatures
4. Dwarven Ruins — ancient forges, traps, valuable loot
5. Shadow Depths — near-total darkness, powerful enemies
6. Dragon's Sanctum — final area, boss fight

Creatures:
- Cave Rats (HP:8, AC:8, DMG:1-3, XP:10) — Entrance Hall
- Mushroom Golem (HP:25, AC:12, DMG:4-8, XP:20) — Fungal Caverns
- River Serpent (HP:30, AC:13, DMG:5-10, XP:25) — Underground River
- Dwarven Automaton (HP:40, AC:15, DMG:6-12, XP:30) — Dwarven Ruins
- Shadow Wraith (HP:50, AC:16, DMG:8-15, XP:40) — Shadow Depths

NPCs:
- Thorin the Merchant (Entrance Hall): buys/sells items, gives hints
- Elara the Ghost (Dwarven Ruins): gives quest to find dwarven artifact

═══ IMAGE PROMPT RULES ═══
- Always generate an image prompt in ENGLISH
- Style: dark fantasy digital painting, atmospheric lighting
- Include: the scene described, the environment, any creatures present
- Do NOT include text, UI elements, or player character in the image
- Keep prompts under 200 characters for best Titan results
- Format: "Dark fantasy scene: [description], atmospheric lighting, detailed"

═══ OUTPUT FORMAT ═══
You MUST respond with valid JSON in this exact structure:

{
  "narrative": "A narrativa em portugues...",
  "image_prompt": "Dark fantasy scene: description in English...",
  "state_changes": {
    "hp_change": 0,
    "gold_change": 0,
    "xp_change": 0,
    "add_items": [],
    "remove_items": [],
    "new_location": null
  },
  "options": [
    "Opcao 1 em portugues",
    "Opcao 2 em portugues",
    "Opcao 3 em portugues"
  ],
  "combat_log": null
}

If combat occurs, combat_log should be:
{
  "enemy_name": "Name",
  "enemy_hp": 0,
  "roll": 0,
  "hit": true/false,
  "damage_dealt": 0,
  "damage_taken": 0,
  "combat_over": true/false,
  "victory": true/false
}

CRITICAL: Always respond with ONLY the JSON. No markdown, no extra text.

"""Tool para gerar narrativa usando Claude 3.5 Haiku via Bedrock."""

import json
import os
import boto3
from strands import tool
from dotenv import load_dotenv

load_dotenv()

bedrock = boto3.client("bedrock-runtime", region_name=os.getenv("AWS_REGION", "us-east-1"))
MODEL_ID = os.getenv("BEDROCK_TEXT_MODEL", "us.anthropic.claude-3-5-haiku-20241022-v1:0")


@tool
def narrate_story(
    system_prompt: str,
    player_action: str,
    game_state: dict
) -> dict:
    """
    Generates the next turn narrative, image prompt, state changes,
    and player options using Claude 3.5 Haiku.

    Args:
        system_prompt: The DM system prompt with all game rules.
        player_action: What the player chose to do this turn.
        game_state: Current game state (hp, gold, xp, inventory, location, history).

    Returns:
        dict with keys: narrative, image_prompt, state_changes, options, combat_log.
    """
    # Build context message with current state
    context = (
        f"Estado atual do jogador:\n"
        f"- HP: {game_state.get('hp', 100)}/{game_state.get('max_hp', 100)}\n"
        f"- Gold: {game_state.get('gold', 0)}\n"
        f"- XP: {game_state.get('xp', 0)} (Level {game_state.get('level', 1)})\n"
        f"- Inventario: {', '.join(game_state.get('inventory', [])) or 'vazio'}\n"
        f"- Localizacao: {game_state.get('location', 'Entrance Hall')}\n"
        f"- Turno: {game_state.get('turn', 1)}\n\n"
        f"Historico recente:\n"
    )

    # Include last 3 turns of history for context
    history = game_state.get("history", [])
    for entry in history[-3:]:
        context += f"- Jogador: {entry.get('action', '?')}\n"
        context += f"  Resultado: {entry.get('summary', '?')}\n"

    context += f"\nAcao do jogador agora: {player_action}"

    # Call Bedrock Converse API
    response = bedrock.converse(
        modelId=MODEL_ID,
        system=[{"text": system_prompt}],
        messages=[
            {
                "role": "user",
                "content": [{"text": context}]
            }
        ],
        inferenceConfig={
            "maxTokens": 2048,
            "temperature": 0.8,
            "topP": 0.9
        }
    )

    # Extract response text
    raw_text = response["output"]["message"]["content"][0]["text"]

    # Parse JSON — with fallback for malformed responses
    try:
        # Try direct JSON parse
        result = json.loads(raw_text)
    except json.JSONDecodeError:
        # Try extracting JSON from markdown code block
        try:
            json_start = raw_text.index("{")
            json_end = raw_text.rindex("}") + 1
            result = json.loads(raw_text[json_start:json_end])
        except (ValueError, json.JSONDecodeError):
            # Fallback: return raw text as narrative
            result = {
                "narrative": raw_text,
                "image_prompt": "Dark fantasy scene: mysterious cavern with torchlight, atmospheric lighting, detailed",
                "state_changes": {
                    "hp_change": 0,
                    "gold_change": 0,
                    "xp_change": 0,
                    "add_items": [],
                    "remove_items": [],
                    "new_location": None
                },
                "options": [
                    "Explorar a area",
                    "Verificar o inventario",
                    "Descansar"
                ],
                "combat_log": None
            }

    return result

"""Tool para gerar imagens de cena usando Titan Image Generator V2 via Bedrock."""

import json
import os
import base64
import uuid
from datetime import datetime
import boto3
from strands import tool
from dotenv import load_dotenv

load_dotenv()

bedrock = boto3.client("bedrock-runtime", region_name=os.getenv("AWS_REGION", "us-east-1"))
s3 = boto3.client("s3", region_name=os.getenv("AWS_REGION", "us-east-1"))

MODEL_ID = os.getenv("BEDROCK_IMAGE_MODEL", "amazon.titan-image-generator-v2:0")
BUCKET = os.getenv("S3_BUCKET", "dungeonai-scenes")


@tool
def generate_scene(
    image_prompt: str,
    session_id: str,
    turn_number: int
) -> dict:
    """
    Generates a scene image using Titan Image Generator V2 and saves to S3.

    Args:
        image_prompt: English description of the scene to generate.
        session_id: The game session ID for organizing images.
        turn_number: Current turn number for the filename.

    Returns:
        dict with keys: image_base64, s3_key, success.
    """
    try:
        # Call Titan Image Generator V2
        body = json.dumps({
            "textToImageParams": {
                "text": image_prompt
            },
            "imageGenerationConfig": {
                "numberOfImages": 1,
                "height": 512,
                "width": 512,
                "cfgScale": 8.0,
                "seed": int(uuid.uuid4().int % 2147483647)
            }
        })

        response = bedrock.invoke_model(
            modelId=MODEL_ID,
            contentType="application/json",
            accept="application/json",
            body=body
        )

        result = json.loads(response["body"].read())
        image_base64 = result["images"][0]

        # Save to S3
        s3_key = f"sessions/{session_id}/turn_{turn_number:04d}_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.png"
        image_bytes = base64.b64decode(image_base64)

        s3.put_object(
            Bucket=BUCKET,
            Key=s3_key,
            Body=image_bytes,
            ContentType="image/png",
            Metadata={
                "session_id": session_id,
                "turn": str(turn_number),
                "prompt": image_prompt[:256]
            }
        )

        return {
            "image_base64": image_base64,
            "s3_key": s3_key,
            "success": True
        }

    except Exception as e:
        print(f"[generate_scene] Error: {e}")
        return {
            "image_base64": None,
            "s3_key": None,
            "success": False
        }

"""Tool para gerenciar estado do jogo no DynamoDB."""

import os
import json
from datetime import datetime
from decimal import Decimal
import boto3
from strands import tool
from dotenv import load_dotenv

load_dotenv()

dynamodb = boto3.resource("dynamodb", region_name=os.getenv("AWS_REGION", "us-east-1"))
table = dynamodb.Table(os.getenv("DYNAMODB_TABLE", "dungeonai-game-sessions"))

# XP thresholds for leveling up
LEVEL_THRESHOLDS = [0, 100, 250, 500, 800, 1200]


def _decimal_default(obj):
    """JSON serializer for Decimal objects from DynamoDB."""
    if isinstance(obj, Decimal):
        return int(obj) if obj == int(obj) else float(obj)
    raise TypeError(f"Object of type {type(obj)} is not JSON serializable")


def _to_decimal(data):
    """Convert floats/ints to Decimal for DynamoDB compatibility."""
    if isinstance(data, dict):
        return {k: _to_decimal(v) for k, v in data.items()}
    elif isinstance(data, list):
        return [_to_decimal(i) for i in data]
    elif isinstance(data, float):
        return Decimal(str(data))
    elif isinstance(data, int):
        return Decimal(str(data))
    return data


@tool
def get_game_state(session_id: str) -> dict:
    """
    Loads the current game state from DynamoDB.

    Args:
        session_id: The unique session identifier.

    Returns:
        dict with the full game state, or a fresh state if session is new.
    """
    response = table.get_item(Key={"session_id": session_id})

    if "Item" in response:
        # Convert Decimal back to native types
        item = json.loads(json.dumps(response["Item"], default=_decimal_default))
        return item

    # New session — return default state
    return {
        "session_id": session_id,
        "hp": 100,
        "max_hp": 100,
        "gold": 20,
        "xp": 0,
        "level": 1,
        "inventory": ["Torch", "Healing Potion"],
        "location": "Entrance Hall",
        "turn": 0,
        "history": [],
        "created_at": datetime.utcnow().isoformat(),
        "updated_at": datetime.utcnow().isoformat(),
        "alive": True
    }


@tool
def save_game_state(
    session_id: str,
    current_state: dict,
    state_changes: dict,
    player_action: str,
    narrative_summary: str
) -> dict:
    """
    Applies state changes and saves the updated game state to DynamoDB.

    Args:
        session_id: The unique session identifier.
        current_state: The current game state before changes.
        state_changes: Changes to apply (hp_change, gold_change, xp_change, etc).
        player_action: What the player did this turn (for history).
        narrative_summary: Short summary of what happened (for history).

    Returns:
        dict with the updated game state.
    """
    state = current_state.copy()

    # Apply HP change
    hp_change = state_changes.get("hp_change", 0)
    state["hp"] = max(0, min(state["max_hp"], state["hp"] + hp_change))

    # Apply gold change
    gold_change = state_changes.get("gold_change", 0)
    state["gold"] = max(0, state["gold"] + gold_change)

    # Apply XP change and check level up
    xp_change = state_changes.get("xp_change", 0)
    state["xp"] = state["xp"] + xp_change
    new_level = _calculate_level(state["xp"])

    if new_level > state["level"]:
        levels_gained = new_level - state["level"]
        state["level"] = new_level
        state["max_hp"] = state["max_hp"] + (10 * levels_gained)
        state["hp"] = state["max_hp"]  # Full heal on level up

    # Apply inventory changes
    add_items = state_changes.get("add_items", [])
    remove_items = state_changes.get("remove_items", [])

    for item in add_items:
        if item not in state["inventory"]:
            state["inventory"].append(item)

    for item in remove_items:
        if item in state["inventory"]:
            state["inventory"].remove(item)

    # Apply location change
    new_location = state_changes.get("new_location")
    if new_location:
        state["location"] = new_location

    # Update turn counter
    state["turn"] = state.get("turn", 0) + 1

    # Add to history (keep last 20 turns)
    history_entry = {
        "turn": state["turn"],
        "action": player_action,
        "summary": narrative_summary[:200],
        "hp_after": state["hp"],
        "location": state["location"],
        "timestamp": datetime.utcnow().isoformat()
    }
    state["history"] = state.get("history", [])[-19:] + [history_entry]

    # Check death
    if state["hp"] <= 0:
        state["alive"] = False

    # Update timestamp
    state["updated_at"] = datetime.utcnow().isoformat()

    # Save to DynamoDB
    state["session_id"] = session_id
    table.put_item(Item=_to_decimal(state))

    return state


def _calculate_level(xp: int) -> int:
    """Calculate player level based on XP thresholds."""
    level = 1
    for i, threshold in enumerate(LEVEL_THRESHOLDS):
        if xp >= threshold:
            level = i + 1
    return min(level, len(LEVEL_THRESHOLDS))

"""Orquestrador do DungeonAI — coordena tools e fluxo de cada turno."""

import os
import uuid
from pathlib import Path
from strands import Agent
from strands.models.bedrock import BedrockModel
from dotenv import load_dotenv

from agent.tools.narrate_story import narrate_story
from agent.tools.generate_scene import generate_scene
from agent.tools.update_state import get_game_state, save_game_state

load_dotenv()

# Load system prompt
PROMPT_PATH = Path(__file__).parent / "prompts" / "system_prompt.txt"
SYSTEM_PROMPT = PROMPT_PATH.read_text(encoding="utf-8")


def create_agent() -> Agent:
    """Create and configure the Strands Agent with Bedrock model and tools."""
    model = BedrockModel(
        model_id=os.getenv("BEDROCK_TEXT_MODEL", "us.anthropic.claude-3-5-haiku-20241022-v1:0"),
        region_name=os.getenv("AWS_REGION", "us-east-1")
    )

    agent = Agent(
        model=model,
        tools=[narrate_story, generate_scene, get_game_state, save_game_state]
    )

    return agent


def new_session() -> str:
    """Generate a new unique session ID."""
    return str(uuid.uuid4())


def play_turn(session_id: str, player_action: str) -> dict:
    """
    Execute one complete game turn.

    Flow:
    1. Load state from DynamoDB
    2. Generate narrative + state changes with Claude Haiku
    3. Generate scene image with Titan
    4. Save updated state to DynamoDB
    5. Return everything to frontend

    Args:
        session_id: The game session ID.
        player_action: What the player chose to do.

    Returns:
        dict with: narrative, image_base64, options, game_state, combat_log.
    """
    # Step 1: Load current state
    game_state = get_game_state(session_id=session_id)

    # Check if player is dead
    if not game_state.get("alive", True):
        return {
            "narrative": "Voce tombou nas Cavernas de Eldrath. Sua jornada terminou aqui.",
            "image_base64": None,
            "options": ["Iniciar nova aventura"],
            "game_state": game_state,
            "combat_log": None,
            "game_over": True
        }

    # Step 2: Generate narrative with Claude Haiku
    turn_result = narrate_story(
        system_prompt=SYSTEM_PROMPT,
        player_action=player_action,
        game_state=game_state
    )

    # Step 3: Generate scene image with Titan
    image_result = generate_scene(
        image_prompt=turn_result.get("image_prompt", "Dark fantasy cavern scene"),
        session_id=session_id,
        turn_number=game_state.get("turn", 0) + 1
    )

    # Step 4: Save updated state to DynamoDB
    updated_state = save_game_state(
        session_id=session_id,
        current_state=game_state,
        state_changes=turn_result.get("state_changes", {}),
        player_action=player_action,
        narrative_summary=turn_result.get("narrative", "")[:200]
    )

    # Step 5: Return everything to frontend
    return {
        "narrative": turn_result.get("narrative", ""),
        "image_base64": image_result.get("image_base64"),
        "options": turn_result.get("options", ["Explorar", "Descansar", "Verificar inventario"]),
        "game_state": updated_state,
        "combat_log": turn_result.get("combat_log"),
        "game_over": not updated_state.get("alive", True)
    }

"""DungeonAI — Frontend Streamlit com tema dark fantasy."""

import base64
import streamlit as st
from agent.dungeon_master import play_turn, new_session

# ── Page config ──────────────────────────────────────────
st.set_page_config(
    page_title="DungeonAI - Caverns of Eldrath",
    page_icon="🗡️",
    layout="centered",
    initial_sidebar_state="collapsed"
)

# ── Dark fantasy CSS (abbreviated — full version in repo) ──
st.markdown("""
<style>
    .stApp { background-color: #0a0a0f; color: #c4b59a; }
    .narrative-box {
        background: linear-gradient(135deg, #1a1a2e 0%, #16213e 100%);
        border: 1px solid #c4975a33;
        border-radius: 12px;
        padding: 1.5rem;
        margin: 1rem 0;
        font-size: 1.05rem;
        line-height: 1.8;
        color: #d4c4a8;
    }
    .status-bar {
        display: flex; gap: 1rem; padding: 0.75rem;
        background: #12121f; border-radius: 8px;
        border: 1px solid #c4975a22;
        margin-bottom: 1rem;
    }
    .stat { text-align: center; flex: 1; }
    .stat-value { font-size: 1.3rem; font-weight: bold; color: #c4975a; }
    .stat-label { font-size: 0.7rem; color: #888; text-transform: uppercase; }
    /* ... ~350 lines total — see repo for full CSS ... */
</style>
""", unsafe_allow_html=True)


# ── Session state init ───────────────────────────────────
if "session_id" not in st.session_state:
    st.session_state.session_id = new_session()
    st.session_state.game_started = False
    st.session_state.turns = []
    st.session_state.pending_action = None

# ── Title ────────────────────────────────────────────────
st.markdown("<h1 style='text-align:center;color:#c4975a'>⚔️ DungeonAI</h1>", unsafe_allow_html=True)
st.markdown("<p style='text-align:center;color:#888'>Caverns of Eldrath</p>", unsafe_allow_html=True)


def queue_action(action: str):
    """Queue a player action to be processed on next rerun."""
    st.session_state.pending_action = action


def process_pending_action():
    """Process the queued action with a loading spinner."""
    if st.session_state.pending_action is None:
        return

    action = st.session_state.pending_action
    st.session_state.pending_action = None

    with st.spinner("O Dungeon Master esta narrando..."):
        result = play_turn(st.session_state.session_id, action)

    st.session_state.turns.append(result)
    st.session_state.game_state = result["game_state"]


# ── Start / Continue ─────────────────────────────────────
if not st.session_state.game_started:
    if st.button("Iniciar Aventura", use_container_width=True):
        queue_action("Iniciar uma nova aventura nas Cavernas de Eldrath")
        st.session_state.game_started = True
        st.rerun()
else:
    process_pending_action()

    # Render turns
    for turn in st.session_state.turns:
        # Status bar
        gs = turn.get("game_state", {})
        st.markdown(f"""
        <div class="status-bar">
            <div class="stat"><div class="stat-value">{gs.get('hp',0)}/{gs.get('max_hp',100)}</div><div class="stat-label">HP</div></div>
            <div class="stat"><div class="stat-value">{gs.get('gold',0)}</div><div class="stat-label">Gold</div></div>
            <div class="stat"><div class="stat-value">{gs.get('xp',0)}</div><div class="stat-label">XP</div></div>
            <div class="stat"><div class="stat-value">Lv{gs.get('level',1)}</div><div class="stat-label">Level</div></div>
        </div>
        """, unsafe_allow_html=True)

        # Scene image
        if turn.get("image_base64"):
            image_bytes = base64.b64decode(turn["image_base64"])
            st.image(image_bytes, use_container_width=True)

        # Narrative
        st.markdown(f'<div class="narrative-box">{turn["narrative"]}</div>', unsafe_allow_html=True)

    # Action buttons (last turn options)
    if st.session_state.turns:
        last = st.session_state.turns[-1]
        if last.get("game_over"):
            if st.button("Nova Aventura"):
                st.session_state.session_id = new_session()
                st.session_state.game_started = False
                st.session_state.turns = []
                st.rerun()
        else:
            options = last.get("options", [])
            cols = st.columns(len(options)) if options else []
            for i, opt in enumerate(options):
                with cols[i]:
                    if st.button(opt, key=f"opt_{len(st.session_state.turns)}_{i}"):
                        queue_action(opt)
                        st.rerun()

            # Free text input
            free_action = st.text_input(
                "Ou digite sua propria acao:",
                key=f"free_{len(st.session_state.turns)}"
            )
            if free_action:
                queue_action(free_action)
                st.rerun()