Spaces:
Running
Running
File size: 13,612 Bytes
0e61be5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 | """
Game engine for running Wikipedia speedruns with optional visualization.
Supports both headless execution (fast) and Playwright visualization (shows
the actual Wikipedia page with links highlighted).
"""
from __future__ import annotations
import logging
import time
from typing import TYPE_CHECKING
from src.agents.base import AgentContext
from src.config import (
MAX_STEPS,
PLAYWRIGHT_HEADLESS,
PLAYWRIGHT_TIMEOUT,
WIKIPEDIA_BASE_URL,
)
from src.game.state import GameResult, GameState
from src.wikipedia.scraper import WikiScraper
if TYPE_CHECKING:
from playwright.sync_api import Page
from src.agents.base import Agent
logger = logging.getLogger(__name__)
class GameEngine:
"""
Runs Wikipedia speedrun games with optional browser visualization.
The engine handles:
- Fetching pages and extracting links
- Running the agent's decision loop
- Optional Playwright visualization with link highlighting
- Recording game state and results
"""
# CSS for highlighting links
HIGHLIGHT_CSS = """
.wiki-speedrun-highlight {
outline: 3px solid red !important;
background-color: rgba(255, 0, 0, 0.1) !important;
}
.wiki-speedrun-chosen {
outline: 3px solid green !important;
background-color: rgba(0, 255, 0, 0.2) !important;
}
.wiki-speedrun-target {
outline: 3px solid gold !important;
background-color: rgba(255, 215, 0, 0.2) !important;
}
"""
def __init__(
self,
visualize: bool = False,
headless: bool = PLAYWRIGHT_HEADLESS,
slow_mo: int = 0,
) -> None:
"""
Initialize the game engine.
Args:
visualize: Whether to show browser visualization
headless: Run browser in headless mode (only if visualize=True)
slow_mo: Slow down Playwright operations by this many ms
"""
self._visualize = visualize
self._headless = headless
self._slow_mo = slow_mo
self._scraper = WikiScraper()
# Playwright resources (lazy-initialized)
self._browser = None
self._context = None
self._page: Page | None = None
def _init_browser(self) -> None:
"""Initialize Playwright browser for visualization."""
if self._page is not None:
return
from playwright.sync_api import sync_playwright
logger.info("Starting Playwright browser...")
self._playwright = sync_playwright().start()
self._browser = self._playwright.chromium.launch(
headless=self._headless,
slow_mo=self._slow_mo,
)
self._context = self._browser.new_context(
viewport={"width": 1280, "height": 900},
)
self._page = self._context.new_page()
self._page.set_default_timeout(PLAYWRIGHT_TIMEOUT)
# Inject our highlight CSS
self._page.add_style_tag(content=self.HIGHLIGHT_CSS)
def _close_browser(self) -> None:
"""Clean up Playwright resources."""
if self._page:
self._page.close()
self._page = None
if self._context:
self._context.close()
self._context = None
if self._browser:
self._browser.close()
self._browser = None
if hasattr(self, "_playwright") and self._playwright:
self._playwright.stop()
self._playwright = None
def _navigate_to(self, title: str) -> list[str]:
"""
Navigate to a Wikipedia page and extract links.
Uses Playwright if visualizing, otherwise just scrapes.
Returns:
List of article titles linked from this page
"""
# Always use BeautifulSoup for reliable link extraction
links = self._scraper.get_links(title)
# If visualizing, also navigate the browser
if self._visualize and self._page:
url = self._scraper.title_to_url(title)
self._page.goto(url, wait_until="load")
# Re-inject CSS (lost on navigation)
try:
self._page.add_style_tag(content=self.HIGHLIGHT_CSS)
except Exception:
pass # Page might not be ready
return links
def _extract_links_playwright(self) -> list[str]:
"""Extract article links from current page using Playwright."""
# JavaScript to extract links from main content
js_code = """
() => {
const content = document.querySelector('.mw-parser-output');
if (!content) {
console.log('No .mw-parser-output found');
return [];
}
const links = [];
const seen = new Set();
// Get links from paragraphs, lists, tables
const elements = content.querySelectorAll('p a, li a, td a, th a, dd a');
for (const link of elements) {
const href = link.getAttribute('href');
if (!href || !href.startsWith('/wiki/')) continue;
// Skip special namespaces
const excluded = ['Wikipedia:', 'Help:', 'Template:', 'Category:',
'Portal:', 'File:', 'Special:', 'Talk:', 'User:'];
// Strip anchors and query params, then decode
const path = href.replace('/wiki/', '').split('#')[0].split('?')[0];
const title = decodeURIComponent(path).replace(/_/g, ' ');
if (excluded.some(prefix => title.startsWith(prefix))) continue;
// Skip if in navbox, infobox, etc.
let skip = false;
let parent = link.parentElement;
while (parent && parent !== content) {
const classes = parent.className || '';
if (/navbox|infobox|sidebar|reflist|references|toc/.test(classes)) {
skip = true;
break;
}
parent = parent.parentElement;
}
if (skip) continue;
if (!seen.has(title)) {
seen.add(title);
links.push(title);
}
}
return links;
}
"""
return self._page.evaluate(js_code)
def _highlight_links(self, links: list[str], target: str) -> None:
"""Highlight available links on the page."""
if not self._visualize or not self._page:
return
# Convert link titles to href format
href_parts = [link.replace(" ", "_") for link in links]
target_href = target.replace(" ", "_") if target in links else None
# Single JS call to highlight all links
self._page.evaluate("""
({hrefParts, targetHref}) => {
// Clear previous highlights
document.querySelectorAll('.wiki-speedrun-highlight, .wiki-speedrun-target')
.forEach(el => el.classList.remove('wiki-speedrun-highlight', 'wiki-speedrun-target'));
// Build a Set for fast lookup
const hrefSet = new Set(hrefParts);
// Single pass through all links
document.querySelectorAll('a').forEach(link => {
if (!link.href || !link.href.includes('/wiki/')) return;
// Extract the wiki page name from href
const match = link.href.match(/\\/wiki\\/([^#?]+)/);
if (!match) return;
const pageName = decodeURIComponent(match[1]);
if (pageName === targetHref) {
link.classList.add('wiki-speedrun-target');
} else if (hrefSet.has(pageName)) {
link.classList.add('wiki-speedrun-highlight');
}
});
}
""", {"hrefParts": href_parts, "targetHref": target_href})
def _show_chosen_link(self, chosen: str) -> None:
"""Briefly highlight the chosen link before clicking."""
if not self._visualize or not self._page:
return
href_chosen = chosen.replace(" ", "_")
self._page.evaluate("""
(hrefPart) => {
document.querySelectorAll('a').forEach(link => {
if (!link.href || !link.href.includes('/wiki/')) return;
const match = link.href.match(/\\/wiki\\/([^#?]+)/);
if (!match) return;
const pageName = decodeURIComponent(match[1]);
if (pageName === hrefPart) {
link.classList.remove('wiki-speedrun-highlight', 'wiki-speedrun-target');
link.classList.add('wiki-speedrun-chosen');
}
});
}
""", href_chosen)
# Scroll the chosen link into view
self._page.evaluate("""
(hrefPart) => {
for (const link of document.querySelectorAll('a')) {
if (!link.href || !link.href.includes('/wiki/')) continue;
const match = link.href.match(/\\/wiki\\/([^#?]+)/);
if (!match) continue;
const pageName = decodeURIComponent(match[1]);
if (pageName === hrefPart) {
link.scrollIntoView({ behavior: 'smooth', block: 'center' });
break;
}
}
}
""", href_chosen)
# Brief pause to show the selection
time.sleep(0.5)
def run(
self,
agent: Agent,
start: str,
target: str,
max_steps: int = MAX_STEPS,
) -> GameResult:
"""
Run a complete Wikipedia speedrun game.
Args:
agent: The agent to play the game
start: Starting article title
target: Target article title
max_steps: Maximum clicks before game is lost
Returns:
GameResult with full game record
"""
logger.info(f"Starting game: '{start}' -> '{target}' with {agent.name}")
# Initialize visualization if needed
if self._visualize or agent.requires_visualization:
self._visualize = True
self._headless = False # Force visible for human agents
self._init_browser()
# Initialize game state
state = GameState(
start_title=start,
target_title=target,
current_title=start,
)
state.start_time_ms = time.time() * 1000
# Notify agent
agent.on_game_start(start, target)
try:
# Main game loop
while not state.is_won and state.click_count < max_steps:
# Get available links from current page
available_links = self._navigate_to(state.current_title)
if not available_links:
logger.warning(f"No links found on '{state.current_title}'")
break
# Highlight links if visualizing
self._highlight_links(available_links, target)
# Check if target is directly reachable
if target in available_links:
logger.info(f"Target '{target}' is reachable!")
# Build context for agent
context = AgentContext(
current_title=state.current_title,
target_title=target,
available_links=available_links,
path_so_far=list(state.path),
step_count=state.click_count,
)
# Let agent choose
decision_start = time.time() * 1000
chosen = agent.choose_link(context)
decision_time = time.time() * 1000 - decision_start
# Validate choice
if chosen not in available_links:
logger.error(f"Agent chose invalid link: '{chosen}'")
raise ValueError(f"Invalid link choice: '{chosen}'")
logger.info(
f"Step {state.click_count + 1}: '{state.current_title}' -> '{chosen}'"
)
# Show the chosen link before navigating
self._show_chosen_link(chosen)
# Record step and update state
state.record_step(chosen, available_links, decision_time)
# Game finished
total_time = time.time() * 1000 - state.start_time_ms
result = state.to_result(agent.name, total_time)
# Notify agent
agent.on_game_end(result.won, result.path)
if result.won:
logger.info(
f"Won! Path ({result.total_clicks} clicks): "
f"{' -> '.join(result.path)}"
)
else:
logger.info(f"Lost after {result.total_clicks} clicks")
return result
finally:
# Clean up browser if we own it
if self._visualize and not agent.requires_visualization:
self._close_browser()
def close(self) -> None:
"""Clean up resources."""
self._close_browser()
def __enter__(self) -> "GameEngine":
return self
def __exit__(self, *args) -> None:
self.close()
|