Skip to Content

Simulating Multi-Agent Worlds: Managing State with Google ADK

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.

Step 1: The Blueprint

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
        
world.py

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
        ...
            
tools.py

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.
standalone_agent.py

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())
            
Demo

Final Result

Here is the visual outcome of a typical simulation turn in the explorer world:

in AI
Mastering Parallel Agents in Google ADK: The Travel Planner Case Study