In this capstone project, you will build a live cricket match analytics AI agent that autonomously processes real-time match data, generates contextual insights, and recommends strategic decisions to coaches and commentators. The system integrates multiple specialized sub-agents working in concert: a state-tracker agent that maintains match context such as wickets, runs, and required run rate; a pattern-recognition agent that identifies batting and bowling trends; and a decision-recommender agent that synthesizes insights and suggests tactical actions.
The system is built on agentic workflows with tool-use patterns, meaning agents invoke external tools — including APIs for player statistics, historical match data, and ball-by-ball feeds — and collaborate through a coordinator agent that routes requests and aggregates outputs. This architecture demonstrates production-grade agent design principles: autonomous goal decomposition, tool orchestration, context persistence across agent calls, error recovery, and real-time performance under streaming data.
Completing this project positions you to architect agent systems across a wide range of domains, including sports analytics, supply-chain optimization, financial trading, and customer service — all fields where multi-agent coordination and rapid decision-making under uncertainty are critical capabilities.
Analogy🏏Cricket
🏏 Think of it like cricket: A Test match requires multiple specialized roles working in coordinated harmony—the captain (agent orchestrator) delegates decisions to batting advisors (performance agents), bowling strategists (prediction agents), and fielding coordinators (resource allocation agents). Just as the captain reads the pitch conditions, player form, and opposition strategy to make tactical calls, your agent system reads match telemetry, historical data, and probability models to recommend actions. The batting advisor might say 'wickets are falling; shift to defense,' the bowling strategist might say 'the opposing batsman has an off-side weakness; adjust field placements,' and the prediction agent might say 'at current run rate, we'll reach 250 by the 65th over.' Each agent uses specific tools—access to scorecard data, historical performance databases, and statistical models—just as captains consult their coaches, the dressing room team, and gut feel. The key insight is that no single agent has perfect information; instead, agents communicate their recommendations, verify them against shared facts (the live scorecard), and the system converges on the best decision. This mirrors how agentic workflows succeed: not through monolithic AI, but through specialized agents with limited scope, clear tool access, and structured communication protocols.
Learning Objectives
- Design and implement multi-agent systems with clear agent roles (state tracking, pattern recognition, decision recommendation) and orchestration patterns.
- Build tool-using agents that autonomously invoke external APIs/data sources and parse results to inform decisions—e.g., agents fetch player stats, historical data, real-time ball-by-ball feeds.
- Implement agentic workflow patterns: coordinator agents routing requests, feedback loops, context propagation, and error recovery across distributed agent calls.
- Engineer real-time state management in streaming data contexts: maintain match context, update incrementally as new balls/wickets occur, and ensure consistency across agent viewpoints.
- Develop agent evaluation frameworks: measure agent recommendation accuracy, latency, and collaborative effectiveness; implement testing for edge cases (unusual match situations, data gaps).
- Apply production-readiness patterns: structured logging, graceful error handling, rate-limiting on tool calls, fallback strategies when external data unavailable.
Technical Requirements
- Multi-Agent Architecture: Implement at least three specialized agents (state tracker, pattern analyzer, decision recommender) with distinct responsibilities and autonomous execution.
- Tool Integration: Each agent must call at least two external tools (APIs, data repositories, or simulated data sources) to fetch contextual information—player stats, bowling economy, historical matchups.
- Workflow Coordination: Design a coordinator/orchestrator that routes requests to appropriate agents, aggregates results, handles conflicting recommendations, and maintains execution order.
- Real-Time State Management: Maintain a centralized match state model (ScoreState, InningsTracker) updated as new ball events arrive; ensure all agents read consistent state and handle concurrent updates safely.
- Streaming Data Handling: Simulate or integrate with a live ball-by-ball data feed; agents must process events incrementally and produce insights within latency budget (e.g., under 500ms per ball).
- Prompt Engineering & Reasoning: Use structured prompts for each agent role; include examples of past match situations and expected agent reasoning—demonstrate that agents understand cricket context, not just string manipulation.
- Error Resilience & Fallbacks: Handle missing data (external API down), data inconsistencies, and invalid match states; agents must degrade gracefully and provide best-effort recommendations.
- Evaluation & Monitoring: Implement metrics to assess agent recommendations (accuracy vs ground truth, recommendation diversity, response latency) and logging to replay/debug agent decision processes.
Architecture & Design
The system architecture follows a modular, agent-centric pattern with a clear separation of concerns. At its core is a MatchCoordinator that receives ball-by-ball events and orchestrates a pipeline of specialized agents. The StateTrackerAgent maintains a canonical MatchState object containing the current score, wickets, run-rate metrics, and match phase — powerplay, middle overs, or death — updated after each delivery.
The PatternAnalyzerAgent processes both historical and streaming data by querying a PlayerStatsRepository that contains career statistics, recent form, and head-to-head records. It then applies pattern-detection logic — for example, identifying that a batsman has a lower strike rate against spinners, or that a bowler concedes more boundaries in death overs — to produce trend insights for downstream agents.
The DecisionRecommenderAgent receives these patterns alongside the current match state, applies decision-making rules such as recommending a short-ball bowling strategy when the required run rate exceeds ten and the batsman has a weak record against short-pitch deliveries, and returns ranked recommendations accompanied by confidence scores.
Agents communicate through an event-driven message bus. The coordinator publishes MatchEvents — including BallDelivered, WicketFallen, and Milestone — agents subscribe and react to these events, and recommendations flow back through a ResultQueue. Data persistence is handled by a match database that stores all deliveries, wickets, and agent outputs, complemented by a cache layer for high-frequency queries such as player statistics and recent in-match patterns.
The architecture prioritizes observability: every agent call logs its input context, reasoning steps, tool invocations, and output, enabling post-match analysis of agent behavior and iterative refinement. Error handling is equally deliberate — agents catch exceptions during tool calls, implement retry logic with exponential backoff, and return 'uncertain' recommendations rather than crashing when data is incomplete.
Analogy🏏Cricket
🏏 Think of it like cricket: The architecture mirrors the India cricket team's decision-making structure during a Test match—Rohit Sharma (MatchState broker) is the captain who holds the current state: 'It's the 45th over, we're 120/3, Jasprit Bumrah has bowled 8 overs.' The assistant coach (PredictionAgent) analyzes historical data and says 'In similar conditions, teams average 280; we're tracking at 0.85 run rate, so predict final score of 240 ± 15.' The bowling coach (StrategyRecommenderAgent) suggests 'The left-arm quick has a 65% success rate against this batsman; move third slip closer.' The fielding coach (MatchPerformanceAgent) reports 'Our catching success today is 40%, below our 52% average; tighten positioning.' Each coach has specific tools—access to scorecards, statistical databases, player profiles. When a new batsman arrives (wicket event), the coaches instantly react with updated advice. The captain merges their inputs: if all agree 'tighten the field,' he acts; if coaches disagree (one says 'aggressive,' another says 'defensive'), he applies a consensus rule. The team physiotherapist (AuditLogger) documents every decision and outcome, so after the match they can analyze which recommendations worked and which failed. This reveals why agentic architecture is powerful: specialists don't interfere with each other, information flows through a trusted broker, and decisions are made faster because parallel analysis beats sequential debate.
python
# Architecture Skeleton: Cricket AI Agent System
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum
from datetime import datetime
import json
# ===== DATA Models =====
class MatchPhase(Enum):
POWERPLAY = "powerplay"
MIDDLE_OVERS = "middle_overs"
DEATH = "death"
class BallOutcome(Enum):
DOT = 0
SINGLE = 1
BOUNDARY = 4
WICKET = "W"
@dataclass
class BallEvent:
"""Represents a single ball delivery in cricket."""
match_id: str
over_number: int
ball_number: int
bowler: str
batter: str
runs: int
outcome: BallOutcome
timestamp: datetime
ball_type: str # "pace", "spin", "yorker", etc.
@dataclass
class MatchState:
"""Centralized match context maintained by StateTrackerAgent."""
match_id: str
innings: int
team_batting: str
team_fielding: str
runs_scored: int
wickets_down: int
balls_faced: int
target: Optional[int] = None
phase: MatchPhase = MatchPhase.POWERPLAY
last_six_balls: List[int] = field(default_factory=list)
recent_patterns: Dict[str, str] = field(default_factory=dict)
@property
def run_rate(self) -> float:
overs = self.balls_faced / 6
return self.runs_scored / overs if overs > 0 else 0
@property
def required_run_rate(self) -> Optional[float]:
if self.target is None:
return None
overs_remaining = (120 - self.balls_faced) / 6
runs_needed = self.target - self.runs_scored
return runs_needed / overs_remaining if overs_remaining > 0 else None
@dataclass
class Recommendation:
"""Agent recommendation with confidence and reasoning."""
agent_name: str
action: str
confidence: float # 0.0 to 1.0
reasoning: str
timestamp: datetime
@dataclass
class PlayerStats:
"""Historical player statistics."""
player_name: str
career_runs: int
matches_played: int
avg_strike_rate: float
recent_form: float # avg runs in last 5 innings
vs_pace_avg: float
vs_spin_avg: float
death_overs_avg: float
# ===== File Structure =====
PROJECT_STRUCTURE = """
cricket_ai_agent/
├── agents/
│ ├── __init__.py
│ ├── state_tracker_agent.py # Maintains match state
│ ├── pattern_analyzer_agent.py # Identifies trends
│ ├── decision_recommender_agent.py # Recommends actions
│ └── base_agent.py # Abstract agent base class
├── coordinators/
│ ├── __init__.py
│ └── match_coordinator.py # Orchestrates agent calls
├── tools/
│ ├── __init__.py
│ ├── player_stats_tool.py # Fetches player data
│ ├── historical_data_tool.py # Queries past matches
│ └── real_time_feed_tool.py # Ball-by-ball stream
├── models/
│ ├── __init__.py
│ └── cricket_models.py # Data classes (above)
├── persistence/
│ ├── __init__.py
│ ├── match_database.py # Stores match data
│ └── cache_layer.py # In-memory cache
├── evaluation/
│ ├── __init__.py
│ ├── metrics.py # Agent evaluation metrics
│ └── test_suite.py # Test cases
├── config.yaml # Configuration (API keys, thresholds)
├── main.py # Entry point
└── requirements.txt # Dependencies
"""
# ===== Base Agent Class =====
class Agent:
"""Abstract base class for all agents."""
def __init__(self, agent_name: str, logger=None):
self.agent_name = agent_name
self.logger = logger
self.call_history = []
def execute(self, match_state: MatchState, context: Dict) -> Dict:
"""Execute agent logic. Subclasses override this."""
raise NotImplementedError
def log_execution(self, input_data: Dict, output: Dict):
"""Log agent execution for observability."""
log_entry = {
"agent": self.agent_name,
"timestamp": datetime.now().isoformat(),
"input": input_data,
"output": output
}
self.call_history.append(log_entry)
if self.logger:
self.logger.info(json.dumps(log_entry))
# ===== Coordinator Blueprint =====
class MatchCoordinator:
"""Orchestrates multi-agent workflow for live match analysis."""
def __init__(self):
self.state: Optional[MatchState] = None
self.agents: Dict[str, Agent] = {}
self.recommendations: List[Recommendation] = []
def register_agent(self, agent: Agent):
"""Register an agent with the coordinator."""
self.agents[agent.agent_name] = agent
def process_ball_event(self, event: BallEvent) -> List[Recommendation]:
"""Process a single ball and invoke agents."""
# 1. Update state (StateTrackerAgent updates self.state)
# 2. Analyze patterns (PatternAnalyzerAgent produces insights)
# 3. Recommend actions (DecisionRecommenderAgent produces recommendations)
# 4. Return aggregated recommendations
recommendations = []
# Agent execution order ensures state is current before analysis
for agent_name in ["StateTracker", "PatternAnalyzer", "DecisionRecommender"]:
if agent_name in self.agents:
agent = self.agents[agent_name]
context = {
"current_event": event,
"match_state": self.state,
"previous_recommendations": recommendations
}
result = agent.execute(self.state, context)
recommendations.extend(result.get("recommendations", []))
return recommendations
print("Architecture Definition Complete")
print("\nProject Structure:")
print(PROJECT_STRUCTURE)Phase 1 — Core Implementation
Phase 1 focuses on establishing the foundational agent infrastructure. This involves implementing the StateTrackerAgent, which parses ball-by-ball events and maintains authoritative match state, integrating a basic PatternAnalyzerAgent that queries a player statistics repository and identifies straightforward patterns such as a batter being in form or a bowler being economical, and building the MatchCoordinator that sequences these agents and routes events between them.
During this phase, the system operates on mock and static data — specifically, a pre-loaded PlayerStats repository populated with realistic cricket player data drawn from IPL and international cricket, alongside a simulated ball-by-ball stream represented as a JSON file or an in-memory event queue covering a full T20 innings. The primary focus is on agent abstraction, tool invocation patterns, and the event processing pipeline rather than on complex reasoning. Agents apply straightforward rules, and the coordinator is responsible for ensuring correct sequencing and context propagation throughout.
Analogy🏏Cricket
🏏 Think of it like cricket: Virat Kohli during an IPL match constantly reads the scoreboard, bowler form, and field placement. He doesn't decide his next shot by asking consultants; he directly observes (run rate is 6.5, needs 8+, two overs left), calls on experience (last bowler was short-ball specialist, I've been successful with pull shots), and decides. Phase 1 builds this observational agent. The MatchPerformanceAgent is like Virat analyzing the field in real time: 'I see the ball is swinging, three slips are in place, bowler's economy is low—I should play defensively for the next two overs.' It uses three tools: check-the-scorecard (fetch current runs, overs), check-the-bowler-record (fetch bowler's economy rate, recent wickets), and estimate-run-projection (calculate what score we'll reach at current pace). Each tool call is recorded, so later we can analyze which observations led to which decisions. The insight is that an agent's quality depends on its observation tools and how it uses them—Virat succeeds because he observes cricket conditions accurately and reasons about them correctly, not because he's overloaded with advice.
python
# Phase 1: Core Implementation - StateTrackerAgent & PatternAnalyzerAgent
# 🏏 Production-Ready Agentic System for Cricket Match Analysis
from datetime import datetime
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from enum import Enum
import json
import logging
# ===== Base Agent Class =====
class Agent:
"""Base class for all specialized agents in the system."""
def __init__(self, agent_name: str, logger=None):
self.agent_name = agent_name
self.logger = logger or logging.getLogger(agent_name)
def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
"""Execute agent logic. To be overridden by subclasses."""
raise NotImplementedError
# ===== Data Models =====
@dataclass
class CricketPlayer:
"""Represents a cricket player with stats."""
player_name: str
player_id: int
role: str # "batter", "bowler", "allrounder"
matches_played: int = 0
runs_scored: int = 0
wickets_taken: int = 0
average_runs: float = 0.0
@dataclass
class BallEvent:
"""Represents a single ball delivery in cricket."""
ball_number: int
bowler: str
batter: str
runs: int
wicket: bool = False
dot_ball: bool = False # Ball where no runs scored
timestamp: datetime = field(default_factory=datetime.now)
@dataclass
class InningsState:
"""Represents the state of an innings."""
innings_count: int
batting_team: str
bowling_team: str
total_runs: int = 0
wickets_fallen: int = 0
overs_completed: float = 0.0
current_batter: str = ""
striker_runs: int = 0
non_striker_runs: int = 0
balls_faced: List[BallEvent] = field(default_factory=list)
@dataclass
class MatchState:
"""Canonical match state maintained by StateTrackerAgent."""
match_id: str
match_date: datetime
teams: tuple # (team1, team2)
innings_history: List[InningsState] = field(default_factory=list)
current_innings: Optional[InningsState] = None
match_status: str = "not_started" # "not_started", "live", "completed"
# ===== Phase 1: StateTrackerAgent =====
class StateTrackerAgent(Agent):
"""
Maintains canonical match state, updated after each ball.
Like a batter 'settling in' by understanding the pitch and pace.
"""
def __init__(self, logger=None):
super().__init__("StateTracker", logger)
self.match_state: Optional[MatchState] = None
def initialize_match(self, match_id: str, team1: str, team2: str) -> MatchState:
"""Initialize a new match state - 'settling in' phase."""
self.match_state = MatchState(
match_id=match_id,
match_date=datetime.now(),
teams=(team1, team2),
match_status="live"
)
self.logger.info(f"Match initialized: {team1} vs {team2}")
return self.match_state
def start_innings(self, batting_team: str, bowling_team: str, innings_count: int) -> InningsState:
"""Start a new innings and update match state."""
new_innings = InningsState(
innings_count=innings_count,
batting_team=batting_team,
bowling_team=bowling_team,
current_batter=batting_team
)
self.match_state.current_innings = new_innings
self.match_state.innings_history.append(new_innings)
self.logger.info(f"Innings {innings_count} started: {batting_team} batting")
return new_innings
def update_with_ball(self, ball_event: BallEvent) -> InningsState:
"""
Update match state after each ball delivery.
Like reading each delivery to understand the bowler's line and pace.
"""
innings = self.match_state.current_innings
# Update runs
innings.total_runs += ball_event.runs
innings.striker_runs += ball_event.runs
# Record the ball
innings.balls_faced.append(ball_event)
# Update overs (6 balls = 1 over)
balls_count = len(innings.balls_faced)
innings.overs_completed = (balls_count // 6) + (balls_count % 6) / 10
# Update wickets
if ball_event.wicket:
innings.wickets_fallen += 1
self.logger.warning(f"Wicket! {ball_event.batter} out. Total: {innings.wickets_fallen}-{innings.total_runs}")
elif ball_event.dot_ball:
self.logger.debug(f"Dot ball by {ball_event.bowler}")
else:
self.logger.info(f"{ball_event.batter} scored {ball_event.runs} runs")
return innings
def get_current_state(self) -> Dict[str, Any]:
"""Retrieve the current canonical match state."""
if not self.match_state or not self.match_state.current_innings:
return {"error": "No active match"}
innings = self.match_state.current_innings
return {
"match_id": self.match_state.match_id,
"innings": innings.innings_count,
"batting_team": innings.batting_team,
"total_runs": innings.total_runs,
"wickets": innings.wickets_fallen,
"overs": innings.overs_completed,
"balls_delivered": len(innings.balls_faced),
"striker_runs": innings.striker_runs
}
def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
"""Execute state tracking logic."""
action = context.get("action")
if action == "initialize":
self.initialize_match(
context["match_id"],
context["team1"],
context["team2"]
)
return {"status": "match_initialized"}
elif action == "start_innings":
self.start_innings(
context["batting_team"],
context["bowling_team"],
context["innings_count"]
)
return {"status": "innings_started"}
elif action == "record_ball":
ball = BallEvent(**context["ball_event"])
self.update_with_ball(ball)
return {"status": "ball_recorded", "state": self.get_current_state()}
elif action == "get_state":
return self.get_current_state()
return {"error": "Unknown action"}
# ===== Phase 1: PatternAnalyzerAgent =====
class PatternAnalyzerAgent(Agent):
"""
Identifies straightforward patterns from player stats and match state.
Like reading the pitch - understanding conditions without aggressive risk-taking.
"""
def __init__(self, player_database: Dict[str, CricketPlayer], logger=None):
super().__init__("PatternAnalyzer", logger)
self.player_database = player_database
self.patterns_detected: List[Dict[str, Any]] = []
def analyze_batter_form(self, batter_name: str) -> Dict[str, Any]:
"""Analyze current form of a batter."""
if batter_name not in self.player_database:
return {"error": f"Player {batter_name} not found"}
player = self.player_database[batter_name]
form_status = "good" if player.average_runs > 40 else "average" if player.average_runs > 25 else "poor"
pattern = {
"player": batter_name,
"matches_played": player.matches_played,
"total_runs": player.runs_scored,
"average_runs": player.average_runs,
"form_status": form_status
}
self.patterns_detected.append(pattern)
self.logger.info(f"Batter {batter_name} analysis: {form_status} form (avg: {player.average_runs})")
return pattern
def analyze_bowler_effectiveness(self, bowler_name: str) -> Dict[str, Any]:
"""Analyze effectiveness of a bowler."""
if bowler_name not in self.player_database:
return {"error": f"Player {bowler_name} not found"}
player = self.player_database[bowler_name]
economy = "good" if player.wickets_taken > 20 else "moderate" if player.wickets_taken > 10 else "developing"
pattern = {
"player": bowler_name,
"matches_played": player.matches_played,
"total_wickets": player.wickets_taken,
"effectiveness": economy
}
self.patterns_detected.append(pattern)
self.logger.info(f"Bowler {bowler_name} analysis: {economy} effectiveness (wickets: {player.wickets_taken})")
return pattern
def identify_matchup_patterns(self, batter_name: str, bowler_name: str) -> Dict[str, Any]:
"""Identify patterns in historical batter vs bowler matchups."""
pattern = {
"matchup": f"{batter_name} vs {bowler_name}",
"prediction": "head-to-head analysis requires historical data",
"confidence": "medium"
}
self.patterns_detected.append(pattern)
self.logger.info(f"Matchup pattern detected: {batter_name} vs {bowler_name}")
return pattern
def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
"""Execute pattern analysis logic."""
action = context.get("action")
if action == "analyze_batter":
return self.analyze_batter_form(context["batter_name"])
elif action == "analyze_bowler":
return self.analyze_bowler_effectiveness(context["bowler_name"])
elif action == "analyze_matchup":
return self.identify_matchup_patterns(
context["batter_name"],
context["bowler_name"]
)
elif action == "get_patterns":
return {"detected_patterns": self.patterns_detected}
return {"error": "Unknown action"}
# ===== Phase 1 Demo: Settling In =====
if __name__ == "__main__":
# Setup logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(message)s')
# Create player database - like reviewing team sheet before match
player_db = {
"Rohit Sharma": CricketPlayer("Rohit Sharma", 1, "batter", 150, 7000, 0, 46.67),
"Virat Kohli": CricketPlayer("Virat Kohli", 2, "batter", 180, 7500, 0, 41.67),
"Jasprit Bumrah": CricketPlayer("Jasprit Bumrah", 10, "bowler", 100, 0, 125, 0),
"Hardik Pandya": CricketPlayer("Hardik Pandya", 3, "allrounder", 120, 3500, 80, 29.17),
}
# Initialize agents
state_tracker = StateTrackerAgent()
pattern_analyzer = PatternAnalyzerAgent(player_db)
print("\n" + "="*70)
print("🏏 PHASE 1: SETTLING IN - Foundation Phase")
print("="*70)
# Step 1: Initialize match
print("\n[Step 1] Match Initialization - Settling In")
state_tracker.execute({
"action": "initialize",
"match_id": "IND-AUS-2024-001",
"team1": "India",
"team2": "Australia"
})
# Step 2: Start innings
print("\n[Step 2] Innings Start - Reading the Pitch")
state_tracker.execute({
"action": "start_innings",
"batting_team": "India",
"bowling_team": "Australia",
"innings_count": 1
})
# Step 3: Record some balls
print("\n[Step 3] Recording Ball Deliveries - Understanding Pace & Line")
balls = [
BallEvent(1, "Jasprit Bumrah", "Rohit Sharma", 0, dot_ball=True),
BallEvent(2, "Jasprit Bumrah", "Rohit Sharma", 4),
BallEvent(3, "Jasprit Bumrah", "Rohit Sharma", 1),
BallEvent(4, "Jasprit Bumrah", "Virat Kohli", 0, dot_ball=True),
BallEvent(5, "Jasprit Bumrah", "Virat Kohli", 6),
BallEvent(6, "Jasprit Bumrah", "Virat Kohli", 1),
]
for ball in balls:
state_tracker.execute({
"action": "record_ball",
"ball_event": {
"ball_number": ball.ball_number,
"bowler": ball.bowler,
"batter": ball.batter,
"runs": ball.runs,
"wicket": ball.wicket,
"dot_ball": ball.dot_ball
}
})
# Step 4: Get current state
print("\n[Step 4] Current Match State")
current_state = state_tracker.execute({"action": "get_state"})
print(f"State: {json.dumps(current_state, indent=2)}")
# Step 5: Analyze player patterns
print("\n[Step 5] Pattern Analysis - Reading Form & Effectiveness")
print("\nAnalyzing Batters:")
pattern_analyzer.execute({"action": "analyze_batter", "batter_name": "Rohit Sharma"})
pattern_analyzer.execute({"action": "analyze_batter", "batter_name": "Virat Kohli"})
print("\nAnalyzing Bowlers:")
pattern_analyzer.execute({"action": "analyze_bowler", "bowler_name": "Jasprit Bumrah"})
print("\nAnalyzing Matchups:")
pattern_analyzer.execute({
"action": "analyze_matchup",
"batter_name": "Rohit Sharma",
"bowler_name": "Jasprit Bumrah"
})
# Summary
print("\n" + "="*70)
print("✅ Phase 1 Complete: Foundation established")
print(" - StateTrackerAgent has 'settled in' to parse match state")
print(" - PatternAnalyzerAgent has 'read the pitch' of player form")
print("="*70)
Phase 2 — Feature Completion
Phase 2 introduces the DecisionRecommenderAgent, which synthesizes patterns and match state into actionable tactical recommendations, and also implements feedback loops through which recommendations are evaluated and progressively refined. This phase extends the system with tool-use patterns, enabling agents to invoke simulated external APIs for real-time data fetching. For example, the PatternAnalyzerAgent calls a player_stats_api to retrieve up-to-date career statistics, a head_to_head_api to query historical matchups between a specific batter and bowler, and a recent_form_api to assess current player form.
Robust error handling is also introduced at this stage, including retry logic with exponential backoff and fallback strategies for tool invocations that fail. A recommendation aggregator is added to receive suggestions from the DecisionRecommenderAgent, resolve conflicts between contradictory recommendations, and score each suggestion by confidence and relevance to the current match phase.
Phase 2 also incorporates simple prompt engineering to improve the transparency of agent reasoning. Agent outputs are structured through templates that explicitly articulate the basis for each recommendation — for instance, stating the pattern name, the current match state including runs and phase, and the resulting recommended action — making the reasoning chain legible to developers and end users alike.
Analogy🏏Cricket
🏏 Think of it like cricket: In the middle overs of a T20 innings (when batters are accelerating and the match narrative is forming), the coaching staff escalates from information-gathering to active decision-making. The batting coach not only analyzes the bowler's stats but recommends specific shots ('This bowler concedes more boundaries to off-side drives in overs 7-10; target that area'), the fielding captain makes dynamic field adjustments based on recommendations (moving fielders from square leg to deep mid-wicket after identifying a batter's preference), and the bowling coach suggests bowling changes (recommending Jasprit Bumrah instead of a medium-pacer if the batter is vulnerable to yorkers). Each recommendation is grounded in multiple data points: recent form (batsman's last 5 innings), head-to-head record (how Virat Kohli has faced this bowler historically), and match context (which phase, what's the run-rate pressure). Recommendations sometimes conflict—one coach might suggest aggressive batting while another emphasizes consolidation—so the captain aggregates inputs, weighs confidence, and makes a unified decision. Similarly, Phase 2 adds the DecisionRecommenderAgent that integrates state and patterns into concrete actions ('The batter is in good form (pattern), required run-rate is 10 (state), and their vs-pace average is high (pattern), so recommend aggressive batting'), tools that fetch real-time data (like coaches accessing live stats during a match), and a mechanism to handle conflicting recommendations. This reveals why agent systems become powerful in Phase 2: they don't just observe and analyze—they act, learn from feedback, and improve decision quality through iteration.
python
# Phase 2: DecisionRecommenderAgent & Tool Integration
# Cricket Analytics: From Information-Gathering to Active Decision-Making
from abc import ABC, abstractmethod
import time
from enum import Enum
from typing import Dict, List, Any
from dataclasses import dataclass
# ===== Tool Base Class & Implementations =====
class Tool(ABC):
"""Abstract base for agent tools."""
def __init__(self, tool_name: str, tool_description: str):
self.tool_name = tool_name
self.tool_description = tool_description
@abstractmethod
def execute(self, **kwargs) -> Dict[str, Any]:
"""Execute the tool with given parameters."""
pass
class BowlerStatsAnalyzerTool(Tool):
"""Analyzes bowler statistics against batter patterns."""
def __init__(self):
super().__init__(
"bowler_stats_analyzer",
"Analyzes bowler statistics, boundary patterns, and weaknesses in specific overs"
)
# Mock bowler statistics database
self.bowler_data = {
"Jasprit Bumrah": {
"yorker_accuracy": 0.78,
"boundaries_conceded_overs_7_10": 2,
"favorable_zones": ["yorker line", "high full"],
"weak_areas": ["short ball vs aggressive batters"]
},
"Mohammed Siraj": {
"swing_average": 0.65,
"boundaries_conceded_overs_7_10": 5,
"favorable_zones": ["off-stump line", "good length"],
"weak_areas": ["short ball", "leg-side boundaries"]
}
}
def execute(self, bowler_name: str, over_range: tuple = None) -> Dict[str, Any]:
"""Analyze bowler performance."""
bowler = self.bowler_data.get(bowler_name, {})
analysis = {
"bowler": bowler_name,
"stats": bowler,
"recommendation": None,
"confidence": 0.85
}
if bowler_name == "Jasprit Bumrah" and over_range and over_range[0] <= 7 <= over_range[1]:
analysis["recommendation"] = "Bumrah has low boundary count in overs 7-10 due to yorker accuracy. Use off-side drives sparingly."
elif bowler_name == "Mohammed Siraj" and over_range and over_range[0] <= 7 <= over_range[1]:
analysis["recommendation"] = "Siraj concedes more boundaries in middle overs. Target off-side drives and aggressive short-ball counters."
return analysis
class FieldPositionRecommenderTool(Tool):
"""Recommends optimal field placements based on batter preferences."""
def __init__(self):
super().__init__(
"field_position_recommender",
"Recommends dynamic field adjustments based on batter shot patterns"
)
self.batter_patterns = {
"Rohit Sharma": {
"preferred_shot": "off-side drives",
"weak_zone": "yorker line",
"current_field": ["point", "cover", "mid-off"],
"optimal_field": ["deep mid-wicket", "fine leg", "long-on"]
},
"Virat Kohli": {
"preferred_shot": "backfoot drives",
"weak_zone": "short ball outside off",
"current_field": ["slip", "gully", "cover"],
"optimal_field": ["third man", "square leg", "deep point"]
}
}
def execute(self, batter_name: str, bowler_name: str, current_over: int) -> Dict[str, Any]:
"""Recommend field adjustments."""
pattern = self.batter_patterns.get(batter_name, {})
recommendation = {
"batter": batter_name,
"bowler": bowler_name,
"over": current_over,
"adjustment": None,
"reason": None
}
if batter_name == "Rohit Sharma" and current_over >= 7:
recommendation["adjustment"] = pattern.get("optimal_field")
recommendation["reason"] = f"Rohit prefers {pattern['preferred_shot']} in middle overs. Shift fielders from square leg to deep mid-wicket."
return recommendation
class MatchContextAnalyzerTool(Tool):
"""Analyzes match context (overs, run rate, wickets)."""
def __init__(self):
super().__init__(
"match_context_analyzer",
"Analyzes match situation, acceleration phase, and strategic windows"
)
def execute(self, current_over: int, runs_scored: int, wickets_lost: int, total_overs: int = 20) -> Dict[str, Any]:
"""Analyze match context."""
overs_remaining = total_overs - current_over
run_rate = runs_scored / current_over if current_over > 0 else 0
acceleration_phase = 7 <= current_over <= 15
context = {
"current_over": current_over,
"runs_scored": runs_scored,
"wickets_lost": wickets_lost,
"run_rate": round(run_rate, 2),
"overs_remaining": overs_remaining,
"acceleration_phase": acceleration_phase,
"strategy": None
}
if acceleration_phase:
context["strategy"] = "MIDDLE OVERS: Time for aggressive batting. Escalate decision-making to active recommendations."
elif current_over <= 6:
context["strategy"] = "POWERPLAY: Gather information on bowlers, establish batting patterns."
else:
context["strategy"] = "DEATH OVERS: Conservative batting, calculated risks only."
return context
# ===== Decision Recommender Agent =====
@dataclass
class CricketPlayer:
"""Represents a cricket player."""
name: str
role: str # "batter", "bowler", "all-rounder"
form: float # 0.0 to 1.0
class DecisionRecommenderAgent:
"""Active decision-making agent that recommends batting/fielding strategies."""
def __init__(self, agent_id: str):
self.agent_id = agent_id
self.tools = {
"bowler_analyzer": BowlerStatsAnalyzerTool(),
"field_recommender": FieldPositionRecommenderTool(),
"match_analyzer": MatchContextAnalyzerTool()
}
self.decision_log = []
def analyze_and_recommend(self, match_state: Dict[str, Any]) -> Dict[str, Any]:
"""
Main decision loop: Gather data from tools, synthesize recommendations.
Escalates from information-gathering (early overs) to active decision-making (middle overs).
"""
current_over = match_state["current_over"]
batter = match_state["batter"]
bowler = match_state["bowler"]
print(f"\n🏏 [Agent {self.agent_id}] Analyzing match state at Over {current_over}")
print(f" Batter: {batter.name} | Bowler: {bowler.name}")
# Step 1: Match Context Analysis
context_analysis = self.tools["match_analyzer"].execute(
current_over=current_over,
runs_scored=match_state["runs_scored"],
wickets_lost=match_state["wickets_lost"]
)
print(f" 📊 Match Strategy: {context_analysis['strategy']}")
# Step 2: Escalate decision complexity based on match phase
recommendation = {"phase": None, "actions": []}
if context_analysis["acceleration_phase"]:
# MIDDLE OVERS: Active decision-making phase
print(f" ⚡ ESCALATING to active decision-making (Middle Overs)...")
# Analyze bowler weaknesses
bowler_analysis = self.tools["bowler_analyzer"].execute(
bowler_name=bowler.name,
over_range=(current_over - 1, current_over + 3)
)
print(f" 🎯 Bowler Analysis: {bowler_analysis['recommendation']}")
recommendation["actions"].append({
"type": "batting_guidance",
"detail": bowler_analysis["recommendation"]
})
# Recommend field adjustments (for fielding captain)
field_rec = self.tools["field_recommender"].execute(
batter_name=batter.name,
bowler_name=bowler.name,
current_over=current_over
)
print(f" 🔄 Field Adjustment: {field_rec['reason']}")
recommendation["actions"].append({
"type": "field_adjustment",
"positions": field_rec["adjustment"],
"detail": field_rec["reason"]
})
recommendation["phase"] = "middle_overs_escalation"
else:
# POWERPLAY/DEATH: Information-gathering or conservative approach
recommendation["phase"] = "information_gathering" if current_over <= 6 else "death_overs"
recommendation["actions"].append({
"type": "strategic_note",
"detail": context_analysis["strategy"]
})
# Log decision
self.decision_log.append({
"timestamp": time.time(),
"over": current_over,
"recommendation": recommendation
})
return recommendation
def get_recommendation_summary(self) -> str:
"""Summarize all recommendations made during innings."""
summary = f"\n📋 Decision Summary for Agent {self.agent_id}:\n"
summary += f" Total decisions: {len(self.decision_log)}\n"
escalation_count = sum(1 for d in self.decision_log if d["recommendation"]["phase"] == "middle_overs_escalation")
summary += f" Active escalations (middle overs): {escalation_count}\n"
return summary
# ===== Simulation =====
def simulate_t20_innings():
"""Simulate a T20 innings with active agent decision-making."""
print("=" * 80)
print("🏏 T20 INNINGS SIMULATION: Capstone Production-Ready Agentic System")
print("=" * 80)
# Create players
rohit = CricketPlayer("Rohit Sharma", "batter", 0.92)
bumrah = CricketPlayer("Jasprit Bumrah", "bowler", 0.88)
siraj = CricketPlayer("Mohammed Siraj", "bowler", 0.80)
# Create agents
batting_coach = DecisionRecommenderAgent("Batting-Coach-001")
fielding_captain = DecisionRecommenderAgent("Captain-Field-001")
# Simulate progression through overs
match_states = [
{"current_over": 3, "batter": rohit, "bowler": bumrah, "runs_scored": 18, "wickets_lost": 0},
{"current_over": 7, "batter": rohit, "bowler": siraj, "runs_scored": 52, "wickets_lost": 0}, # ESCALATION POINT
{"current_over": 10, "batter": rohit, "bowler": bumrah, "runs_scored": 85, "wickets_lost": 1},
{"current_over": 16, "batter": rohit, "bowler": siraj, "runs_scored": 132, "wickets_lost": 2}, # Death overs
]
for state in match_states:
batting_rec = batting_coach.analyze_and_recommend(state)
time.sleep(0.5)
# Print summaries
print("\n" + "=" * 80)
print(batting_coach.get_recommendation_summary())
print("=" * 80)
print("\n✅ Capstone Demo Complete: Information-gathering escalated to active decision-making\n")
if __name__ == "__main__":
simulate_t20_innings()