{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Module 4: Build a Word Game Environment\n", "\n", "Build a letter-guessing (Hangman-style) environment from scratch using the OpenEnv pattern.\n", "\n", "**Time:** ~30 min · **Difficulty:** Intermediate · **GPU:** Not required" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "!pip install -q openenv-core\n!git clone --depth=1 -q https://github.com/meta-pytorch/OpenEnv.git 2>/dev/null || true\n\nimport sys, os\nrepo = os.path.abspath('OpenEnv')\nfor p in [repo, os.path.join(repo, 'src')]:\n if p not in sys.path:\n sys.path.insert(0, p)\nprint(\"Setup complete!\")" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 1. Define the Types\n", "\n", "Every OpenEnv environment starts with its data contracts: what actions can you take, what do you observe, what metadata exists?" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from dataclasses import dataclass, field\n", "from typing import List, Optional, Dict, Any\n", "\n", "# These would normally go in models.py\n", "\n", "@dataclass\n", "class WordGameAction:\n", " \"\"\"Player guesses a single letter.\"\"\"\n", " guess: str\n", " metadata: Dict[str, Any] = field(default_factory=dict)\n", "\n", "@dataclass\n", "class WordGameObservation:\n", " \"\"\"What the player sees after each guess.\"\"\"\n", " done: bool\n", " reward: Optional[float]\n", " masked_word: str # e.g., \"p_th_n\"\n", " guessed_letters: List[str] # All letters tried\n", " attempts_remaining: int\n", " message: str # Feedback text\n", " metadata: Dict[str, Any] = field(default_factory=dict)\n", "\n", "@dataclass\n", "class WordGameState:\n", " \"\"\"Episode metadata.\"\"\"\n", " episode_id: Optional[str] = None\n", " step_count: int = 0\n", " target_word: str = \"\"\n", " max_attempts: int = 6\n", "\n", "print(\"Types defined: WordGameAction, WordGameObservation, WordGameState\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 2. Implement the Environment\n", "\n", "The environment implements three methods: `reset()`, `step()`, and `state`. This is where the game logic lives." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import random\nimport uuid\n\nWORDS = [\n \"python\", \"neural\", \"tensor\", \"matrix\", \"vector\",\n \"kernel\", \"lambda\", \"signal\", \"binary\", \"cipher\",\n \"model\", \"layer\", \"epoch\", \"batch\", \"token\",\n]\n\nclass WordGameEnvironment:\n \"\"\"A letter-guessing game environment following the OpenEnv pattern.\"\"\"\n\n def __init__(self):\n self._state = WordGameState()\n self._target = \"\"\n self._guessed = set()\n self._remaining = 6\n\n def reset(self) -> WordGameObservation:\n \"\"\"Start a new episode with a random word.\"\"\"\n self._target = random.choice(WORDS)\n self._guessed = set()\n self._remaining = 10\n self._state = WordGameState(\n episode_id=str(uuid.uuid4()),\n step_count=0,\n target_word=self._target,\n max_attempts=10,\n )\n return WordGameObservation(\n done=False,\n reward=None,\n masked_word=self._mask(),\n guessed_letters=[],\n attempts_remaining=self._remaining,\n message=f\"Guess letters in a {len(self._target)}-letter word!\",\n )\n\n def step(self, action: WordGameAction) -> WordGameObservation:\n \"\"\"Process a letter guess.\"\"\"\n letter = action.guess.lower().strip()\n self._state.step_count += 1\n\n # Already guessed?\n if letter in self._guessed:\n return WordGameObservation(\n done=False,\n reward=0.0,\n masked_word=self._mask(),\n guessed_letters=sorted(self._guessed),\n attempts_remaining=self._remaining,\n message=f\"Already guessed '{letter}'. Try another.\",\n )\n\n self._guessed.add(letter)\n\n if letter in self._target:\n message = f\"'{letter}' is in the word!\"\n else:\n self._remaining -= 1\n message = f\"'{letter}' is not in the word.\"\n\n # Check win/lose\n masked = self._mask()\n won = \"_\" not in masked\n lost = self._remaining <= 0\n done = won or lost\n\n if won:\n reward = 1.0\n message = f\"You got it! The word was '{self._target}'.\"\n elif lost:\n reward = 0.0\n message = f\"Out of attempts. The word was '{self._target}'.\"\n else:\n reward = 0.0\n\n return WordGameObservation(\n done=done,\n reward=reward,\n masked_word=masked,\n guessed_letters=sorted(self._guessed),\n attempts_remaining=self._remaining,\n message=message,\n )\n\n @property\n def state(self) -> WordGameState:\n return self._state\n\n def _mask(self) -> str:\n \"\"\"Show guessed letters, hide the rest.\"\"\"\n return \"\".join(c if c in self._guessed else \"_\" for c in self._target)\n\nprint(\"WordGameEnvironment defined.\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 3. Test the Environment Directly\n", "\n", "Before wiring up HTTP, test the pure game logic." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "env = WordGameEnvironment()\n", "obs = env.reset()\n", "print(f\"Word: {obs.masked_word} ({len(obs.masked_word)} letters)\")\n", "print(f\"Message: {obs.message}\")\n", "print(f\"Attempts: {obs.attempts_remaining}\")\n", "print()\n", "\n", "# Play with common letters\n", "for letter in [\"e\", \"a\", \"t\", \"n\", \"o\", \"r\", \"s\", \"i\", \"l\"]:\n", " if obs.done:\n", " break\n", " obs = env.step(WordGameAction(guess=letter))\n", " print(f\" Guess '{letter}': {obs.masked_word} ({obs.message})\")\n", "\n", "print(f\"\\nFinal: reward={obs.reward}, done={obs.done}\")\n", "print(f\"State: episode={env.state.episode_id[:8]}..., steps={env.state.step_count}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 4. Write Policies\n", "\n", "Let's write two policies and compare them." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import string\n", "\n", "class RandomLetterPolicy:\n", " \"\"\"Guess random unused letters.\"\"\"\n", " name = \"Random\"\n", "\n", " def select_action(self, obs: WordGameObservation) -> WordGameAction:\n", " available = [c for c in string.ascii_lowercase if c not in obs.guessed_letters]\n", " return WordGameAction(guess=random.choice(available))\n", "\n", "\n", "class FrequencyPolicy:\n", " \"\"\"Guess by English letter frequency.\"\"\"\n", " name = \"Frequency\"\n", " FREQ_ORDER = \"etaoinshrdlcumwfgypbvkjxqz\"\n", "\n", " def select_action(self, obs: WordGameObservation) -> WordGameAction:\n", " for letter in self.FREQ_ORDER:\n", " if letter not in obs.guessed_letters:\n", " return WordGameAction(guess=letter)\n", " return WordGameAction(guess=\"a\") # fallback\n", "\n", "\n", "def evaluate(env, policy, episodes=100):\n", " wins = 0\n", " total_steps = 0\n", " for _ in range(episodes):\n", " obs = env.reset()\n", " while not obs.done:\n", " action = policy.select_action(obs)\n", " obs = env.step(action)\n", " if obs.reward and obs.reward > 0:\n", " wins += 1\n", " total_steps += env.state.step_count\n", " return wins / episodes, total_steps / episodes\n", "\n", "\n", "env = WordGameEnvironment()\n", "\n", "for policy in [RandomLetterPolicy(), FrequencyPolicy()]:\n", " win_rate, avg_steps = evaluate(env, policy)\n", " print(f\"{policy.name:15s} — Win rate: {win_rate*100:.1f}%, Avg steps: {avg_steps:.1f}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": "Frequency should significantly outperform random. With technical vocabulary and individual letter guessing, both win rates are modest — but Frequency is typically 5–10× better than Random. Increase `max_attempts` in `WordGameEnvironment` (e.g. to 15) to see higher absolute win rates." }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 5. Wire Up FastAPI\n", "\n", "In a real deployment, you'd create `server/app.py` with:\n", "\n", "```python\n", "from openenv.core.env_server import create_fastapi_app\n", "from environment import WordGameEnvironment\n", "\n", "app = create_fastapi_app(WordGameEnvironment)\n", "```\n", "\n", "That single call creates all endpoints: `/ws`, `/reset`, `/step`, `/state`, `/health`, `/web`, `/docs`.\n", "\n", "Let's simulate the server locally to demonstrate the full stack." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Write the environment files to disk for deployment\n", "import os\n", "\n", "os.makedirs('word_game/server', exist_ok=True)\n", "\n", "# models.py — uses Pydantic (Action, Observation, State are Pydantic BaseModel subclasses)\n", "models_code = '''\n", "from typing import List, Optional\n", "from openenv.core.env_server import Action, Observation, State\n", "\n", "\n", "class WordGameAction(Action):\n", " \"\"\"Player guesses a single letter.\"\"\"\n", " guess: str\n", "\n", "\n", "class WordGameObservation(Observation):\n", " \"\"\"What the player sees after each guess.\n", "\n", " Note: done and reward are inherited from Observation.\n", " \"\"\"\n", " masked_word: str # e.g. \"p_th_n\"\n", " guessed_letters: List[str] # All letters tried\n", " attempts_remaining: int\n", " message: str # Feedback text\n", "\n", "\n", "class WordGameState(State):\n", " \"\"\"Episode metadata.\n", "\n", " Note: episode_id and step_count are inherited from State.\n", " \"\"\"\n", " target_word: str = \"\"\n", " max_attempts: int = 6\n", "'''\n", "\n", "with open('word_game/models.py', 'w') as f:\n", " f.write(models_code)\n", "\n", "# client.py — uses EnvClient (WebSocket-based)\n", "client_code = '''\n", "from openenv.core.env_client import EnvClient\n", "from openenv.core.client_types import StepResult\n", "from .models import WordGameAction, WordGameObservation, WordGameState\n", "\n", "\n", "class WordGameEnv(EnvClient[WordGameAction, WordGameObservation, WordGameState]):\n", " def _step_payload(self, action: WordGameAction) -> dict:\n", " return {\"guess\": action.guess}\n", "\n", " def _parse_result(self, payload: dict) -> StepResult:\n", " obs_data = payload.get(\"observation\", {})\n", " return StepResult(\n", " observation=WordGameObservation(\n", " done=payload.get(\"done\", False),\n", " reward=payload.get(\"reward\"),\n", " masked_word=obs_data.get(\"masked_word\", \"\"),\n", " guessed_letters=obs_data.get(\"guessed_letters\", []),\n", " attempts_remaining=obs_data.get(\"attempts_remaining\", 0),\n", " message=obs_data.get(\"message\", \"\"),\n", " ),\n", " reward=payload.get(\"reward\"),\n", " done=payload.get(\"done\", False),\n", " )\n", "\n", " def _parse_state(self, payload: dict) -> WordGameState:\n", " return WordGameState(\n", " episode_id=payload.get(\"episode_id\"),\n", " step_count=payload.get(\"step_count\", 0),\n", " target_word=payload.get(\"target_word\", \"\"),\n", " max_attempts=payload.get(\"max_attempts\", 6),\n", " )\n", "'''\n", "\n", "with open('word_game/client.py', 'w') as f:\n", " f.write(client_code)\n", "\n", "# server/app.py\n", "app_code = '''\n", "from openenv.core.env_server import create_fastapi_app\n", "from ..models import WordGameAction, WordGameObservation\n", "from .environment import WordGameEnvironment\n", "\n", "app = create_fastapi_app(WordGameEnvironment, WordGameAction, WordGameObservation)\n", "'''\n", "\n", "with open('word_game/server/app.py', 'w') as f:\n", " f.write(app_code)\n", "\n", "print('Created word_game/models.py (Pydantic models)')\n", "print('Created word_game/client.py (EnvClient subclass)')\n", "print('Created word_game/server/app.py')\n", "print()\n", "print('Next steps:')\n", "print(' 1. Add server/environment.py with WordGameEnvironment class')\n", "print(' 2. Test locally: uvicorn word_game.server.app:app --reload')\n", "print(' 3. Deploy: openenv push --repo-id username/word-game')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 6. The Client\n\nThe client translates between your typed models and JSON over the wire. Three methods:\n\n```python\nclass WordGameEnv(EnvClient[WordGameAction, WordGameObservation, WordGameState]):\n def _step_payload(self, action):\n return {\"guess\": action.guess}\n\n def _parse_result(self, payload):\n return StepResult(\n observation=WordGameObservation(**payload),\n reward=payload.get(\"reward\", 0),\n done=payload[\"done\"],\n )\n\n def _parse_state(self, payload):\n return WordGameState(**payload)\n```\n\nUsers of your environment would then write:\n\n```python\nfrom word_game import WordGameEnv, WordGameAction\n\nwith WordGameEnv(base_url=\"https://username-word-game.hf.space\").sync() as env:\n result = env.reset()\n result = env.step(WordGameAction(guess=\"e\"))\n print(result.observation.masked_word)\n```" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 7. Scaffold with `openenv init`\n", "\n", "Instead of writing everything by hand, use the CLI:\n", "\n", "```bash\n", "openenv init word_game\n", "cd word_game\n", "# Edit models.py, server/environment.py, client.py\n", "uv run server # Test locally\n", "openenv push # Deploy to HF Spaces\n", "```\n", "\n", "This creates the full directory structure. You fill in your types and game logic." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Summary\n", "\n", "You built a complete OpenEnv environment:\n", "\n", "| File | What it does | Lines of code |\n", "|------|-------------|---------------|\n", "| `models.py` | Action, Observation, State types | ~30 |\n", "| `server/environment.py` | Game logic (reset, step, state) | ~60 |\n", "| `client.py` | HTTP client (3 parsing methods) | ~25 |\n", "| `server/app.py` | FastAPI wiring | ~3 |\n", "\n", "The pattern is always the same: **types → server logic → client → container**.\n", "\n", "**Next:** [Module 5](../module-5/README.md) — Training a model to play games with GRPO." ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3.11.0" } }, "nbformat": 4, "nbformat_minor": 4 }