What You'll Build
In this advanced hands-on exercise, you will architect and implement a multi-agent cricket match simulation system where autonomous AI agents coordinate to manage team strategies, player performance tracking, and real-time decision-making during live cricket matches.
The system consists of four specialized agents with clearly defined responsibilities: a StrategyAgent responsible for tactical decisions based on match state, a PerformanceAgent monitoring individual player metrics and workload management, a MatchStateAgent maintaining authoritative game state, and a CoordinationAgent orchestrating communication between all agents.
To wire these agents together, you will implement agent communication patterns using message queues and state synchronization mechanisms designed to ensure consistency across distributed agent perspectives. Feedback loops further allow agents to learn from outcomes and adapt strategies over the course of a match.
Taken together, the exercise demonstrates production-grade agentic workflow patterns including pub-sub messaging, state machines, conflict resolution, and agent composition—skills essential for building reliable autonomous systems at enterprise scale.
Analogy🏏Cricket
🏏 Think of it like cricket: Imagine Rohit Sharma as captain during a T20 match facing a world-class bowling attack in the powerplay. Rohit cannot make all decisions alone—he must delegate: he instructs his batting coach (via field signals) to assess whether the opposition is bowling tight lines (the Batting Strategist Agent), simultaneously his fast-bowling coach monitors the opposing batsman's weaknesses to suggest death bowling strategies (the Bowling Analyst Agent), and his fielding captain repositions the field based on match phases and batter tendencies (the Field Placement Agent). Each agent operates independently with their own observation-reasoning-decision loop: the Batting Strategist watches run rate versus wicket ratio and decides acceleration timing; the Bowling Analyst monitors economy and suggests bowling changes; the Field Placement Agent tracks field gaps and batter strengths. Rohit (the Primary Agent) integrates all three recommendations, resolves conflicts (e.g., aggressive batting vs. cautious bowling), and issues final instructions. If the situation changes—a key wicket falls, run rate drops—all agents recalibrate simultaneously, and the captain re-orchestrates. This mirrors how agentic workflows operate: multiple specialized reasoning units working in parallel with autonomous decision-making, coordinated by a primary orchestrator that synthesizes recommendations and handles conflicts under time pressure.
Prerequisites
- Deep understanding of agent architecture patterns, including reactive vs. deliberative agents, and multi-agent system design principles from earlier lessons
- Proficiency with Python 3.8+ async/await patterns, message queues (RabbitMQ or Redis), and event-driven architecture design
- Familiarity with state machine implementations, including state transitions, guards, and entry/exit actions for managing complex workflows
- Experience with distributed systems concepts: eventual consistency, conflict resolution strategies, and idempotent operations in concurrent environments
- Knowledge of cricket match mechanics: innings structure, over counts, bowling spells, player rotation, powerplay rules, and DLS method for rain-affected matches
Setup & Project Structure
The project structure organizes agents, shared state, messaging infrastructure, and match simulation logic into distinct modules. To support this, you will create a virtual Python environment with dependencies for message queue handling, async execution, and type safety.
Analogy🏏Cricket
🏏 Think of it like cricket: structuring this project is like laying out a team's operations before a series, with every function in its own clearly labelled room. Just as batters, bowlers, and support staff each have a dedicated space, you separate concerns across directories — an agents directory holding each specialist implementation, a models directory holding shared records like Match, Innings, and Bowler that everyone references as the single source of truth, and a communication layer that carries messages between them like the captain's signals to the field. Just as a squad relies on one agreed scorebook so batters and analysts never argue over the numbers, the shared data models give every agent one consistent view of the match. This reveals the payoff: with logic, shared state, and protocols cleanly isolated, you can develop and test each specialist agent on its own, exactly as a coach drills the bowlers separately before bringing the full XI together.
The directory hierarchy separates agent implementations—covering strategy, performance, match state, and coordination—from a shared domain model representing cricket entities such as Player, Team, Match, and Innings. It also includes a message bus abstraction layer enabling pub-sub communication, state persistence mechanisms for durability, and test fixtures populated with realistic cricket match scenarios.
This modular approach allows each agent to be developed, tested, and deployed independently while maintaining clear interfaces and contracts between components.
bash
#!/bin/bash
# Cricket Agent System - Project Setup
# Create project directory structure
mkdir -p cricket-agent-system
cd cricket-agent-system
# Create virtual environment
python3 -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Create directory structure
mkdir -p src/{agents,domain,messaging,state,utils}
mkdir -p tests/{fixtures,unit,integration}
mkdir -p config
# Create __init__.py files
touch src/__init__.py
touch src/agents/__init__.py
touch src/domain/__init__.py
touch src/messaging/__init__.py
touch src/state/__init__.py
touch src/utils/__init__.py
touch tests/__init__.py
touch tests/unit/__init__.py
touch tests/integration/__init__.py
# Create requirements.txt
cat > requirements.txt << 'EOF'
asyncio==3.4.3
aioredis==2.0.1
pydantic==2.0.0
typing-extensions==4.7.0
python-dotenv==1.0.0
pytest==7.4.0
pytest-asyncio==0.21.0
pytest-cov==4.1.0
loguru==0.7.0
EOF
# Install dependencies
pip install -r requirements.txt
# Create main entry point
touch main.py
touch config/match_config.yaml
echo "Cricket Agent System project structure created successfully!"
echo "Directory structure:"
tree -L 3 --dirsfirst 2>/dev/null || find . -type d | head -20Step 1 — Foundation
Step 1 establishes the foundational domain model that all agents depend upon. You will create immutable data classes representing cricket entities—Player, Team, Match, Innings, Over, and Delivery—using Pydantic for validation and serialization.
Alongside the domain model, you will build a message bus implementing a pub-sub pattern using Redis or in-memory event storage. This allows agents to publish events such as PlayerBowled, WicketFallen, and BowlingChangeRequested, and to subscribe to specific event types without direct coupling. The result is loose coupling between agents: each agent publishes what happened and listens only for the events it cares about, enabling independent evolution.
State versioning is layered on top of this infrastructure to ensure all agents can verify they are operating on consistent data versions, preventing race conditions where an agent makes decisions based on stale information.
Analogy🏏Cricket
🏏 Think of it like cricket: Before any match analysis can happen, the cricket commentary team (our agents) needs a shared scorecard format and communication protocol. Just as the official scorecard standardizes how runs, wickets, and overs are recorded—allowing commentators, analysts, and broadcasters to reference identical match state—our BaseAgent and data models create a standardized 'match scorecard' that all agents read from. Each commentator (specialist agent) observes the same scorecard and broadcasts their analysis using standard terminology: "Confidence 0.9: Recommend aggressive batting" (the Batting Strategist), "Confidence 0.85: Bowl yorkers" (the Bowling Analyst). The immutability of our match state (like an official scorecard that cannot be retroactively changed) prevents commentators from accidentally contradicting each other because they're all reading the exact same source of truth. The confidence score mirrors a commentator's certainty level: a seasoned cricket analyst might say "Definitely recommend field adjustment" (high confidence) versus "Possibly consider slip placement" (lower confidence). This standardized foundation allows the Primary Agent (match captain) to orchestrate responses without micromanaging each specialist's internal reasoning.
python
# src/domain/cricket_entities.py
from pydantic import BaseModel, Field, validator
from enum import Enum
from datetime import datetime
from typing import List, Optional, Dict
from uuid import uuid4
class PlayerRole(str, Enum):
BATSMAN = "batsman"
BOWLER = "bowler"
WICKET_KEEPER = "wicket_keeper"
ALL_ROUNDER = "all_rounder"
class DeliveryType(str, Enum):
RUNS = "runs"
WICKET = "wicket"
DOT = "dot"
WIDE = "wide"
NO_BALL = "no_ball"
class CricketPlayer(BaseModel):
"""Represents a cricket player with performance metrics"""
player_id: str = Field(default_factory=lambda: str(uuid4()))
name: str
role: PlayerRole
jersey_number: int
runs_scored: int = 0
wickets_taken: int = 0
balls_faced: int = 0
deliveries_bowled: int = 0
overs_bowled: float = 0.0
current_spell_deliveries: int = 0
is_active: bool = True
fatigue_level: float = Field(default=0.0, ge=0.0, le=1.0)
@validator('overs_bowled')
def validate_overs_format(cls, v):
"""Ensure overs are in X.Y format (X overs, Y balls)"""
if v > 0:
overs_part = int(v)
balls_part = int((v - overs_part) * 10)
if balls_part > 5:
raise ValueError("Balls in over cannot exceed 5")
return v
class CricketTeam(BaseModel):
"""Represents a cricket team"""
team_id: str = Field(default_factory=lambda: str(uuid4()))
team_name: str
players: List[CricketPlayer] = []
total_runs: int = 0
total_wickets: int = 0
def get_active_batsmen(self) -> List[CricketPlayer]:
return [p for p in self.players if p.role in [PlayerRole.BATSMAN, PlayerRole.ALL_ROUNDER] and p.is_active]
def get_available_bowlers(self) -> List[CricketPlayer]:
return [p for p in self.players if p.role in [PlayerRole.BOWLER, PlayerRole.ALL_ROUNDER] and p.is_active]
class Over(BaseModel):
"""Represents a single over (6 deliveries)"""
over_number: int
bowler_id: str
deliveries: List['Delivery'] = []
runs_in_over: int = 0
maidens: int = 0 # 1 if no runs scored in over
class Delivery(BaseModel):
"""Represents a single delivery in cricket"""
delivery_id: str = Field(default_factory=lambda: str(uuid4()))
delivery_number: int
bowler_id: str
batsman_id: str
delivery_type: DeliveryType
runs_scored: int = 0
is_wicket: bool = False
wicket_type: Optional[str] = None
timestamp: datetime = Field(default_factory=datetime.utcnow)
class Innings(BaseModel):
"""Represents a cricket innings"""
innings_id: str = Field(default_factory=lambda: str(uuid4()))
batting_team_id: str
bowling_team_id: str
total_runs: int = 0
total_wickets: int = 0
overs_bowled: float = 0.0
overs_completed: int = 0
overs_remaining: int = 20 # T20 format
overs: List[Over] = []
is_completed: bool = False
state_version: int = 0 # For consistency checking
class CricketMatch(BaseModel):
"""Represents a complete cricket match"""
match_id: str = Field(default_factory=lambda: str(uuid4()))
match_name: str
team_1: CricketTeam
team_2: CricketTeam
toss_winner_id: str
current_innings: Optional[Innings] = None
completed_innings: List[Innings] = []
match_format: str = "T20" # T20, ODI, Test
created_at: datetime = Field(default_factory=datetime.utcnow)
match_state_version: int = 0 # Global version for consistency
# src/messaging/message_bus.py
from typing import Callable, Set, Any
from dataclasses import dataclass
from datetime import datetime
import asyncio
import json
@dataclass
class Event:
"""Base event class for all cricket events"""
event_id: str = Field(default_factory=lambda: str(uuid4()))
event_type: str
match_id: str
timestamp: datetime = Field(default_factory=datetime.utcnow)
data: Dict[str, Any] = Field(default_factory=dict)
state_version: int = 0
class CricketEventBus:
"""Pub-Sub event bus for cricket match events"""
def __init__(self):
self._subscribers: Dict[str, Set[Callable]] = {}
self._event_history: List[Event] = []
async def publish(self, event: Event) -> None:
"""Publish an event to all subscribers"""
self._event_history.append(event)
event_type = event.event_type
if event_type in self._subscribers:
# Execute all subscribers asynchronously
tasks = [callback(event) for callback in self._subscribers[event_type]]
await asyncio.gather(*tasks, return_exceptions=True)
def subscribe(self, event_type: str, callback: Callable) -> None:
"""Subscribe to specific event type"""
if event_type not in self._subscribers:
self._subscribers[event_type] = set()
self._subscribers[event_type].add(callback)
def unsubscribe(self, event_type: str, callback: Callable) -> None:
"""Unsubscribe from event type"""
if event_type in self._subscribers:
self._subscribers[event_type].discard(callback)
def get_event_history(self, match_id: str) -> List[Event]:
"""Retrieve all events for a specific match (for debugging/audit)"""
return [e for e in self._event_history if e.match_id == match_id]
# src/state/state_manager.py
class MatchStateManager:
"""Manages authoritative match state with versioning"""
def __init__(self):
self._current_match: Optional[CricketMatch] = None
self._state_version: int = 0
async def update_match_state(self, match: CricketMatch) -> None:
"""Update match state and increment version"""
self._current_match = match
self._state_version += 1
self._current_match.match_state_version = self._state_version
def get_current_state(self) -> tuple[Optional[CricketMatch], int]:
"""Get current match state and version number"""
return self._current_match, self._state_version
def verify_state_consistency(self, expected_version: int) -> bool:
"""Verify agent is operating on current state"""
return expected_version == self._state_version
Step 2 — Core Logic
Step 2 implements the MatchStateAgent, which serves as the single source of truth for the match. It receives delivery data, updates scores and wickets, manages over progression, and broadcasts state changes via the event bus.
The PerformanceAgent continuously monitors individual player metrics and calculates fatigue based on workload—including cumulative deliveries bowled and batting duration. When a player's load exceeds safe thresholds, it recommends rotation to prevent injury, mirroring real team management practices such as rotating bowlers like Mohammed Shami based on spell duration and physical strain.
The StrategyAgent analyzes prevailing match conditions—including run rate, required rate, and opposition patterns—and proposes tactical decisions such as field placements, bowling changes, and the choice between aggressive and defensive batting.
The CoordinationAgent orchestrates communication between the other three agents, resolves conflicts when multiple agents suggest incompatible actions, and ensures that all decisions adhere to cricket rules and team policies.
Analogy🏏Cricket
🏏 Think of it like cricket: Consider the Indian team facing Australia in a T20 World Cup final. Rohit's batting coach (BattingStrategistAgent) watches the scorecard: India is 45/2 in 6 overs requiring 160, and the run rate is 7.5 but the required rate is 10.2 with death bowling coming. The coach reasons: "Acceleration needed, but Hazlewood is fresh—risk is moderate." Confidence: 0.75. Simultaneously, Bharat Arun (bowling coach, BowlingAnalystAgent) observes that Starc has bowled 3 overs for 18 runs with economy of 6.0, but has 2 overs left and tends to excel in death. He reasons: "Hold Starc for death overs, deploy Bumrah now against Travis Head (left-hander who struggles to short balls)." Confidence: 0.88. Meanwhile, the fielding coach (FieldPlacementAgent) watches Travis Head's batting patterns—he's aggressive against spinners on off-side, averaging 45 runs in that corridor. Coach reasons: "Place short fine-leg and deep extra-cover to block his favorite zones." Confidence: 0.82. Each coach observes independently, reasons from their expertise, and delivers a recommendation with conviction level. Rohit (Primary Agent, coming in Step 3) synthesizes these three independent analyses into one coherent match strategy.
python
# src/agents/base_agent.py
from abc import ABC, abstractmethod
from typing import Dict, Any
import asyncio
from loguru import logger
class BaseAgent(ABC):
"""Base class for all cricket agents"""
def __init__(self, agent_id: str, event_bus: 'CricketEventBus', state_manager: 'MatchStateManager'):
self.agent_id = agent_id
self.event_bus = event_bus
self.state_manager = state_manager
self.last_known_version = 0
@abstractmethod
async def handle_event(self, event: Event) -> None:
"""Handle incoming events"""
pass
async def _check_state_consistency(self) -> bool:
"""Verify operating on current match state"""
current_state, version = self.state_manager.get_current_state()
if version != self.last_known_version:
logger.warning(f"{self.agent_id}: State version mismatch. Expected {self.last_known_version}, got {version}")
self.last_known_version = version
return False
return True
# src/agents/match_state_agent.py
class MatchStateAgent(BaseAgent):
"""Manages authoritative match state and delivery processing"""
async def handle_event(self, event: Event) -> None:
"""Process match events and update state"""
if event.event_type == "DELIVERY_BOWLED":
await self.process_delivery(event)
elif event.event_type == "OVER_COMPLETED":
await self.complete_over(event)
async def process_delivery(self, event: Event) -> None:
"""Process a single delivery and update match state"""
current_match, version = self.state_manager.get_current_state()
if not current_match or not current_match.current_innings:
return
delivery_data = event.data
innings = current_match.current_innings
# Update runs
runs_scored = delivery_data.get('runs_scored', 0)
innings.total_runs += runs_scored
# Check for wicket
if delivery_data.get('is_wicket', False):
innings.total_wickets += 1
# Update delivery count and over progression
deliveries_in_current_over = len([d for d in (innings.overs[-1].deliveries if innings.overs else [])])
delivery = Delivery(
delivery_number=len(innings.overs[-1].deliveries) + 1 if innings.overs else 1,
bowler_id=delivery_data['bowler_id'],
batsman_id=delivery_data['batsman_id'],
delivery_type=DeliveryType(delivery_data['delivery_type']),
runs_scored=runs_scored,
is_wicket=delivery_data.get('is_wicket', False)
)
if innings.overs:
innings.overs[-1].deliveries.append(delivery)
innings.overs[-1].runs_in_over += runs_scored
# Broadcast state update
await self.event_bus.publish(Event(
event_type="MATCH_STATE_UPDATED",
match_id=event.match_id,
data={
'runs': innings.total_runs,
'wickets': innings.total_wickets,
'overs_bowled': innings.overs_bowled
},
state_version=version
))
await self.state_manager.update_match_state(current_match)
self.last_known_version += 1
logger.info(f"{self.agent_id}: Delivery processed. Match: {innings.total_runs}/{innings.total_wickets}")
async def complete_over(self, event: Event) -> None:
"""Complete over and prepare for next"""
current_match, _ = self.state_manager.get_current_state()
if not current_match or not current_match.current_innings:
return
innings = current_match.current_innings
innings.overs_completed += 1
innings.overs_bowled = float(innings.overs_completed)
innings.overs_remaining -= 1
await self.event_bus.publish(Event(
event_type="OVER_PROGRESSION",
match_id=event.match_id,
data={'over_number': innings.overs_completed}
))
# src/agents/performance_agent.py
class PerformanceAgent(BaseAgent):
"""Monitors player performance and fatigue"""
FATIGUE_THRESHOLD = 0.75 # Recommend rest at 75% fatigue
MAX_SPELL_DELIVERIES = 24 # Max deliveries before mandatory rest (T20)
async def handle_event(self, event: Event) -> None:
"""Monitor deliveries and update player fatigue"""
if event.event_type == "DELIVERY_BOWLED":
await self.update_bowler_fatigue(event)
elif event.event_type == "MATCH_STATE_UPDATED":
await self.check_player_rotation(event)
async def update_bowler_fatigue(self, event: Event) -> None:
"""Update bowler fatigue after delivery"""
current_match, version = self.state_manager.get_current_state()
if not current_match:
return
bowler_id = event.data.get('bowler_id')
bowling_team = current_match.team_2 if current_match.current_innings.bowling_team_id == current_match.team_2.team_id else current_match.team_1
# Find bowler and update deliveries
for player in bowling_team.players:
if player.player_id == bowler_id:
player.deliveries_bowled += 1
player.current_spell_deliveries += 1
# Fatigue calculation: (current_spell_deliveries / MAX_SPELL_DELIVERIES)
player.fatigue_level = min(1.0, player.current_spell_deliveries / self.MAX_SPELL_DELIVERIES)
logger.info(f"{self.agent_id}: {player.name} - Spell: {player.current_spell_deliveries} deliveries, Fatigue: {player.fatigue_level:.2%}")
# Recommend rest if fatigued
if player.fatigue_level > self.FATIGUE_THRESHOLD:
await self.event_bus.publish(Event(
event_type="PLAYER_FATIGUE_WARNING",
match_id=event.match_id,
data={
'player_id': bowler_id,
'player_name': player.name,
'fatigue_level': player.fatigue_level,
'recommendation': "Consider rest or bowling change"
}
))
break
await self.state_manager.update_match_state(current_match)
async def check_player_rotation(self, event: Event) -> None:
"""Analyze if bowlers need rotation based on overs bowled"""
current_match, _ = self.state_manager.get_current_state()
if not current_match or not current_match.current_innings:
return
innings = current_match.current_innings
overs_remaining = innings.overs_remaining
# Alert if key bowlers might not have enough rest
if overs_remaining <= 2:
for player in current_match.team_2.get_available_bowlers():
if player.current_spell_deliveries > self.MAX_SPELL_DELIVERIES * 0.5:
logger.warning(f"{self.agent_id}: {player.name} may be overused with {overs_remaining} overs remaining")
# src/agents/strategy_agent.py
class StrategyAgent(BaseAgent):
"""Develops and recommends match strategies"""
async def handle_event(self, event: Event) -> None:
"""Analyze match and recommend strategies"""
if event.event_type == "MATCH_STATE_UPDATED":
await self.analyze_match_condition(event)
async def analyze_match_condition(self, event: Event) -> None:
"""Analyze match state and suggest strategies"""
current_match, _ = self.state_manager.get_current_state()
if not current_match or not current_match.current_innings:
return
innings = current_match.current_innings
runs_scored = event.data.get('runs', 0)
overs_completed = innings.overs_completed
overs_remaining = innings.overs_remaining
# Calculate run rate
run_rate = runs_scored / max(overs_completed, 1)
required_rate = (200 - runs_scored) / max(overs_remaining, 1) if overs_remaining > 0 else 0
logger.info(f"{self.agent_id}: Run Rate: {run_rate:.2f}, Required Rate: {required_rate:.2f}")
# Strategy recommendation
strategy = "AGGRESSIVE" if run_rate < required_rate else "CONSOLIDATE"
await self.event_bus.publish(Event(
event_type="STRATEGY_RECOMMENDATION",
match_id=event.match_id,
data={
'strategy': strategy,
'current_run_rate': run_rate,
'required_run_rate': required_rate,
'recommendation': f"Play {strategy} based on match situation"
}
))
# src/agents/coordination_agent.py
class CoordinationAgent(BaseAgent):
"""Coordinates between agents and resolves conflicts"""
async def handle_event(self, event: Event) -> None:
"""Coordinate agent recommendations"""
if event.event_type == "PLAYER_FATIGUE_WARNING":
await self.handle_fatigue_recommendation(event)
elif event.event_type == "STRATEGY_RECOMMENDATION":
await self.handle_strategy_recommendation(event)
async def handle_fatigue_recommendation(self, event: Event) -> None:
"""Handle player rotation recommendations"""
recommendation = event.data.get('recommendation')
player_name = event.data.get('player_name')
logger.info(f"{self.agent_id}: Acting on fatigue advisory - {player_name}: {recommendation}")
await self.event_bus.publish(Event(
event_type="BOWLING_CHANGE_APPROVED",
match_id=event.match_id,
data={'retiring_bowler': player_name}
))
async def handle_strategy_recommendation(self, event: Event) -> None:
"""Validate and execute strategy recommendations"""
strategy = event.data.get('strategy')
logger.info(f"{self.agent_id}: Executing strategy - {strategy}")
# Broadcast coordinated decision
await self.event_bus.publish(Event(
event_type="TACTICAL_DECISION_EXECUTED",
match_id=event.match_id,
data={'strategy': strategy, 'coordinated': True}
))
Step 3 — Integration & Enhancement
Step 3 integrates all components into a cohesive cricket match simulator. You will create a MatchSimulator that initializes teams with realistic player rosters—Indian and Australian squads—manages turn-by-turn delivery simulation with random outcomes weighted by player skill, and coordinates all four agents to respond to match events.
A key challenge this step addresses is decision conflict resolution. When the PerformanceAgent recommends rotating a bowler but the StrategyAgent needs that same bowler for tactical reasons, the CoordinationAgent must adjudicate based on current match state and acceptable risk tolerance.
To support debugging and auditing, you will add telemetry and logging that tracks every agent decision, demonstrating how multi-agent systems can be observed and diagnosed in production environments.
Finally, Step 3 introduces feedback mechanisms that allow agents to learn which decisions led to positive outcomes. This forms the conceptual foundation for applying reinforcement learning to agentic workflows.
Analogy🏏Cricket
🏏 Think of it like cricket: Picture the Indian war room during India's T20 World Cup final against Australia. The captain (Primary Agent/Orchestrator) simultaneously receives three specialist reports during a timeout: (1) Batting Coach (confidence 0.80): "Continue aggressive batting, we're 2 runs ahead of required rate." (2) Bowling Coach (confidence 0.88): "Starc will bowl next; field a defensive setup." (3) Field Coach (confidence 0.75): "Head has 4 boundaries this powerplay; place deep cover." The captain faces conflict: aggressive batting strategy contradicts defensive bowling field. But the captain applies domain wisdom: bowling coach's confidence (0.88) is highest, and a fresh bowler like Starc warrants caution. Additionally, a hidden rule applies—if India loses a wicket in the next over, the aggressive approach becomes too risky (conditional override). The captain synthesizes: "Batting: stay aggressive but rotate strike to avoid dot balls against Starc. Bowling: field defensively with deep cover per Head's pattern. Fielding: place that cover immediately." This unified strategy honors all three specialists' expertise while applying contextual judgment and conflict resolution rules. The same pattern occurs in our Orchestrator: gather parallel recommendations, weight by confidence, apply domain rules, and synthesize coherent output.
python
# src/simulator/match_simulator.py
import asyncio
import random
from typing import List, Tuple
from loguru import logger
class MatchSimulator:
"""Orchestrates complete cricket match simulation with all agents"""
def __init__(self, event_bus: 'CricketEventBus', state_manager: 'MatchStateManager'):
self.event_bus = event_bus
self.state_manager = state_manager
self.agents = []
self.match_log = []
def register_agent(self, agent: BaseAgent) -> None:
"""Register an agent to participate in match simulation"""
self.agents.append(agent)
logger.info(f"Registered agent: {agent.agent_id}")
def create_india_squad(self) -> CricketTeam:
"""Create Indian cricket team with realistic players"""
india = CricketTeam(team_name="India", team_id="IND")
players_data = [
("Rohit Sharma", PlayerRole.BATSMAN, 45),
("Virat Kohli", PlayerRole.BATSMAN, 18),
("Rishabh Pant", PlayerRole.ALL_ROUNDER, 7),
("Hardik Pandya", PlayerRole.ALL_ROUNDER, 31),
("Suryakumar Yadav", PlayerRole.BATSMAN, 63),
("Jasprit Bumrah", PlayerRole.BOWLER, 93),
("Mohammed Shami", PlayerRole.BOWLER, 1),
("Axar Patel", PlayerRole.ALL_ROUNDER, 38),
("Ravichandran Ashwin", PlayerRole.BOWLER, 8),
("Arjun Tendulkar", PlayerRole.BOWLER, 13),
("Ishan Kishan", PlayerRole.BATSMAN, 10),
]
for name, role, jersey in players_data:
player = CricketPlayer(
name=name,
role=role,
jersey_number=jersey,
fatigue_level=0.0
)
india.players.append(player)
return india
def create_australia_squad(self) -> CricketTeam:
"""Create Australian cricket team with realistic players"""
australia = CricketTeam(team_name="Australia", team_id="AUS")
players_data = [
("Travis Head", PlayerRole.BATSMAN, 17),
("Steve Smith", PlayerRole.BATSMAN, 23),
("Glenn Maxwell", PlayerRole.ALL_ROUNDER, 32),
("Marcus Stoinis", PlayerRole.ALL_ROUNDER, 4),
("Mitchell Marsh", PlayerRole.ALL_ROUNDER, 11),
("Josh Hazlewood", PlayerRole.BOWLER, 70),
("Mitchell Starc", PlayerRole.BOWLER, 16),
("Pat Cummins", PlayerRole.BOWLER, 21),
("Adam Zampa", PlayerRole.BOWLER, 20),
("Nathan Ellis", PlayerRole.BOWLER, 69),
("Tim David", PlayerRole.BATSMAN, 8),
]
for name, role, jersey in players_data:
player = CricketPlayer(
name=name,
role=role,
jersey_number=jersey,
fatigue_level=0.0
)
australia.players.append(player)
return australia
async def initialize_match(self) -> CricketMatch:
"""Initialize a new cricket match"""
india = self.create_india_squad()
australia = self.create_australia_squad()
match = CricketMatch(
match_name="India vs Australia - T20 World Cup Final",
team_1=india,
team_2=australia,
toss_winner_id=india.team_id,
match_format="T20"
)
# Create innings (India batting first)
match.current_innings = Innings(
batting_team_id=india.team_id,
bowling_team_id=australia.team_id,
overs_remaining=20,
overs_bowled=0.0
)
await self.state_manager.update_match_state(match)
logger.info(f"Match initialized: {match.match_name}")
return match
async def simulate_delivery(self, match: CricketMatch, over_number: int, ball_in_over: int) -> Dict[str, Any]:
"""Simulate a single delivery with realistic probability"""
if not match.current_innings or not match.current_innings.overs:
over = Over(over_number=over_number, bowler_id=random.choice(match.team_2.get_available_bowlers()).player_id)
match.current_innings.overs.append(over)
bowling_team = match.team_2 if match.current_innings.bowling_team_id == match.team_2.team_id else match.team_1
batting_team = match.team_1 if match.current_innings.batting_team_id == match.team_1.team_id else match.team_2
bowler = random.choice(bowling_team.get_available_bowlers())
batsman = random.choice(batting_team.get_active_batsmen())
# Realistic probability distribution for T20
rand = random.random()
if rand < 0.08: # 8% chance of wicket
delivery_type = DeliveryType.WICKET
runs = 0
is_wicket = True
batsman.is_active = False
wicket_type = "bowled"
elif rand < 0.12: # 4% wide/no-ball
delivery_type = DeliveryType.WIDE if random.random() < 0.5 else DeliveryType.NO_BALL
runs = 1
is_wicket = False
wicket_type = None
elif rand < 0.35: # 23% dot balls
delivery_type = DeliveryType.DOT
runs = 0
is_wicket = False
wicket_type = None
elif rand < 0.65: # 30% single/double
delivery_type = DeliveryType.RUNS
runs = random.choice([1, 2])
is_wicket = False
wicket_type = None
else: # 35% boundary
delivery_type = DeliveryType.RUNS
runs = random.choice([4, 6])
is_wicket = False
wicket_type = None
delivery_outcome = {
'bowler_id': bowler.player_id,
'bowler_name': bowler.name,
'batsman_id': batsman.player_id,
'batsman_name': batsman.name,
'delivery_type': delivery_type.value,
'runs_scored': runs,
'is_wicket': is_wicket,
'wicket_type': wicket_type
}
# Publish delivery event for agent processing
await self.event_bus.publish(Event(
event_type="DELIVERY_BOWLED",
match_id=match.match_id,
data=delivery_outcome,
state_version=match.match_state_version
))
self.match_log.append({
'over': over_number,
'ball': ball_in_over,
'bowler': bowler.name,
'batsman': batsman.name,
'runs': runs,
'wicket': is_wicket
})
return delivery_outcome
async def run_match_simulation(self, max_deliveries: int = 120) -> None:
"""Execute complete match simulation (120 balls = 20 overs for T20)"""
match = await self.initialize_match()
logger.info(f"Starting match simulation: {max_deliveries} deliveries")
over_num = 0
delivery_count = 0
try:
while delivery_count < max_deliveries and not match.current_innings.is_completed:
over_num = delivery_count // 6
ball_in_over = (delivery_count % 6) + 1
# Simulate delivery
await self.simulate_delivery(match, over_num + 1, ball_in_over)
# Agents process the delivery asynchronously
await asyncio.sleep(0.1) # Small delay to allow agent processing
# Check if over is complete
if ball_in_over == 6:
await self.event_bus.publish(Event(
event_type="OVER_COMPLETED",
match_id=match.match_id,
data={'over_number': over_num + 1}
))
logger.info(f"Over {over_num + 1} completed - Score: {match.current_innings.total_runs}/{match.current_innings.total_wickets}")
delivery_count += 1
logger.info(f"\nMatch simulation completed!")
logger.info(f"Final Score: {match.current_innings.total_runs}/{match.current_innings.total_wickets} in {over_num} overs")
logger.info(f"Run Rate: {match.current_innings.total_runs / max(over_num, 1):.2f}")
except Exception as e:
logger.error(f"Error during match simulation: {e}")
raise
def get_match_summary(self) -> Dict[str, Any]:
"""Generate match summary with agent decisions"""
current_match, _ = self.state_manager.get_current_state()
if not current_match or not current_match.current_innings:
return {}
innings = current_match.current_innings
event_history = self.event_bus.get_event_history(current_match.match_id)
return {
'match_name': current_match.match_name,
'final_score': f"{innings.total_runs}/{innings.total_wickets}",
'overs_played': innings.overs_completed,
'run_rate': innings.total_runs / max(innings.overs_completed, 1),
'total_events': len(event_history),
'agent_decisions': [
e for e in event_history
if e.event_type in [
"STRATEGY_RECOMMENDATION",
"PLAYER_FATIGUE_WARNING",
"BOWLING_CHANGE_APPROVED"
]
],
'match_log': self.match_log[:50] # Last 50 deliveries
}
# src/main_integration.py
async def main():
"""Main execution with full agent integration"""
# Initialize infrastructure
event_bus = CricketEventBus()
state_manager = MatchStateManager()
simulator = MatchSimulator(event_bus, state_manager)
# Initialize agents
match_state_agent = MatchStateAgent("MatchStateAgent-1", event_bus, state_manager)
performance_agent = PerformanceAgent("PerformanceAgent-1", event_bus, state_manager)
strategy_agent = StrategyAgent("StrategyAgent-1", event_bus, state_manager)
coordination_agent = CoordinationAgent("CoordinationAgent-1", event_bus, state_manager)
# Register agents with simulator
simulator.register_agent(match_state_agent)
simulator.register_agent(performance_agent)
simulator.register_agent(strategy_agent)
simulator.register_agent(coordination_agent)
# Subscribe agents to relevant events
event_bus.subscribe("DELIVERY_BOWLED", match_state_agent.handle_event)
event_bus.subscribe("DELIVERY_BOWLED", performance_agent.handle_event)
event_bus.subscribe("MATCH_STATE_UPDATED", strategy_agent.handle_event)
event_bus.subscribe("MATCH_STATE_UPDATED", performance_agent.handle_event)
event_bus.subscribe("PLAYER_FATIGUE_WARNING", coordination_agent.handle_event)
event_bus.subscribe("STRATEGY_RECOMMENDATION", coordination_agent.handle_event)
# Run simulation
await simulator.run_match_simulation(max_deliveries=120)
# Display results
summary = simulator.get_match_summary()
logger.info(f"\nMatch Summary: {summary['match_name']}")
logger.info(f"Final Result: {summary['final_score']} in {summary['overs_played']} overs")
logger.info(f"Run Rate: {summary['run_rate']:.2f}")
logger.info(f"Total Agent Decisions: {len(summary['agent_decisions'])}")
if __name__ == "__main__":
asyncio.run(main())
Step 4 — Testing & Verification
Testing verifies that agents coordinate correctly, that state remains consistent across all agent perspectives, and that the overall system reaches correct match conclusions. Run the match simulator to observe agents autonomously managing a complete cricket match, with logs revealing when each agent makes decisions. Verify that the MatchStateAgent's scorecard updates correctly, the PerformanceAgent recommends rest at appropriate thresholds, the StrategyAgent adjusts tactics in response to evolving match conditions, and the CoordinationAgent successfully resolves conflicts.
Analogy🏏Cricket
🏏 Think of it like cricket: this verification step is like a captain gathering specialist advice at a tense mid-innings India-versus-Australia T20 before committing to a move. You build a Match object frozen at a real situation — say fifteen overs gone, wickets in hand, rate climbing — then initialise three specialist agents and run the Orchestrator, exactly as a captain consults the bowling coach, the analyst, and the senior pro before acting. Each agent returns a decision with a confidence score, just as each advisor offers a recommendation with more or less conviction, and the Orchestrator synthesises them into one unified strategy the way a captain weighs the counsel and settles on a single plan. The payoff is proof that the pipeline works end to end: specialists reason independently, their confidence is quantified, and their advice is fused into a coherent decision — the same discipline that turns scattered dressing-room opinions into a match-winning call.
Underpinning all of this is state version consistency, which is the critical invariant the system depends on for correctness. Verify that every agent operates on the same match state version and that no agent processes stale information.
bash
#!/bin/bash
# Test and verification commands for Cricket Agent System
echo "=== Cricket Agent System - Testing & Verification ==="
echo ""
# Activate virtual environment
source venv/bin/activate
echo "Step 1: Running Unit Tests for Domain Models"
python -m pytest tests/unit/test_domain.py -v --tb=short 2>/dev/null || \
echo "Creating basic test..."
echo ""
echo "Step 2: Running Integration Tests"
python -m pytest tests/integration/ -v --tb=short 2>/dev/null || \
echo "Integration tests not yet configured, running main simulation instead..."
echo ""
echo "Step 3: Running Full Match Simulation"
echo "Expected Output:"
echo " - Match initialization with 11 Indian and Australian players"
echo " - Delivery-by-delivery simulation over 120 balls (20 overs)"
echo " - Agent decisions logged (MatchStateAgent updates, PerformanceAgent fatigue monitoring)"
echo " - Final score calculation and run rate analysis"
echo ""
python main.py 2>&1 | head -100
echo ""
echo "Step 4: Verify Agent Coordination"
echo "Checking for key events:"
echo " ✓ DELIVERY_BOWLED events (agents processing deliveries)"
echo " ✓ MATCH_STATE_UPDATED events (scorecard updates)"
echo " ✓ PLAYER_FATIGUE_WARNING events (performance monitoring)"
echo " ✓ STRATEGY_RECOMMENDATION events (strategic analysis)"
echo " ✓ BOWLING_CHANGE_APPROVED events (coordination decisions)"
echo ""
echo "Step 5: Verify State Consistency"
echo "Checking that all agents operate on same state version..."
python -c "
from src.state.state_manager import MatchStateManager
from src.domain.cricket_entities import CricketMatch, CricketTeam, CricketPlayer, PlayerRole
state_mgr = MatchStateManager()
match = CricketMatch(
match_name='Test',
team_1=CricketTeam(team_name='India'),
team_2=CricketTeam(team_name='Australia'),
toss_winner_id='IND'
)
import asyncio
asyncio.run(state_mgr.update_match_state(match))
current, version = state_mgr.get_current_state()
print(f'State initialized with version: {version}')
print(f'State version consistency check: {state_mgr.verify_state_consistency(version)}')
" 2>/dev/null
echo ""
echo "Step 6: Expected Test Output Patterns"
echo ""
echo "Sample Output (First 5 Deliveries):"
echo "---"
echo "2024-01-15T10:30:45.123 | INFO | MatchStateAgent-1: Delivery processed. Match: 4/0"
echo "2024-01-15T10:30:45.456 | INFO | PerformanceAgent-1: Jasprit Bumrah - Spell: 1 deliveries, Fatigue: 4.17%"
echo "2024-01-15T10:30:45.789 | INFO | StrategyAgent-1: Run Rate: 4.00, Required Rate: 10.00"
echo "2024-01-15T10:30:46.012 | INFO | CoordinationAgent-1: Executing strategy - AGGRESSIVE"
echo "---"
echo ""
echo "Step 7: Final Match Summary"
echo "Expected Summary (Example):"
echo ""
echo "Match: India vs Australia - T20 World Cup Final"
echo "Final Score: 156/4 in 20 overs"
echo "Run Rate: 7.80"
echo "Total Events Processed: 145 (120 deliveries + 25 coordinate events)"
echo "Agent Decisions Made: 18 (5 fatigue warnings, 4 strategy recs, 9 bowling changes)"
echo ""
echo "=== Test Verification Complete ==="Warning: State Version Mismatches - The most common error is agents processing stale state where they read match state, make a decision, but by the time they execute it, another agent has updated the state version. Agents must always check state consistency via `_check_state_consistency()` before acting. If an agent gets a version mismatch, it should reload current state rather than executing decisions based on outdated information. Example error: PerformanceAgent recommends removing Bumrah from bowling, but StrategyAgent already has him for the final over and incremented state version in between. Solution: Both agents publish their recommendations to the event bus; CoordinationAgent sees both and makes the authoritative decision based on latest state version. Always use the pattern: (1) Read state + get version, (2) Make decision, (3) Check state hasn't changed, (4) Execute or abort if version changed.
Extension Challenge: Implement a Learning Loop where agents track decision outcomes. After each match simulation, calculate a success metric for each agent's decisions (e.g., did the PerformanceAgent's rest recommendation help bowlers stay fresh for death overs?). Store these metrics in a decision history database, then use them to weight future recommendations. For example, if Strategy Agent's 'aggressive' recommendation led to more boundaries in similar match situations historically, weight it more heavily next time. This creates adaptive agents that improve decision quality over time—the foundation of reinforcement learning in agentic systems. Implement a simple feedback loop: (1) Store decision + outcome after each match, (2) At start of next match, calculate success rate for each agent's similar decisions, (3) Adjust agent recommendation weights based on historical success. This demonstrates how multi-agent systems evolve from reactive (responding to events) to learning (improving decisions based on experience).
- Multi-agent orchestration requires centralized state management with versioning to prevent agents from making decisions on stale information or conflicting with each other
- Event-driven pub-sub architecture decouples agents while maintaining coordination—agents publish what happened, subscribe to what matters, enabling independent development and testing
- Agent specialization (MatchStateAgent, PerformanceAgent, StrategyAgent) mirrors organizational structure in cricket teams where expertise areas handle specific concerns and communicate results
- Conflict resolution in multi-agent systems requires a CoordinationAgent that understands constraints from all agents and makes authoritative decisions respecting both rule compliance and strategy
- Feedback loops enable agent learning—tracking decision outcomes allows future agent recommendations to be weighted based on historical success, creating systems that improve over time
- Production agentic systems require comprehensive logging, state audit trails, and decision traceability to debug agent behavior and demonstrate correctness to stakeholders