Simulating Multi-Agent Worlds: Managing State with Google ADK
Introduction
Welcome to the LLM Agent Game tutorial! In this guide, we'll demonstrate how to create a grid-based survival game to Google's Agent Development Kit (ADK).
In this world, explorers like Alice, Bob, and Charlie compete for wealth ($) while managing their stamina. Each explorer has a unique personality: Aggressive, Peaceful, or Survivalist. We use ADK's Runner to execute their turns in a turn-based simulation loop.
Project Architecture
We've structured the project to separate world physics from agent decision-making. Here is the layout:
/agent_game/
├── world.py # Physics Engine: Grid, wealth, and movement rules
├── tools.py # Interface: Game sensing and action tools
├── prompts.py # Personas: Alice, Bob, and Charlie's principles
├── agent.py # ADK LlmAgent definitions
├── standalone_agent.py # All-in-one script for easy testing
├── run_game.sh # Bash script to launch the simulation
└── .env # API Keys for Gemini
Step 2: The Simulation Engine
The ExplorerWorld class handles the grid, resource distribution, and the results of actions like move, gather, and attack.
class ExplorerWorld:
def move(self, name, direction):
# Logic to update (x, y) coordinates
...
def gather_wealth(self, name):
# Logic to collect resource from the current cell
...
def attack(self, attacker_name, defender_name):
# Stamina-based combat resolution
...
Step 3: Game Interaction Tools
Agents use specialized tools to perceive their environment and execute actions. We provide:
get_game_state: Allows agents to "see" nearby wealth and other players.perform_action: Translates agent decisions into world state changes.
Step 4: The Standalone Implementation
For quick iteration and teaching, we've combined everything into a single file. This script initializes the world, creates the agents, and manages the turn-based loop using ADK's Runner.
import os
import asyncio
import random
import copy
from dotenv import load_dotenv
from google.adk.agents.llm_agent import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.adk.utils.context_utils import Aclosing
from google.genai import types
# Load Environment Variables
load_dotenv()
# --- WORLD LOGIC ---
class WorldError(Exception):
pass
class ExplorerWorld:
def __init__(self, map_size):
self.map_size = map_size
self.scope_size = 2
self.map = [[0 for _ in range(map_size)] for _ in range(map_size)]
self.explorers = {}
self.max_wealth = 10
self.max_stamina = 10
def random_initialize_map(self, wealth_density=0.1):
for i in range(self.map_size):
for j in range(self.map_size):
if random.random() < wealth_density:
self.map[i][j] = 1
def add_explorer(self, name, x=None, y=None, stamina=None):
if x is None: x = random.randint(0, self.map_size - 1)
if y is None: y = random.randint(0, self.map_size - 1)
self.explorers[name] = {"x": x, "y": y, "wealth": 0, "stamina": stamina or self.max_stamina}
def move(self, name, direction):
e = self.explorers[name]
if direction == "up": e["y"] = min(self.map_size-1, e["y"]+1)
elif direction == "down": e["y"] = max(0, e["y"]-1)
elif direction == "left": e["x"] = max(0, e["x"]-1)
elif direction == "right": e["x"] = min(self.map_size-1, e["x"]+1)
e["stamina"] -= 1
def gather_wealth(self, name):
e = self.explorers[name]
if self.map[e["y"]][e["x"]] > 0:
e["wealth"] += self.map[e["y"]][e["x"]]
self.map[e["y"]][e["x"]] = 0
e["stamina"] -= 1
else: raise WorldError("No wealth here.")
def rest(self, name):
e = self.explorers[name]
e["stamina"] = min(e["stamina"] + 3, self.max_stamina)
def attack(self, attacker_name, defender_name):
a, d = self.explorers[attacker_name], self.explorers.get(defender_name)
if not d: return
if a["stamina"] > d["stamina"]:
a["wealth"] += d["wealth"]
del self.explorers[defender_name]
else:
d["wealth"] += a["wealth"]
del self.explorers[attacker_name]
def get_surroundings(self, name):
e = self.explorers.get(name)
if not e: return []
x, y = e["x"], e["y"]
min_x, max_x = max(0, x-2), min(self.map_size, x+3)
min_y, max_y = max(0, y-2), min(self.map_size, y+3)
view = [[self.map[_y][_x] for _x in range(min_x, max_x)] for _y in range(min_y, max_y)]
for other_name, other_e in self.explorers.items():
ox, oy = other_e["x"], other_e["y"]
if min_x <= ox < max_x and min_y <= oy < max_y:
label = 'Yourself' if other_name == name else other_name
view[oy - min_y][ox - min_x] = (label, view[oy - min_y][ox - min_x])
return view[::-1]
def get_allowed_actions(self, name):
e = self.explorers.get(name)
if not e: return []
allowed = ['rest', 'move up', 'move down', 'move left', 'move right']
if self.map[e['y']][e['x']] > 0: allowed.append("gather")
for other_name, other_e in self.explorers.items():
if other_name != name and abs(e['x']-other_e['x']) + abs(e['y']-other_e['y']) == 1:
allowed.append("attack")
return allowed
def __repr__(self):
grid = [['.' for _ in range(self.map_size)] for _ in range(self.map_size)]
for y in range(self.map_size):
for x in range(self.map_size):
if self.map[y][x] > 0: grid[y][x] = '$'
for name, exp in self.explorers.items():
grid[exp['y']][exp['x']] = name[0].upper()
return "\n".join([" ".join(row) for row in reversed(grid)])
# --- GLOBAL WORLD ---
world = ExplorerWorld(7)
world.random_initialize_map(wealth_density=0.6)
world.add_explorer("Alice")
world.add_explorer("Bob")
world.add_explorer("Charlie")
# --- TOOLS ---
async def get_game_state(agent_name: str) -> dict:
surroundings = world.get_surroundings(agent_name)
e = world.explorers.get(agent_name)
allowed = world.get_allowed_actions(agent_name)
return {
"status": f"Stamina: {e['stamina']}, Wealth: {e['wealth']}",
"allowed_actions": allowed,
"surroundings": str(surroundings)
}
async def perform_action(agent_name: str, action: str) -> str:
action = action.lower()
try:
if 'move' in action: world.move(agent_name, action.split(" ")[1])
elif 'gather' in action: world.gather_wealth(agent_name)
elif 'rest' in action: world.rest(agent_name)
elif 'attack' in action:
for name, other in world.explorers.items():
if name != agent_name and abs(world.explorers[agent_name]['x']-other['x']) + abs(world.explorers[agent_name]['y']-other['y']) == 1:
world.attack(agent_name, name)
return f"Attacked {name}!"
return f"Performed {action}"
except Exception as e: return f"Error: {e}"
# --- AGENTS ---
PROMPT = "You are {name}, an explorer. Principles: {principles}. Use get_game_state and perform_action. Turn-based game."
alice = LlmAgent(name="Alice", model="gemini-2.0-flash", tools=[get_game_state, perform_action],
instruction=PROMPT.format(name="Alice", principles="Aggressive, attack others."))
bob = LlmAgent(name="Bob", model="gemini-2.0-flash", tools=[get_game_state, perform_action],
instruction=PROMPT.format(name="Bob", principles="Peaceful, gather wealth."))
charlie = LlmAgent(name="Charlie", model="gemini-2.0-flash", tools=[get_game_state, perform_action],
instruction=PROMPT.format(name="Charlie", principles="Avoid others, just survive."))
async def main():
service = InMemorySessionService()
runners = {
"Alice": Runner(app_name="agent_game", agent=alice, session_service=service, auto_create_session=True),
"Bob": Runner(app_name="agent_game", agent=bob, session_service=service, auto_create_session=True),
"Charlie": Runner(app_name="agent_game", agent=charlie, session_service=service, auto_create_session=True),
}
for turn in range(5):
print(f"\n--- TURN {turn} ---")
print(world)
for name, runner in runners.items():
if name in world.explorers:
print(f"\n> {name}'s turn:")
msg = types.Content(parts=[types.Part(text=f"It is your turn. Current state: {world.explorers[name]}")])
async with Aclosing(runner.run_async(user_id="user", session_id=f"session_{name}", new_message=msg)) as agen:
async for event in agen:
if event.content:
print(f"Agent says: {event.content.parts[0].text}")
if __name__ == "__main__":
asyncio.run(main())
Final Result
Here is the visual outcome of a typical simulation turn in the explorer world:
