What You'll Build
In this checkpoint exercise, you will construct a multi-agent agentic workflow system that autonomously analyzes live cricket match data, makes real-time strategic recommendations, and manages decision-making hierarchies. The system demonstrates advanced agent coordination patterns through a Primary Agent that orchestrates the overall workflow by decomposing match scenarios into discrete sub-tasks and delegating them to specialist agents — namely a Batting Strategist Agent, a Bowling Analyst Agent, and a Field Placement Agent — each of which operates with its own independent reasoning loop. The Primary Agent then synthesizes their outputs into a coherent, unified match strategy.
The architecture implements continuous feedback loops in which agent decisions are validated against the current match state. Uncertainty is managed through confidence thresholds, and the system adapts its strategy dynamically in response to events such as wicket loss, changes in run rate, or shifts in opposition strength.
This exercise covers agent composition, inter-agent communication protocols, state management across distributed decision-making units, and how agentic systems handle conflicting recommendations produced by parallel reasoning chains. The patterns explored here reflect production approaches used in sports analytics platforms and autonomous decision systems that must operate under real-time constraints and incomplete information.
Prerequisites
- Understanding of agent-based systems, including observation→reasoning→action loops and agent state management.
- Proficiency in asynchronous programming patterns and concurrent execution models for multi-agent coordination.
- Knowledge of API design for inter-agent communication, including message passing and result aggregation protocols.
- Familiarity with decision-making frameworks under uncertainty, confidence scoring, and conflict resolution heuristics.
- Basic understanding of cricket match dynamics: innings structure, powerplay, death overs, wickets, run rate, economy rate.
Setup & Project Structure
Begin by creating a structured Python project directory that isolates agent logic, match state management, and communication protocols into clearly defined modules. The project structure separates concerns across four directories: an `agents/` directory containing individual agent implementations, a `models/` directory holding shared data structures such as Match, Innings, and Bowler records, an `orchestrator/` directory managing Primary Agent logic and agent coordination, and a `utils/` directory providing helper functions for state updates and conflict resolution. This separation ensures that agents operate independently while maintaining clear communication boundaries, making the overall system both scalable and testable.
Install the required dependencies before proceeding. These include `pydantic` for type-safe data models representing match state and agent decisions, `asyncio` for concurrent agent execution, and `typing` for advanced type hints. Create a `requirements.txt` file to lock dependency versions and simplify environment reproduction across different machines.
#!/bin/bash
# Project: cricket-match-analytics-agent
# Setup and directory structure
mkdir -p cricket-analytics-agent
cd cricket-analytics-agent
# Create directory structure
mkdir -p agents orchestrator models utils tests
# Initialize Python project
touch __init__.py agents/__init__.py orchestrator/__init__.py models/__init__.py utils/__init__.py
# Create requirements.txt
cat > requirements.txt << 'EOF'
pydantic==2.0.0
typings-extensions==4.5.0
aiofiles==23.1.0
python-dotenv==1.0.0
EOF
# Install dependencies
pip install -r requirements.txt
# Create project files (placeholder structure)
touch agents/batting_strategist.py agents/bowling_analyst.py agents/field_placement.py
touch orchestrator/primary_agent.py orchestrator/coordinator.py
touch models/match_state.py models/agent_decision.py
touch utils/conflict_resolver.py utils/confidence_scorer.py
touch main.py tests/test_agents.py
echo 'Cricket Analytics Agent project initialized successfully!'
echo 'Project structure:'
tree -L 2 || find . -type f -name '*.py' | head -20Step 1 — Foundation
Step 1 establishes the foundational data models and base agent architecture from which all specialist agents inherit. Using Pydantic, you will define immutable data structures that represent cricket match state: a `Match` model containing teams, format, and current innings; an `InningsState` model tracking runs, wickets, balls faced, and the current batter; a `BowlerStats` model recording overs bowled, runs conceded, wickets taken, and economy; and an `AgentDecision` model capturing recommendation text, a confidence score between 0 and 1, and the agent's reasoning.
The `BaseAgent` abstract class defines the core agent loop through three stages: observe, which grants the agent access to match state; reason, which processes those observations; and act, which generates decisions accompanied by confidence scores. This foundation ensures that all specialist agents follow identical communication protocols, making orchestration reliable and predictable.
The confidence score system is particularly critical because it quantifies uncertainty and enables the Primary Agent to rank conflicting recommendations from parallel agents. Match state is kept immutable throughout this architecture specifically to prevent race conditions when multiple agents execute concurrently.
# models/match_state.py
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
from abc import ABC, abstractmethod
class BowlerStats(BaseModel):
"""Immutable bowler statistics for the current match."""
bowler_name: str
overs_bowled: float = Field(ge=0)
runs_conceded: int = Field(ge=0)
wickets_taken: int = Field(ge=0)
economy_rate: float = Field(ge=0.0)
maiden_overs: int = Field(default=0, ge=0)
class Config:
frozen = True
class InningsState(BaseModel):
"""Immutable current innings state snapshot."""
team_name: str
runs_scored: int = Field(ge=0)
wickets_lost: int = Field(ge=0, le=10)
balls_faced: int = Field(ge=0)
current_batter: str
bowlers_in_spell: List[BowlerStats]
powerplay_active: bool = False
overs_remaining: float = Field(ge=0)
class Config:
frozen = True
@property
def run_rate(self) -> float:
"""Calculate current run rate (runs per over)."""
overs_bowled = self.balls_faced / 6.0
return self.runs_scored / overs_bowled if overs_bowled > 0 else 0.0
@property
def required_rate(self) -> float:
"""Calculate required run rate based on overs remaining."""
runs_remaining = 200 - self.runs_scored # Example target
return runs_remaining / self.overs_remaining if self.overs_remaining > 0 else 0.0
class Match(BaseModel):
"""Immutable match context and state."""
match_id: str
match_type: str # "T20", "ODI", "Test"
batting_team: str
bowling_team: str
current_innings: InningsState
target_score: int = Field(default=180)
timestamp: datetime = Field(default_factory=datetime.now)
class Config:
frozen = True
class AgentDecision(BaseModel):
"""Decision output from any agent with confidence scoring."""
agent_name: str
recommendation: str
confidence_score: float = Field(ge=0.0, le=1.0)
reasoning: str
decision_type: str # "batting", "bowling", "fielding"
class Config:
frozen = True
class BaseAgent(ABC):
"""Abstract base class for all specialist agents."""
def __init__(self, agent_name: str):
self.agent_name = agent_name
self.decision_history: List[AgentDecision] = []
async def execute(self, match_state: Match) -> AgentDecision:
"""Core agent loop: observe → reason → decide."""
# Step 1: Observe (extract relevant match state)
observations = self.observe(match_state)
# Step 2: Reason (process observations)
reasoning_output = self.reason(observations, match_state)
# Step 3: Act (generate decision with confidence)
decision = self.act(reasoning_output, match_state)
# Store decision in history for audit
self.decision_history.append(decision)
return decision
@abstractmethod
def observe(self, match_state: Match) -> dict:
"""Extract relevant observations from match state."""
pass
@abstractmethod
def reason(self, observations: dict, match_state: Match) -> dict:
"""Process observations and generate reasoning."""
pass
@abstractmethod
def act(self, reasoning_output: dict, match_state: Match) -> AgentDecision:
"""Generate decision with confidence score."""
pass
print(f"✓ Foundation models and BaseAgent defined")
print(f"✓ Data structures: Match, InningsState, BowlerStats, AgentDecision")
print(f"✓ BaseAgent abstract class with observe→reason→act loop")Step 2 — Core Logic
Step 2 implements three specialist agents, each of which independently analyzes a specific dimension of the match. The `BattingStrategistAgent` observes the current run rate versus the required rate, wickets remaining, and match phase — such as powerplay versus death overs — then reasons about acceleration opportunities and appropriate levels of risk-taking, producing decisions like "Aggressive batting in powerplay" with a confidence score informed by the quality of the opposition bowlers.
The `BowlingAnalystAgent` monitors the current economy rate, bowler fitness patterns based on overs in a spell, and opposition batter weaknesses. It reasons about which bowlers to deploy and which delivery types maximize the probability of taking a wicket, generating recommendations grounded in those observations.
The `FieldPlacementAgent` observes batter handedness, favorite scoring areas, and the broader match situation. It reasons about optimal fielding positions that simultaneously restrict runs and create wicket opportunities. Each of these three agents operates asynchronously with independent observation-reasoning-action cycles, and their confidence scores reflect data quality — yielding high confidence when patterns are clear, such as when a batsman averages 40 or more against spin, and lower confidence when the available data is sparse or contradictory.
# agents/batting_strategist.py
import asyncio
from models.match_state import BaseAgent, Match, AgentDecision, InningsState
class BattingStrategistAgent(BaseAgent):
"""Analyzes batting strategy based on run rate, wickets, and match phase."""
def __init__(self):
super().__init__(agent_name="BattingStrategistAgent")
def observe(self, match_state: Match) -> dict:
"""Extract batting-relevant observations."""
innings = match_state.current_innings
return {
"current_run_rate": innings.run_rate,
"required_run_rate": innings.required_rate,
"wickets_remaining": 10 - innings.wickets_lost,
"overs_remaining": innings.overs_remaining,
"powerplay_active": innings.powerplay_active,
"current_batter": innings.current_batter,
"runs_scored": innings.runs_scored,
}
def reason(self, observations: dict, match_state: Match) -> dict:
"""Analyze batting situation and determine strategy."""
run_rate_deficit = observations["required_run_rate"] - observations["current_run_rate"]
wickets_remaining = observations["wickets_remaining"]
overs_remaining = observations["overs_remaining"]
# Decision logic based on match phase and situation
if observations["powerplay_active"]:
if run_rate_deficit < 1.0:
strategy = "Conservative batting - consolidate position"
confidence = 0.85
else:
strategy = "Aggressive batting - capitalize powerplay"
confidence = 0.80
else: # Death overs
if run_rate_deficit > 2.0 and wickets_remaining >= 3:
strategy = "Aggressive acceleration - high-risk shots"
confidence = 0.75
elif run_rate_deficit > 2.0 and wickets_remaining < 3:
strategy = "Cautious approach - preserve wickets"
confidence = 0.80
else:
strategy = "Balanced batting - maintain run rate"
confidence = 0.82
return {
"strategy": strategy,
"confidence": confidence,
"run_rate_gap": run_rate_deficit,
"wickets_in_hand": wickets_remaining,
}
def act(self, reasoning_output: dict, match_state: Match) -> AgentDecision:
"""Generate batting decision."""
reasoning_text = f"Run rate deficit: {reasoning_output['run_rate_gap']:.2f} overs. Wickets in hand: {reasoning_output['wickets_in_hand']}. Current batter: {match_state.current_innings.current_batter}."
return AgentDecision(
agent_name=self.agent_name,
recommendation=reasoning_output["strategy"],
confidence_score=reasoning_output["confidence"],
reasoning=reasoning_text,
decision_type="batting"
)
# agents/bowling_analyst.py
from typing import Dict
class BowlingAnalystAgent(BaseAgent):
"""Analyzes bowling strategy and bowler deployment."""
def __init__(self):
super().__init__(agent_name="BowlingAnalystAgent")
def observe(self, match_state: Match) -> dict:
"""Extract bowling-relevant observations."""
innings = match_state.current_innings
bowlers = innings.bowlers_in_spell
return {
"bowlers_in_spell": [(b.bowler_name, b.economy_rate, b.overs_bowled) for b in bowlers],
"opposition_batter": match_state.current_innings.current_batter,
"overs_remaining": innings.overs_remaining,
"runs_conceded": innings.runs_scored,
"phase": "powerplay" if innings.powerplay_active else "death",
}
def reason(self, observations: dict, match_state: Match) -> dict:
"""Analyze bowling situation and recommend changes."""
bowlers = observations["bowlers_in_spell"]
overs_remaining = observations["overs_remaining"]
# Find best performing bowler (lowest economy)
best_bowler = min(bowlers, key=lambda x: x[1])
worst_bowler = max(bowlers, key=lambda x: x[1])
if observations["phase"] == "death":
if worst_bowler[1] > 12.0: # Economy > 12
recommendation = f"Replace {worst_bowler[0]} - economy {worst_bowler[1]:.2f} too high"
confidence = 0.88
else:
recommendation = f"Continue with {best_bowler[0]} - best economy {best_bowler[1]:.2f}"
confidence = 0.80
else: # Powerplay
if best_bowler[2] < 2.0: # Less than 2 overs bowled
recommendation = f"Deploy {best_bowler[0]} - fresh bowler, economy {best_bowler[1]:.2f}"
confidence = 0.85
else:
recommendation = "Rotate bowlers to maintain pressure"
confidence = 0.75
return {
"recommendation": recommendation,
"confidence": confidence,
"best_bowler": best_bowler[0],
"worst_bowler": worst_bowler[0],
}
def act(self, reasoning_output: dict, match_state: Match) -> AgentDecision:
"""Generate bowling decision."""
reasoning_text = f"Best bowler: {reasoning_output['best_bowler']}. Worst performer: {reasoning_output['worst_bowler']}. Opposition: {match_state.current_innings.current_batter}."
return AgentDecision(
agent_name=self.agent_name,
recommendation=reasoning_output["recommendation"],
confidence_score=reasoning_output["confidence"],
reasoning=reasoning_text,
decision_type="bowling"
)
# agents/field_placement.py
class FieldPlacementAgent(BaseAgent):
"""Optimizes fielding positions based on batter patterns and match situation."""
def __init__(self):
super().__init__(agent_name="FieldPlacementAgent")
def observe(self, match_state: Match) -> dict:
"""Extract fielding-relevant observations."""
innings = match_state.current_innings
return {
"current_batter": innings.current_batter,
"runs_conceded": innings.runs_scored,
"wickets_lost": innings.wickets_lost,
"balls_faced": innings.balls_faced,
"phase": "powerplay" if innings.powerplay_active else "middle/death",
}
def reason(self, observations: dict, match_state: Match) -> dict:
"""Determine optimal field placement."""
batter = observations["current_batter"]
phase = observations["phase"]
# Simplified field strategy based on batter and phase
batter_profiles = {
"Rohit Sharma": {"weakness": "yorkers", "field": "leg-side heavy", "conf": 0.82},
"Virat Kohli": {"weakness": "short balls", "field": "short fine-leg, deep midwicket", "conf": 0.85},
"Travis Head": {"weakness": "off-stump line", "field": "short third-man, cover", "conf": 0.80},
"Steve Smith": {"weakness": "leg-side trap", "field": "short leg, deep square leg", "conf": 0.78},
}
batter_info = batter_profiles.get(batter, {"weakness": "variable", "field": "standard", "conf": 0.60})
if phase == "powerplay":
field_setup = f"Attacking field: {batter_info['field']} to exploit {batter_info['weakness']}"
else:
field_setup = f"Death field: boundaries covered, {batter_info['field']}"
return {
"field_setup": field_setup,
"confidence": batter_info["conf"],
"batter_weakness": batter_info["weakness"],
}
def act(self, reasoning_output: dict, match_state: Match) -> AgentDecision:
"""Generate fielding decision."""
reasoning_text = f"Targeting batter weakness: {reasoning_output['batter_weakness']}. Batter: {match_state.current_innings.current_batter}."
return AgentDecision(
agent_name=self.agent_name,
recommendation=reasoning_output["field_setup"],
confidence_score=reasoning_output["confidence"],
reasoning=reasoning_text,
decision_type="fielding"
)
print(f"✓ BattingStrategistAgent implemented")
print(f"✓ BowlingAnalystAgent implemented")
print(f"✓ FieldPlacementAgent implemented")Step 3 — Integration & Enhancement
Step 3 implements the Primary Agent, which serves as the orchestrator of the entire workflow. The `Orchestrator` class uses Python's `asyncio.gather()` to execute all three specialist agents concurrently, capturing their independent recommendations along with the associated confidence scores. It then aggregates these outputs, resolves any conflicts through confidence-weighted voting combined with domain-specific heuristics, and synthesizes a unified match strategy.
Conflict resolution becomes necessary when recommendations contradict one another — for example, when the `BattingStrategistAgent` recommends aggressive batting with a confidence of 0.80 while the `BowlingAnalystAgent` implicitly suggests a slower run rate through defensive field placement. The resolver applies a weighted voting system in which higher confidence scores carry greater influence, while domain-specific rules serve as hard overrides in critical situations. For instance, if only two wickets remain, a defensive strategy will always take precedence over aggressive recommendations regardless of their confidence levels.
The synthesizer generates the final decision narrative, explaining which recommendations were accepted, which were deprioritized, and how the integrated strategy balances all three dimensions — batting, bowling, and fielding. This architecture mirrors real match captaincy: gathering specialist input, weighting it by expertise and certainty, applying domain knowledge to resolve conflicts, and communicating a coherent strategy to the team.
# orchestrator/coordinator.py
import asyncio
from typing import List, Dict
from models.match_state import Match, AgentDecision, BaseAgent
class ConflictResolver:
"""Resolves conflicts between agent recommendations using weighted voting and domain rules."""
@staticmethod
def resolve_conflicts(decisions: List[AgentDecision], match_state: Match) -> Dict:
"""
Apply weighted voting and domain-specific rules to resolve conflicting recommendations.
Higher confidence = higher weight in final decision.
Domain rules override if match situation is critical (e.g., few wickets remaining).
"""
# Group decisions by type for conflict detection
batting_decisions = [d for d in decisions if d.decision_type == "batting"]
bowling_decisions = [d for d in decisions if d.decision_type == "bowling"]
fielding_decisions = [d for d in decisions if d.decision_type == "fielding"]
# Domain-specific rule: if wickets < 3, always prioritize conservative strategy
wickets_remaining = 10 - match_state.current_innings.wickets_lost
if wickets_remaining < 3:
# Override to conservative (defensive) approach
return {
"strategy_override": "CRITICAL_SITUATION",
"reason": f"Only {wickets_remaining} wickets remaining - default to conservative strategy",
"batting_approach": "Consolidate, avoid risky shots",
"bowling_approach": "Maintain defensive fields, prioritize wicket-taking",
}
# Standard weighted voting: sort by confidence and select highest confidence
best_batting = max(batting_decisions, key=lambda d: d.confidence_score) if batting_decisions else None
best_bowling = max(bowling_decisions, key=lambda d: d.confidence_score) if bowling_decisions else None
best_fielding = max(fielding_decisions, key=lambda d: d.confidence_score) if fielding_decisions else None
return {
"batting_decision": best_batting,
"bowling_decision": best_bowling,
"fielding_decision": best_fielding,
"resolution_method": "Confidence-weighted voting",
}
class MatchOrchestrator:
"""Primary Agent: orchestrates specialist agents and synthesizes unified match strategy."""
def __init__(self, batting_agent: BaseAgent, bowling_agent: BaseAgent, fielding_agent: BaseAgent):
self.batting_agent = batting_agent
self.bowling_agent = bowling_agent
self.fielding_agent = fielding_agent
self.conflict_resolver = ConflictResolver()
self.execution_history = []
async def orchestrate(self, match_state: Match) -> Dict:
"""
Execute all specialist agents concurrently and synthesize decisions.
"""
print(f"\n[ORCHESTRATOR] Analyzing match: {match_state.match_id}")
print(f"[ORCHESTRATOR] Executing agents concurrently...")
# Execute all agents in parallel
try:
batting_decision, bowling_decision, fielding_decision = await asyncio.gather(
self.batting_agent.execute(match_state),
self.bowling_agent.execute(match_state),
self.fielding_agent.execute(match_state),
return_exceptions=True
)
except Exception as e:
print(f"[ERROR] Agent execution failed: {e}")
raise
# Collect all decisions
all_decisions = [
batting_decision, bowling_decision, fielding_decision
]
print(f"[ORCHESTRATOR] Agent decisions collected:")
for decision in all_decisions:
print(f" - {decision.agent_name}: {decision.recommendation} (conf: {decision.confidence_score:.2f})")
# Resolve conflicts
resolved_strategy = self.conflict_resolver.resolve_conflicts(all_decisions, match_state)
# Synthesize final strategy
final_strategy = self._synthesize_strategy(all_decisions, resolved_strategy, match_state)
# Store in history
self.execution_history.append({
"match_id": match_state.match_id,
"timestamp": match_state.timestamp,
"individual_decisions": all_decisions,
"resolved_strategy": resolved_strategy,
"final_strategy": final_strategy,
})
return final_strategy
def _synthesize_strategy(self, decisions: List[AgentDecision], resolved: Dict, match_state: Match) -> Dict:
"""
Synthesize a unified match strategy from specialist recommendations and conflict resolution.
"""
if "strategy_override" in resolved:
# Critical situation override
return {
"status": "CRITICAL_DECISION",
"strategy": resolved["reason"],
"batting_directive": resolved["batting_approach"],
"bowling_directive": resolved["bowling_approach"],
"fielding_directive": "Boundary protection + wicket-taking setup",
"confidence": 0.95, # High confidence for critical situations
"explanation": f"Match situation critical: {resolved['reason']}. Overriding standard analysis.",
}
else:
# Standard synthesis using resolved decisions
batting_dec = resolved["batting_decision"]
bowling_dec = resolved["bowling_decision"]
fielding_dec = resolved["fielding_decision"]
avg_confidence = (batting_dec.confidence_score + bowling_dec.confidence_score + fielding_dec.confidence_score) / 3
return {
"status": "NORMAL_OPERATION",
"strategy": "Integrated Match Strategy",
"batting_directive": batting_dec.recommendation,
"bowling_directive": bowling_dec.recommendation,
"fielding_directive": fielding_dec.recommendation,
"confidence": avg_confidence,
"explanation": (
f"Batting: {batting_dec.recommendation} (conf: {batting_dec.confidence_score:.2f}). "
f"Bowling: {bowling_dec.recommendation} (conf: {bowling_dec.confidence_score:.2f}). "
f"Fielding: {fielding_dec.recommendation} (conf: {fielding_dec.confidence_score:.2f}). "
f"Overall strategy confidence: {avg_confidence:.2f}."
),
}
print(f"✓ ConflictResolver implemented - weighted voting with domain rules")
print(f"✓ MatchOrchestrator implemented - parallel agent execution and synthesis")Step 4 — Testing & Verification
Step 4 demonstrates the complete agentic workflow using realistic cricket match data. You will create a `Match` object representing a mid-innings scenario between India and Australia in a T20 match, initialize all three specialist agents, execute the Orchestrator, and verify that decisions are generated with appropriate confidence scores and properly synthesized into a unified strategy.
The test validates three key behaviors: parallel execution, confirming that all agents respond; conflict resolution logic, confirming that one agent's recommendation prevails based on confidence scoring; and strategy synthesis, confirming that all three dimensions are integrated coherently into the final output. The expected output displays each agent's individual recommendation alongside its confidence score, followed by the final orchestrated strategy.
#!/bin/bash
# Step 4: Run and verify the complete agentic workflow
echo "========================================"
echo "Cricket Analytics Agent - Full Execution"
echo "========================================"
echo ""
echo "[1] Creating test match scenario: India vs Australia T20"
echo "[2] Match state: India batting, 4/6 overs, 42/2 runs, targeting 165"
echo "[3] Initializing specialist agents (Batting, Bowling, Fielding)"
echo "[4] Executing orchestrator - running agents in parallel..."
echo ""
python3 main.py
echo ""
echo "========================================"
echo "Expected Output Structure:"
echo "========================================"
echo ""
echo "[MATCH STATE]"
echo " Match ID: IND-vs-AUS-T20-2024"
echo " Batting Team: India"
echo " Current Score: 42/2 (6 overs)"
echo " Run Rate: 7.0 | Required Rate: 10.8"
echo ""
echo "[AGENT RECOMMENDATIONS]"
echo " BattingStrategistAgent:"
echo " Recommendation: Aggressive acceleration - high-risk shots"
echo " Confidence: 0.75"
echo " BowlingAnalystAgent:"
echo " Recommendation: Deploy fresh bowler, economy 5.2"
echo " Confidence: 0.88"
echo " FieldPlacementAgent:"
echo " Recommendation: Attacking field for Rohit Sharma weakness (yorkers)"
echo " Confidence: 0.82"
echo ""
echo "[ORCHESTRATION & CONFLICT RESOLUTION]"
echo " Wickets remaining: 8 (not critical)"
echo " Highest confidence agent: BowlingAnalystAgent (0.88)"
echo " Resolution method: Confidence-weighted voting"
echo ""
echo "[FINAL SYNTHESIZED STRATEGY]"
echo " Status: NORMAL_OPERATION"
echo " Overall Confidence: 0.82"
echo " Batting Directive: Aggressive acceleration"
echo " Bowling Directive: Deploy fresh bowler"
echo " Fielding Directive: Attacking field for Rohit"
echo " Explanation: Integrated recommendations from 3 specialist agents"
echo ""
echo "========================================"Warning: Race Condition in Concurrent Agent Execution. If agent observe() methods modify shared match state (even unintentionally), concurrent execution causes inconsistent decisions. Solution: Ensure all data models use Pydantic's `frozen=True` config to make state immutable. All agents must use `.copy()` when processing observations. Never allow agents to write to shared state—only the Orchestrator can update match state between iterations. Additionally, avoid using global variables in agents; pass all context through method parameters.
Extension Challenge: Add a fourth agent called 'RiskAssessmentAgent' that monitors weather conditions, player fatigue levels, and match momentum (win probability). This agent should output a risk score (0-1) indicating how conservative vs aggressive the overall strategy should be. Modify the ConflictResolver to incorporate risk assessment—if risk > 0.7, aggressive recommendations are downweighted. Implement a confidence decay function: if an agent's recommendation differs significantly from its historical track record, reduce confidence by 10%. Add support for multi-match learning: store agent decision histories across matches and use them to improve future confidence scoring.
- Multi-agent orchestration requires immutable shared state and concurrent execution (asyncio.gather) to prevent race conditions and enable parallel reasoning.
- Confidence scoring quantifies agent uncertainty, enabling weighted voting for conflict resolution while maintaining interpretability of final decisions.
- Domain-specific rules (e.g., critical situation overrides) ensure realistic agentic behavior that adapts to contextual constraints not captured purely in data.
- Specialist agents maintain clear separation of concerns—batting, bowling, fielding—allowing independent iteration and scaling without cross-agent dependencies.
- Agent decision synthesis must map recommendations back to reasoning chains, creating auditable decision paths that explain why certain strategies were selected over alternatives.
- Inter-agent communication happens through immutable decision objects (AgentDecision), preventing message tampering and ensuring orchestrator reliability under concurrent access.