This capstone project challenges you to build an intelligent AI agent system that analyzes live cricket match data, generates real-time insights, and makes predictive recommendations using agentic workflows. The system integrates multiple autonomous agents that collaborate to monitor match conditions, compute performance metrics, predict outcomes, and suggest strategic decisions.
To accomplish this, you will implement tool-calling mechanisms for data retrieval, decision-making workflows that chain multiple AI reasoning steps, and state management to track agent decisions across match phases. These components come together to demonstrate production-grade agentic patterns, including tool definitions with proper input validation, agent orchestration for concurrent analysis tasks, memory management for long-running match workflows, and graceful error recovery when data is unavailable or predictions are uncertain.
As a portfolio piece, this project demonstrates mastery of modern AI system design — a discipline that is increasingly central to enterprise software development, data science applications, and autonomous decision-making platforms across industries such as fintech, healthcare, and sports analytics.
Learning Objectives
- Design and implement multi-agent systems where specialized agents collaborate asynchronously, each with defined responsibilities, tool sets, and decision scopes.
- Build tool-calling infrastructure with rigorous schema validation, enabling agents to reliably interact with external data sources and computation services.
- Implement agent state machines that track decision history, reasoning chains, and confidence levels, supporting audit trails and explainability for stakeholder review.
- Engineer memory systems that persist agent observations and decisions across long-running workflows, handling state consistency in distributed scenarios.
- Develop orchestration patterns that coordinate multiple concurrent agents, merge their outputs, and resolve conflicts when recommendations diverge or data becomes stale.
- Apply error handling, graceful degradation, and uncertainty quantification so agents provide confidence scores and fallback recommendations when data is incomplete.
Technical Requirements
- Implement at least three specialized agents (MatchPerformanceAgent, PredictionAgent, StrategyRecommenderAgent) with distinct tool sets and decision-making logic.
- Define tool schemas for match data retrieval, historical statistics lookup, statistical computation, and outcome prediction with input validation and error responses.
- Support live match event streaming (wickets, runs, milestones) with agent event handlers that trigger analysis workflows and update shared match state in real-time.
- Persist agent decisions and reasoning chains to a structured audit log, enabling replay, debugging, and stakeholder transparency of AI recommendations.
- Implement confidence scoring where each agent reports decision confidence (0.0–1.0), allowing downstream systems to weight recommendations and flag uncertain predictions.
- Provide a query interface where users can ask agents questions like 'What is the current win probability?' or 'Should we accelerate or consolidate?' and receive multi-agent responses.
- Include comprehensive error handling for missing data, network failures, and stale predictions; agents must gracefully degrade to fallback recommendations.
- Validate all agent outputs against match rules (e.g., recommended batting order must contain valid players; predicted score must be within physical bounds).
Architecture & Design
The system architecture follows a hub-and-spoke agent model centered on a MatchState broker, which serves as the single source of truth for match conditions, player statistics, and ball-by-ball events. Each specialized agent — MatchPerformanceAgent, PredictionAgent, StrategyRecommenderAgent, and RiskAssessmentAgent — is a stateless, asynchronous service that subscribes to match events, retrieves relevant context from the broker, invokes LLM-based reasoning with tool-calling enabled, and publishes decisions back to the broker along with confidence metadata.
The ToolRegistry acts as a centralized registry that defines all tool schemas, validates inputs against JSON schemas, and routes tool calls to the appropriate backend services. These backend services include the MatchDataService for scorecard queries, the StatisticalComputeService for probability models, and the HistoricalAnalyticsService for player performance trends.
The WorkflowOrchestrator manages the overall decision pipeline. When a wicket event fires, it dispatches the event to all subscribed agents, collects their decisions within a defined timeout window, merges non-conflicting recommendations, and escalates conflicts to a consensus algorithm.
The AuditLogger captures every agent decision with full context — including the input state, LLM prompt, tool calls made, outputs, confidence score, and timestamp — enabling post-match analysis and ML-based monitoring of agent quality.
The system is also designed to handle failures gracefully. If an agent crashes, the remaining agents continue operating. If a tool times out, the affected agent receives a retriable error and can invoke fallback logic. If data is stale, agents explicitly report uncertainty rather than producing unreliable predictions.
Phase 1 — Core Implementation
Phase 1 implements the foundational component of the system: a MatchPerformanceAgent that analyzes real-time match conditions — including run rate, wicket loss rate, and current batsman form — by calling data retrieval tools and producing tactical insights. This agent reads the scorecard, identifies whether the batting team is ahead of or behind their target run rate, measures scoring consistency, and flags momentum shifts.
The agent uses tool-calling to fetch player-specific statistics and applies straightforward statistical logic — for example, recommending acceleration when the current run rate falls below the target — to generate confidence-weighted recommendations. In practice, you will define the agent's tool schema, implement the LLM-based analysis loop in which the agent iterates between reasoning and tool invocation, and ensure that all decisions are logged with reasoning traces and confidence metadata.
This phase establishes the core agent-tool interaction pattern and data flow that all subsequent phases of the project extend.
# Phase 1: Core MatchPerformanceAgent Implementation
import asyncio
from datetime import datetime
from typing import Dict, Any, List
import uuid
import json
from enum import Enum
class PerformanceMetric(Enum):
ON_PACE = "on_pace"
AHEAD_OF_PACE = "ahead_of_pace"
BEHIND_PACE = "behind_pace"
CRITICAL = "critical"
class MatchPerformanceAgent:
"""
Analyzes current match performance, identifying momentum, run rate trends,
and recommending tactical adjustments based on live field observations.
Like Virat Kohli observing the IPL match: reads scoreboard, bowler form,
field placement, and makes real-time batting decisions without external consultants.
"""
def __init__(self, agent_id: str = None):
self.agent_id = agent_id or str(uuid.uuid4())
self.observation_history: List[Dict] = []
self.tactical_decisions: List[Dict] = []
async def observe_match_state(self, match_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Phase 1 Step 1: Direct observation of match state
Agent observes: run_rate, balls_remaining, wickets_lost, bowler_form
Returns analysis without asking external systems.
"""
current_runs = match_data.get("current_runs", 0)
balls_faced = match_data.get("balls_faced", 0)
target_score = match_data.get("target_score", 180)
balls_remaining = match_data.get("balls_remaining", 120)
wickets_lost = match_data.get("wickets_lost", 3)
# Calculate current run rate (observed metric)
current_run_rate = (current_runs / balls_faced * 6) if balls_faced > 0 else 0
# Calculate required run rate (what we need)
runs_needed = target_score - current_runs
required_run_rate = (runs_needed / balls_remaining * 6) if balls_remaining > 0 else 0
observation = {
"timestamp": datetime.now().isoformat(),
"agent_id": self.agent_id,
"match_state": {
"current_runs": current_runs,
"target_score": target_score,
"balls_faced": balls_faced,
"balls_remaining": balls_remaining,
"wickets_lost": wickets_lost,
"current_run_rate": round(current_run_rate, 2),
"required_run_rate": round(required_run_rate, 2),
},
"observation_type": "field_state"
}
self.observation_history.append(observation)
return observation
async def analyze_momentum(self, observation: Dict[str, Any]) -> PerformanceMetric:
"""
Phase 1 Step 2: Analyze momentum from observations
Compares current_run_rate vs required_run_rate.
Like Virat reading: "run rate is 6.5, need 8+, two overs left"
"""
match_state = observation["match_state"]
current_rr = match_state["current_run_rate"]
required_rr = match_state["required_run_rate"]
balls_remaining = match_state["balls_remaining"]
# Decision logic based on direct observation
if balls_remaining <= 24: # Final 4 overs - critical phase
if current_rr >= required_rr * 0.95:
return PerformanceMetric.AHEAD_OF_PACE
else:
return PerformanceMetric.CRITICAL
elif current_rr >= required_rr:
return PerformanceMetric.AHEAD_OF_PACE
elif current_rr >= required_rr * 0.9:
return PerformanceMetric.ON_PACE
else:
return PerformanceMetric.BEHIND_PACE
async def identify_bowler_form(self, bowler_name: str, bowler_data: Dict) -> Dict[str, Any]:
"""
Phase 1 Step 3: Assess bowler weakness from observation
Like Virat noting: "this bowler is short-ball specialist, I've been
successful with pull shots against this type"
"""
runs_conceded = bowler_data.get("runs_conceded", 0)
balls_bowled = bowler_data.get("balls_bowled", 0)
wickets_taken = bowler_data.get("wickets_taken", 0)
economy = (runs_conceded / balls_bowled * 6) if balls_bowled > 0 else 0
bowler_analysis = {
"bowler_name": bowler_name,
"economy_rate": round(economy, 2),
"wickets_taken": wickets_taken,
"balls_bowled": balls_bowled,
"weakness": None,
"batting_strategy": None
}
# Direct observation-based assessment
if economy > 10:
bowler_analysis["weakness"] = "loose_lines"
bowler_analysis["batting_strategy"] = "drive_aggressively"
elif economy < 5:
bowler_analysis["weakness"] = "tight_lines"
bowler_analysis["batting_strategy"] = "play_straight_and_wait"
elif runs_conceded > 20:
bowler_analysis["weakness"] = "tired"
bowler_analysis["batting_strategy"] = "attack_short_ball"
else:
bowler_analysis["weakness"] = "balanced"
bowler_analysis["batting_strategy"] = "respect_and_construct"
return bowler_analysis
async def decide_batting_strategy(
self,
performance_metric: PerformanceMetric,
bowler_form: Dict[str, Any],
field_placement: List[str]
) -> Dict[str, Any]:
"""
Phase 1 Step 4: Make tactical decision from observations
Agent calls on experience and direct observation—no external consultants.
"""
decision = {
"timestamp": datetime.now().isoformat(),
"agent_id": self.agent_id,
"performance_status": performance_metric.value,
"bowler_assessment": bowler_form["bowler_name"],
"recommended_action": None,
"risk_level": None,
"confidence": None
}
# Decision logic: Direct observation → Action
if performance_metric == PerformanceMetric.CRITICAL:
if "slip" in field_placement and bowler_form["weakness"] == "loose_lines":
decision["recommended_action"] = "play_defensively_next_two_overs"
decision["confidence"] = 0.85
decision["risk_level"] = "high"
else:
decision["recommended_action"] = "aggressive_shot_making"
decision["confidence"] = 0.75
decision["risk_level"] = "very_high"
elif performance_metric == PerformanceMetric.BEHIND_PACE:
if bowler_form["batting_strategy"] == "drive_aggressively":
decision["recommended_action"] = "target_boundaries_next_over"
decision["confidence"] = 0.80
decision["risk_level"] = "medium"
else:
decision["recommended_action"] = "construct_and_rotate_strike"
decision["confidence"] = 0.78
decision["risk_level"] = "low"
elif performance_metric == PerformanceMetric.ON_PACE:
decision["recommended_action"] = "maintain_current_approach"
decision["confidence"] = 0.90
decision["risk_level"] = "low"
else: # AHEAD_OF_PACE
decision["recommended_action"] = "consolidate_position_build_partnership"
decision["confidence"] = 0.88
decision["risk_level"] = "very_low"
self.tactical_decisions.append(decision)
return decision
async def main():
"""
End-to-end demo: Agent observes, analyzes, and decides like Virat Kohli
"""
# Initialize the MatchPerformanceAgent
virat_agent = MatchPerformanceAgent(agent_id="virat_kohli_agent_v1")
print("🏏 MatchPerformanceAgent: Real-time Field Analysis")
print("=" * 60)
# Scenario 1: Match in progress - On Pace
print("\n📊 Scenario 1: IPL Match - Rohit Sharma's Turn to Bat")
print("-" * 60)
match_data_scenario1 = {
"current_runs": 78,
"balls_faced": 65,
"target_score": 180,
"balls_remaining": 55,
"wickets_lost": 2
}
observation1 = await virat_agent.observe_match_state(match_data_scenario1)
print(f"✓ Observation: {observation1['match_state']}")
performance1 = await virat_agent.analyze_momentum(observation1)
print(f"✓ Momentum Status: {performance1.value}")
jasprit_data = {
"runs_conceded": 15,
"balls_bowled": 18,
"wickets_taken": 1
}
bowler_form1 = await virat_agent.identify_bowler_form("Jasprit Bumrah", jasprit_data)
print(f"✓ Bowler Analysis: {bowler_form1['bowler_name']} - "
f"Economy: {bowler_form1['economy_rate']}, Strategy: {bowler_form1['batting_strategy']}")
field_placement1 = ["slip", "gully", "cover", "mid_off", "mid_on"]
decision1 = await virat_agent.decide_batting_strategy(performance1, bowler_form1, field_placement1)
print(f"✓ Tactical Decision: {decision1['recommended_action']} "
f"(Confidence: {decision1['confidence']}, Risk: {decision1['risk_level']})")
# Scenario 2: Critical phase - Behind pace, need acceleration
print("\n\n📊 Scenario 2: Final Overs - Acceleration Phase Needed")
print("-" * 60)
match_data_scenario2 = {
"current_runs": 132,
"balls_faced": 108,
"target_score": 175,
"balls_remaining": 12,
"wickets_lost": 4
}
observation2 = await virat_agent.observe_match_state(match_data_scenario2)
print(f"✓ Observation: {observation2['match_state']}")
performance2 = await virat_agent.analyze_momentum(observation2)
print(f"✓ Momentum Status: {performance2.value} ⚠️ CRITICAL!")
aggressive_bowler_data = {
"runs_conceded": 28,
"balls_bowled": 18,
"wickets_taken": 0
}
bowler_form2 = await virat_agent.identify_bowler_form("Death Overs Specialist", aggressive_bowler_data)
print(f"✓ Bowler Analysis: {bowler_form2['bowler_name']} - "
f"Economy: {bowler_form2['economy_rate']}, Weakness: {bowler_form2['weakness']}")
field_placement2 = ["deep_mid_wicket", "deep_cover", "long_on"]
decision2 = await virat_agent.decide_batting_strategy(performance2, bowler_form2, field_placement2)
print(f"✓ Tactical Decision: {decision2['recommended_action']} "
f"(Confidence: {decision2['confidence']}, Risk: {decision2['risk_level']})")
# Summary
print("\n\n📈 Agent Performance Summary")
print("=" * 60)
print(f"Total Observations: {len(virat_agent.observation_history)}")
print(f"Total Decisions Made: {len(virat_agent.tactical_decisions)}")
print("\n✅ Phase 1 Complete: Agent observes field, calls on experience,")
print(" makes real-time decisions without external consultants.")
if __name__ == "__main__":
asyncio.run(main())