"""
Pattern 263: Simulation Pattern with LangChain/LangGraph
Purpose: Environment modeling and scenario simulation with intelligent analysis.
Provides Monte Carlo, discrete event, agent-based simulations with LLM insights.
"""
from typing import Dict, Any, List, Optional, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import random
import statistics
import time
class SimulationType(Enum):
"""Types of simulations."""
MONTE_CARLO = "monte_carlo"
DISCRETE_EVENT = "discrete_event"
AGENT_BASED = "agent_based"
SYSTEM_DYNAMICS = "system_dynamics"
class SimulationStatus(Enum):
"""Simulation execution status."""
INITIALIZED = "initialized"
RUNNING = "running"
ANALYZING = "analyzing"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class SimulationScenario:
"""Simulation scenario configuration."""
scenario_id: str
name: str
simulation_type: SimulationType
initial_state: Dict[str, float]
steps: int = 100
iterations: int = 10
parameters: Dict[str, Any] = field(default_factory=dict)
@dataclass
class SimulationMetrics:
"""Metrics from simulation run."""
execution_time: float
total_steps: int
total_iterations: int
convergence_score: float
variance: Dict[str, float]
stability_index: float
@dataclass
class SimulationResult:
"""Complete simulation result."""
scenario: SimulationScenario
final_state: Dict[str, float]
state_history: List[Dict[str, float]]
metrics: SimulationMetrics
insights: List[str]
anomalies: List[str]
recommendations: List[str]
class SimulationState(TypedDict):
"""State for simulation workflow."""
scenario_id: str
simulation_type: str
initial_state: Dict[str, float]
steps: int
iterations: int
parameters: Dict[str, Any]
final_state: Dict[str, float]
state_history: List[Dict[str, float]]
metrics: Dict[str, Any]
insights: List[str]
anomalies: List[str]
recommendations: List[str]
class IntelligentSimulator:
"""
Intelligent Simulator with LangGraph workflow integration.
Features:
- Monte Carlo simulation
- Discrete event simulation
- Agent-based modeling
- LLM-powered pattern detection
- Anomaly detection
- Predictive insights
"""
def __init__(self, model_name: str = "gpt-4", temperature: float = 0.5):
self.llm = ChatOpenAI(model=model_name, temperature=temperature)
self.results: Dict[str, SimulationResult] = {}
self.workflow = self._build_workflow()
# Statistics
self.stats = {
"total_simulations": 0,
"simulations_by_type": {t.value: 0 for t in SimulationType},
"total_steps": 0,
"total_iterations": 0,
"avg_execution_time": 0.0
}
def _build_workflow(self) -> StateGraph:
"""Build LangGraph workflow."""
workflow = StateGraph(SimulationState)
workflow.add_node("initialize", self._initialize_node)
workflow.add_node("run_simulation", self._run_simulation_node)
workflow.add_node("calculate_metrics", self._calculate_metrics_node)
workflow.add_node("detect_anomalies", self._detect_anomalies_node)
workflow.add_node("generate_insights", self._generate_insights_node)
workflow.set_entry_point("initialize")
workflow.add_edge("initialize", "run_simulation")
workflow.add_edge("run_simulation", "calculate_metrics")
workflow.add_edge("calculate_metrics", "detect_anomalies")
workflow.add_edge("detect_anomalies", "generate_insights")
workflow.add_edge("generate_insights", END)
return workflow.compile()
def _initialize_node(self, state: SimulationState) -> SimulationState:
"""Initialize simulation state."""
state["final_state"] = {}
state["state_history"] = []
state["metrics"] = {}
state["insights"] = []
state["anomalies"] = []
state["recommendations"] = []
return state
def _run_simulation_node(self, state: SimulationState) -> SimulationState:
"""Run the simulation."""
sim_type = state["simulation_type"]
if sim_type == SimulationType.MONTE_CARLO.value:
result = self._run_monte_carlo(state)
elif sim_type == SimulationType.AGENT_BASED.value:
result = self._run_agent_based(state)
elif sim_type == SimulationType.DISCRETE_EVENT.value:
result = self._run_discrete_event(state)
else:
result = self._run_monte_carlo(state)
state["final_state"] = result["final_state"]
state["state_history"] = result["state_history"]
return state
def _run_monte_carlo(self, state: SimulationState) -> Dict[str, Any]:
"""Run Monte Carlo simulation."""
initial = state["initial_state"]
steps = state["steps"]
iterations = state["iterations"]
all_final_states = []
state_history = []
for _ in range(iterations):
current_state = initial.copy()
iteration_history = [current_state.copy()]
for step in range(steps):
# Apply random variations
for key in current_state:
variation = random.gauss(0, 0.05) # 5% std dev
current_state[key] = max(0, current_state[key] * (1 + variation))
iteration_history.append(current_state.copy())
all_final_states.append(current_state.copy())
if not state_history:
state_history = iteration_history
# Calculate average final state
final_state = {}
for key in initial.keys():
values = [s[key] for s in all_final_states]
final_state[key] = statistics.mean(values)
return {
"final_state": final_state,
"state_history": state_history,
"all_final_states": all_final_states
}
def _run_agent_based(self, state: SimulationState) -> Dict[str, Any]:
"""Run agent-based simulation."""
initial = state["initial_state"]
steps = state["steps"]
num_agents = state["parameters"].get("num_agents", 10)
# Initialize agents
agents = [initial.copy() for _ in range(num_agents)]
state_history = []
for step in range(steps):
# Agent interactions
for i in range(len(agents)):
# Influence from neighbors
if i > 0 and i Dict[str, Any]:
"""Run discrete event simulation."""
current_state = state["initial_state"].copy()
events = state["parameters"].get("events", [])
state_history = [current_state.copy()]
# Sort events by time
sorted_events = sorted(events, key=lambda e: e.get("time", 0))
for event in sorted_events:
# Apply event effects
if "effects" in event:
for key, delta in event["effects"].items():
if key in current_state:
current_state[key] += delta
state_history.append(current_state.copy())
return {
"final_state": current_state,
"state_history": state_history
}
def _calculate_metrics_node(self, state: SimulationState) -> SimulationState:
"""Calculate simulation metrics."""
history = state["state_history"]
if not history:
return state
# Calculate variance
variance = {}
for key in state["initial_state"].keys():
values = [s[key] for s in history if key in s]
if values:
variance[key] = statistics.variance(values) if len(values) > 1 else 0.0
# Calculate convergence (lower variance at end = higher convergence)
if len(history) > 10:
recent_variance = {}
for key in state["initial_state"].keys():
recent_values = [s[key] for s in history[-10:] if key in s]
if recent_values:
recent_variance[key] = statistics.variance(recent_values) if len(recent_values) > 1 else 0.0
avg_variance = statistics.mean(recent_variance.values()) if recent_variance else 0
convergence = max(0, 1 - min(avg_variance, 1.0))
else:
convergence = 0.5
# Stability index
stability = convergence
state["metrics"] = {
"variance": variance,
"convergence_score": convergence,
"stability_index": stability,
"total_steps": len(history),
"total_iterations": state["iterations"]
}
return state
def _detect_anomalies_node(self, state: SimulationState) -> SimulationState:
"""Detect anomalies in simulation results."""
history = state["state_history"]
anomalies = []
if len(history) 0.5: # 50% change
anomalies.append(f"Sudden change in {key} at step {i}: {change_pct*100:.1f}%")
# Detect extreme values
for key in state["initial_state"].keys():
values = [s[key] for s in history if key in s]
if values:
mean_val = statistics.mean(values)
std_val = statistics.stdev(values) if len(values) > 1 else 0
for i, val in enumerate(values):
if std_val > 0 and abs(val - mean_val) > 3 * std_val:
anomalies.append(f"Outlier in {key} at step {i}: {val:.2f}")
state["anomalies"] = anomalies[:5] # Limit to 5
return state
def _generate_insights_node(self, state: SimulationState) -> SimulationState:
"""Generate AI insights from simulation."""
final_state = state["final_state"]
initial_state = state["initial_state"]
metrics = state["metrics"]
anomalies = state["anomalies"]
# Use LLM for insights
prompt = ChatPromptTemplate.from_template(
"Analyze this simulation result:\n"
"Type: {sim_type}\n"
"Initial State: {initial}\n"
"Final State: {final}\n"
"Convergence: {convergence:.2f}\n"
"Anomalies: {anomalies}\n\n"
"Provide 3 key insights about the simulation behavior."
)
chain = prompt | self.llm
response = chain.invoke({
"sim_type": state["simulation_type"],
"initial": str(initial_state),
"final": str(final_state),
"convergence": metrics.get("convergence_score", 0),
"anomalies": str(anomalies)
})
insights_text = str(response.content)
insights = [insights_text[:200]]
# Generate recommendations
recommendations = []
if metrics.get("convergence_score", 0) 3:
recommendations.append("High number of anomalies detected - review simulation parameters")
for key, var in metrics.get("variance", {}).items():
if var > 1.0:
recommendations.append(f"High variance in {key} - consider stabilization measures")
state["insights"] = insights
state["recommendations"] = recommendations
return state
def simulate(self, scenario: SimulationScenario) -> SimulationResult:
"""Run simulation with workflow."""
start_time = time.time()
initial_state: SimulationState = {
"scenario_id": scenario.scenario_id,
"simulation_type": scenario.simulation_type.value,
"initial_state": scenario.initial_state,
"steps": scenario.steps,
"iterations": scenario.iterations,
"parameters": scenario.parameters,
"final_state": {},
"state_history": [],
"metrics": {},
"insights": [],
"anomalies": [],
"recommendations": []
}
# Run workflow
result_state = self.workflow.invoke(initial_state)
execution_time = time.time() - start_time
# Create result
metrics = SimulationMetrics(
execution_time=execution_time,
total_steps=result_state["metrics"].get("total_steps", 0),
total_iterations=result_state["metrics"].get("total_iterations", 0),
convergence_score=result_state["metrics"].get("convergence_score", 0.0),
variance=result_state["metrics"].get("variance", {}),
stability_index=result_state["metrics"].get("stability_index", 0.0)
)
result = SimulationResult(
scenario=scenario,
final_state=result_state["final_state"],
state_history=result_state["state_history"],
metrics=metrics,
insights=result_state["insights"],
anomalies=result_state["anomalies"],
recommendations=result_state["recommendations"]
)
self.results[scenario.scenario_id] = result
# Update stats
self.stats["total_simulations"] += 1
self.stats["simulations_by_type"][scenario.simulation_type.value] += 1
self.stats["total_steps"] += scenario.steps
self.stats["total_iterations"] += scenario.iterations
if self.stats["total_simulations"] > 0:
total_time = self.stats["avg_execution_time"] * (self.stats["total_simulations"] - 1) + execution_time
self.stats["avg_execution_time"] = total_time / self.stats["total_simulations"]
return result
def get_statistics(self) -> Dict[str, Any]:
"""Get simulation statistics."""
return {
**self.stats,
"total_results_stored": len(self.results)
}
class SimulationPattern:
"""
Comprehensive Simulation Pattern coordinator with LangChain/LangGraph.
"""
def __init__(self, model_name: str = "gpt-4"):
self.simulator = IntelligentSimulator(model_name=model_name)
def create_scenario(
self,
scenario_id: str,
name: str,
simulation_type: SimulationType,
initial_state: Dict[str, float],
steps: int = 100,
iterations: int = 10,
parameters: Optional[Dict[str, Any]] = None
) -> SimulationScenario:
"""Create simulation scenario."""
return SimulationScenario(
scenario_id=scenario_id,
name=name,
simulation_type=simulation_type,
initial_state=initial_state,
steps=steps,
iterations=iterations,
parameters=parameters or {}
)
def run_simulation(self, scenario: SimulationScenario) -> SimulationResult:
"""Run simulation."""
return self.simulator.simulate(scenario)
def get_statistics(self) -> Dict[str, Any]:
"""Get statistics."""
return self.simulator.get_statistics()
if __name__ == "__main__":
print("=== Simulation Pattern with LangChain/LangGraph Demo ===\n")
print("Scenario 1: Monte Carlo Simulation")
print("-" * 50)
pattern = SimulationPattern(model_name="gpt-4")
# Create Monte Carlo scenario
mc_scenario = pattern.create_scenario(
scenario_id="mc-001",
name="Market Price Simulation",
simulation_type=SimulationType.MONTE_CARLO,
initial_state={"price": 100.0, "volume": 1000.0, "volatility": 0.2},
steps=50,
iterations=100
)
print(f"Running {mc_scenario.name}...")
result1 = pattern.run_simulation(mc_scenario)
print(f"\nFinal State:")
for key, value in result1.final_state.items():
print(f" {key}: {value:.2f}")
print(f"\nMetrics:")
print(f" Execution Time: {result1.metrics.execution_time:.3f}s")
print(f" Convergence Score: {result1.metrics.convergence_score:.3f}")
print(f" Stability Index: {result1.metrics.stability_index:.3f}")
if result1.anomalies:
print(f"\nAnomalies Detected: {len(result1.anomalies)}")
for anomaly in result1.anomalies[:3]:
print(f" - {anomaly}")
if result1.insights:
print(f"\nAI Insights:")
for insight in result1.insights:
print(f" {insight}")
if result1.recommendations:
print(f"\nRecommendations:")
for rec in result1.recommendations:
print(f" - {rec}")
print("\n" + "=" * 50)
print("\nScenario 2: Agent-Based Simulation")
print("-" * 50)
ab_scenario = pattern.create_scenario(
scenario_id="ab-001",
name="Opinion Dynamics",
simulation_type=SimulationType.AGENT_BASED,
initial_state={"opinion": 0.5, "confidence": 0.8},
steps=30,
iterations=1,
parameters={"num_agents": 20}
)
print(f"Running {ab_scenario.name}...")
result2 = pattern.run_simulation(ab_scenario)
print(f"\nFinal State:")
for key, value in result2.final_state.items():
print(f" {key}: {value:.2f}")
print(f"\nMetrics:")
print(f" Execution Time: {result2.metrics.execution_time:.3f}s")
print(f" Total Steps: {result2.metrics.total_steps}")
print("\n" + "=" * 50)
print("\nScenario 3: Discrete Event Simulation")
print("-" * 50)
events = [
{"time": 5, "effects": {"queue_length": 10}},
{"time": 10, "effects": {"queue_length": -5}},
{"time": 15, "effects": {"queue_length": 8}},
{"time": 20, "effects": {"queue_length": -12}}
]
de_scenario = pattern.create_scenario(
scenario_id="de-001",
name="Queue Management",
simulation_type=SimulationType.DISCRETE_EVENT,
initial_state={"queue_length": 0.0, "throughput": 100.0},
steps=25,
iterations=1,
parameters={"events": events}
)
print(f"Running {de_scenario.name}...")
result3 = pattern.run_simulation(de_scenario)
print(f"\nFinal State:")
for key, value in result3.final_state.items():
print(f" {key}: {value:.2f}")
print("\n" + "=" * 50)
print("\nStatistics")
print("-" * 50)
stats = pattern.get_statistics()
print(f"Total Simulations: {stats['total_simulations']}")
print(f"Total Steps Simulated: {stats['total_steps']}")
print(f"Total Iterations: {stats['total_iterations']}")
print(f"Average Execution Time: {stats['avg_execution_time']:.3f}s")
print(f"\nSimulations by Type:")
for sim_type, count in stats['simulations_by_type'].items():
if count > 0:
print(f" {sim_type}: {count}")
print("\n" + "=" * 50)
print("Simulation pattern demonstration complete!")
print("=" * 50)
Agentic Architecture Pattern: Simulation
Pattern 263: Simulation Pattern – Line-by-Line Explanation
Overview
This pattern implements environment modeling and scenario simulation with intelligent analysis using LangChain/LangGraph. It supports Monte Carlo, discrete event, and agent-based simulations with LLM-powered insights.
Code Explanation
Lines 1-18: Imports and Setup
"""
Pattern 263: Simulation Pattern with LangChain/LangGraph
Purpose: Environment modeling and scenario simulation with intelligent analysis.
Provides Monte Carlo, discrete event, agent-based simulations with LLM insights.
"""
Explanation: Module docstring describing the pattern’s purpose – providing comprehensive simulation capabilities with AI-powered analysis.
from typing import Dict, Any, List, Optional, TypedDict
Explanation: Import type hints for better code documentation and IDE support. TypedDict is crucial for LangGraph state definitions.
from dataclasses import dataclass, field
Explanation: Import dataclass decorators for creating structured data classes with automatic __init__, __repr__, etc.
from datetime import datetime
Explanation: Import datetime for timestamping simulation results.
Explanation: Import Enum for creating type-safe enumeration classes.
from langchain_core.prompts import ChatPromptTemplate
Explanation: Import LangChain’s prompt template system for structured LLM interactions.
from langchain_openai import ChatOpenAI
Explanation: Import OpenAI chat model integration from LangChain.
from langgraph.graph import StateGraph, END
Explanation: Import LangGraph’s state graph workflow system. StateGraph creates workflows, END marks terminal nodes.
import random
import numpy as np
Explanation: Import libraries for random number generation and numerical operations (Monte Carlo simulations).
Lines 20-36: Enums Definition
class SimulationType(Enum):
"""Types of simulations."""
MONTE_CARLO = "monte_carlo"
DISCRETE_EVENT = "discrete_event"
AGENT_BASED = "agent_based"
SYSTEM_DYNAMICS = "system_dynamics"
Explanation: Enum defining four simulation types:
– MONTE_CARLO: Random sampling-based simulations
– DISCRETE_EVENT: Event-driven state transitions
– AGENT_BASED: Multi-agent interaction modeling
– SYSTEM_DYNAMICS: Continuous system behavior modeling
class SimulationStatus(Enum):
"""Simulation execution status."""
INITIALIZED = "initialized"
RUNNING = "running"
ANALYZING = "analyzing"
COMPLETED = "completed"
FAILED = "failed"
Explanation: Enum tracking simulation lifecycle states from initialization through completion or failure.
Lines 38-72: Dataclasses
@dataclass
class SimulationScenario:
"""Simulation scenario configuration."""
scenario_id: str
name: str
simulation_type: SimulationType
initial_state: Dict[str, float]
steps: int = 100
iterations: int = 10
parameters: Dict[str, Any] = field(default_factory=dict)
Explanation: Dataclass defining simulation scenario configuration:
– scenario_id: Unique identifier
– name: Human-readable name
– simulation_type: Type from SimulationType enum
– initial_state: Starting values (e.g., {“price”: 100.0})
– steps: Number of simulation steps per iteration
– iterations: How many times to run the simulation
– parameters: Additional configuration (default empty dict)
@dataclass
class SimulationMetrics:
"""Metrics from simulation run."""
execution_time: float
total_steps: int
total_iterations: int
convergence_score: float
variance: Dict[str, float]
stability_index: float
Explanation: Dataclass storing simulation performance metrics:
– execution_time: How long simulation took (seconds)
– convergence_score: 0.0-1.0 score indicating if simulation converged
– variance: Variability of each state variable
– stability_index: Overall stability measure
@dataclass
class SimulationResult:
"""Complete simulation result."""
scenario: SimulationScenario
final_state: Dict[str, float]
state_history: List[Dict[str, float]]
metrics: SimulationMetrics
insights: List[str]
anomalies: List[str]
recommendations: List[str]
Explanation: Complete result package containing scenario, final state, full history, metrics, and LLM-generated insights.
Lines 74-87: TypedDict State Definition
class SimulationState(TypedDict):
"""State for simulation workflow."""
scenario_id: str
simulation_type: str
initial_state: Dict[str, float]
steps: int
iterations: int
parameters: Dict[str, Any]
final_state: Dict[str, float]
state_history: List[Dict[str, float]]
metrics: Dict[str, Any]
insights: List[str]
anomalies: List[str]
recommendations: List[str]
Explanation: TypedDict defining the structure of state passed between workflow nodes. This is required by LangGraph to type-check workflow state transitions. Contains all information needed throughout the simulation workflow.
Lines 89-116: IntelligentSimulator Class Initialization
class IntelligentSimulator:
"""
Intelligent Simulator with LangGraph workflow integration.
Features:
- Monte Carlo simulation
- Discrete event simulation
- Agent-based modeling
- LLM-powered pattern detection
- Anomaly detection
- Predictive insights
"""
Explanation: Main simulator class docstring listing all capabilities.
def __init__(self, model_name: str = "gpt-4", temperature: float = 0.5):
self.llm = ChatOpenAI(model=model_name, temperature=temperature)
self.results: Dict[str, SimulationResult] = {}
self.workflow = self._build_workflow()
Explanation: Initialize simulator with:
– LLM for generating insights (temperature 0.5 balances creativity and consistency)
– Results dictionary to store simulation outputs
– Compiled workflow graph
# Statistics
self.stats = {
"total_simulations": 0,
"simulations_by_type": {t.value: 0 for t in SimulationType},
"total_steps": 0,
"total_iterations": 0,
"avg_execution_time": 0.0
}
Explanation: Initialize statistics tracking dictionary with counters for each simulation type and aggregate metrics.
Lines 118-136: Workflow Building
def _build_workflow(self) -> StateGraph:
"""Build LangGraph workflow."""
workflow = StateGraph(SimulationState)
Explanation: Create a new StateGraph with SimulationState as the state type. This enforces type safety.
workflow.add_node("initialize", self._initialize_node)
workflow.add_node("run_simulation", self._run_simulation_node)
workflow.add_node("calculate_metrics", self._calculate_metrics_node)
workflow.add_node("detect_anomalies", self._detect_anomalies_node)
workflow.add_node("generate_insights", self._generate_insights_node)
Explanation: Add 5 nodes to the workflow graph, each pointing to a method that processes the state.
workflow.set_entry_point("initialize")
workflow.add_edge("initialize", "run_simulation")
workflow.add_edge("run_simulation", "calculate_metrics")
workflow.add_edge("calculate_metrics", "detect_anomalies")
workflow.add_edge("detect_anomalies", "generate_insights")
workflow.add_edge("generate_insights", END)
Explanation: Define workflow execution order:
1. Start at “initialize”
2. Flow through run → metrics → anomalies → insights
3. End after insights generation
return workflow.compile()
Explanation: Compile the workflow into an executable graph. This validates the graph structure.
Lines 138-146: Initialize Node
def _initialize_node(self, state: SimulationState) -> SimulationState:
"""Initialize simulation state."""
state["final_state"] = {}
state["state_history"] = []
state["metrics"] = {}
state["insights"] = []
state["anomalies"] = []
state["recommendations"] = []
return state
Explanation: First workflow node – initializes empty containers for results. Takes state dict, modifies it, returns modified state. This pattern is standard for all nodes.
Lines 148-162: Run Simulation Node (Dispatcher)
def _run_simulation_node(self, state: SimulationState) -> SimulationState:
"""Run the simulation."""
sim_type = state["simulation_type"]
if sim_type == "monte_carlo":
result = self._run_monte_carlo(state)
elif sim_type == "agent_based":
result = self._run_agent_based(state)
elif sim_type == "discrete_event":
result = self._run_discrete_event(state)
else:
result = {"final_state": state["initial_state"], "history": []}
Explanation: Dispatcher node that routes to appropriate simulation method based on type. Returns simulation result dict.
state["final_state"] = result["final_state"]
state["state_history"] = result["history"]
return state
Explanation: Update state with simulation results and return.
Lines 164-198: Monte Carlo Simulation
def _run_monte_carlo(self, state: SimulationState) -> Dict[str, Any]:
"""Run Monte Carlo simulation."""
initial = state["initial_state"]
steps = state["steps"]
iterations = state["iterations"]
Explanation: Extract parameters from state.
all_histories = []
for _ in range(iterations):
current = initial.copy()
history = [current.copy()]
Explanation: Run multiple iterations, each starting from initial state. Track history for each iteration.
for _ in range(steps):
new_state = {}
for key, value in current.items():
# Random walk with small variance
change = random.gauss(0, value * 0.02)
new_state[key] = max(0, value + change)
current = new_state
history.append(current.copy())
Explanation: For each step, apply random changes to each state variable using Gaussian distribution (mean=0, std=2% of current value). Ensure non-negative values.
all_histories.append(history)
Explanation: Store this iteration’s history.
# Average across all iterations
final_state = {}
for key in initial.keys():
values = [h[-1][key] for h in all_histories]
final_state[key] = sum(values) / len(values)
Explanation: Calculate final state as average of all iterations’ final values.
# Use history from first iteration for display
return {
"final_state": final_state,
"history": all_histories[0]
}
Explanation: Return averaged final state and first iteration’s history for visualization.
Lines 200-230: Agent-Based Simulation
def _run_agent_based(self, state: SimulationState) -> Dict[str, Any]:
"""Run agent-based simulation."""
initial = state["initial_state"]
steps = state["steps"]
num_agents = state["parameters"].get("num_agents", 10)
Explanation: Extract parameters, including number of agents from parameters dict (default 10).
# Initialize agents
agents = [{key: value for key, value in initial.items()} for _ in range(num_agents)]
Explanation: Create list of agents, each with a copy of initial state.
history = [initial.copy()]
for _ in range(steps):
# Agents interact and influence each other
for i, agent in enumerate(agents):
for key in agent.keys():
# Average with neighbors
neighbors = [agents[j][key] for j in range(num_agents) if j != i]
if neighbors:
influence = sum(neighbors) / len(neighbors)
agent[key] = 0.7 * agent[key] + 0.3 * influence
Explanation: Each step, each agent updates its state by averaging with all other agents (70% own state, 30% average of others). This simulates social influence or opinion dynamics.
# Aggregate state
aggregate = {}
for key in initial.keys():
aggregate[key] = sum(a[key] for a in agents) / num_agents
history.append(aggregate.copy())
Explanation: Calculate aggregate state across all agents and add to history.
return {
"final_state": history[-1],
"history": history
}
Explanation: Return final aggregate state and full history.
Lines 232-253: Discrete Event Simulation
def _run_discrete_event(self, state: SimulationState) -> Dict[str, Any]:
"""Run discrete event simulation."""
current = state["initial_state"].copy()
steps = state["steps"]
events = state["parameters"].get("events", [])
Explanation: Extract events list from parameters. Events are predefined changes at specific times.
history = [current.copy()]
for step in range(steps):
# Apply events at this step
for event in events:
if event.get("time") == step:
effects = event.get("effects", {})
for key, value in effects.items():
if key in current:
current[key] = max(0, current[key] + value)
Explanation: For each step, check if any events occur. If so, apply their effects by modifying state values.
history.append(current.copy())
Explanation: Record state after events applied.
return {
"final_state": current,
"history": history
}
Explanation: Return final state and complete history.
Lines 255-293: Calculate Metrics Node
def _calculate_metrics_node(self, state: SimulationState) -> SimulationState:
"""Calculate simulation metrics."""
history = state["state_history"]
if not history:
return state
Explanation: Extract history, return early if empty.
# Calculate variance for each variable
variance = {}
for key in state["initial_state"].keys():
values = [h[key] for h in history if key in h]
if len(values) > 1:
mean_val = sum(values) / len(values)
variance[key] = sum((v - mean_val) ** 2 for v in values) / len(values)
Explanation: Calculate variance for each state variable across all history steps. Variance measures how much values fluctuate.
# Convergence score (how stable is final state)
if len(history) >= 10:
recent = history[-10:]
recent_variance = {}
for key in state["initial_state"].keys():
values = [h[key] for h in recent if key in h]
if len(values) > 1:
mean_val = sum(values) / len(values)
recent_variance[key] = sum((v - mean_val) ** 2 for v in values) / len(values)
# Lower recent variance = higher convergence
avg_recent_var = sum(recent_variance.values()) / len(recent_variance) if recent_variance else 1.0
convergence_score = 1.0 / (1.0 + avg_recent_var)
else:
convergence_score = 0.5
Explanation: Calculate convergence score (0.0-1.0) by looking at last 10 steps. Lower variance in recent history = higher convergence (system is stabilizing).
# Stability index
avg_variance = sum(variance.values()) / len(variance) if variance else 0.0
stability_index = 1.0 / (1.0 + avg_variance)
Explanation: Calculate overall stability as inverse of average variance. Higher stability = lower variance.
state["metrics"] = {
"variance": variance,
"convergence_score": convergence_score,
"stability_index": stability_index
}
return state
Explanation: Store calculated metrics in state and return.
Lines 295-325: Detect Anomalies Node
def _detect_anomalies_node(self, state: SimulationState) -> SimulationState:
"""Detect anomalies in simulation results."""
history = state["state_history"]
anomalies = []
if len(history) < 2:
state["anomalies"] = anomalies
return state
Explanation: Initialize anomalies list, return early if insufficient history.
# Detect sudden changes
for i in range(1, len(history)):
prev = history[i-1]
curr = history[i]
for key in curr.keys():
if key in prev:
change_pct = abs(curr[key] - prev[key]) / (prev[key] + 1e-10)
if change_pct > 0.5: # 50% change
anomalies.append(
f"Large change in {key} at step {i}: {prev[key]:.2f} → {curr[key]:.2f}"
)
Explanation: Detect anomalies by checking for large percentage changes (>50%) between consecutive steps. The 1e-10 prevents division by zero.
# Detect outliers in final state
for key, value in state["final_state"].items():
initial_value = state["initial_state"].get(key, 0)
if initial_value > 0:
final_ratio = value / initial_value
if final_ratio > 10 or final_ratio < 0.1:
anomalies.append(
f"Extreme final value for {key}: {initial_value:.2f} → {value:.2f}"
)
Explanation: Detect outliers by comparing final values to initial values. Flag if final is >10x or <0.1x initial.
state["anomalies"] = anomalies
return state
Explanation: Store detected anomalies and return state.
Lines 327-373: Generate Insights Node (LLM-Powered)
def _generate_insights_node(self, state: SimulationState) -> SimulationState:
"""Generate insights using LLM."""
prompt = ChatPromptTemplate.from_messages([
("system", "You are an expert in simulation analysis and predictive modeling."),
("user", """Analyze this simulation result:
Simulation Type: {sim_type}
Initial State: {initial}
Final State: {final}
Convergence Score: {convergence}
Stability Index: {stability}
Anomalies: {anomalies}
Provide 3-5 key insights about the simulation behavior, patterns, and recommendations.""")
])
Explanation: Create LLM prompt template with system role and user message containing simulation data.
chain = prompt | self.llm
response = chain.invoke({
"sim_type": state["simulation_type"],
"initial": str(state["initial_state"]),
"final": str(state["final_state"]),
"convergence": state["metrics"].get("convergence_score", 0),
"stability": state["metrics"].get("stability_index", 0),
"anomalies": ", ".join(state["anomalies"]) if state["anomalies"] else "None"
})
Explanation: Create chain (prompt → LLM) and invoke with simulation data. LLM generates natural language insights.
# Parse response into insights
insights_text = response.content
insights = [line.strip() for line in insights_text.split('\n') if line.strip()]
state["insights"] = insights[:5] # Limit to 5 insights
Explanation: Parse LLM response by splitting into lines, removing empty lines, and limiting to 5 insights.
# Generate recommendations
state["recommendations"] = [
"Review anomalies" if state["anomalies"] else "No anomalies detected",
f"Convergence score: {state['metrics'].get('convergence_score', 0):.2f}",
f"System stability: {state['metrics'].get('stability_index', 0):.2f}"
]
return state
Explanation: Generate simple recommendations based on detected anomalies and metrics, then return state.
Lines 375-431: Simulate Method (Main Entry Point)
def simulate(self, scenario: SimulationScenario) -> SimulationResult:
"""Run simulation through workflow."""
start_time = time.time()
Explanation: Main method to run simulation. Start timing.
# Create initial state
initial_state: SimulationState = {
"scenario_id": scenario.scenario_id,
"simulation_type": scenario.simulation_type.value,
"initial_state": scenario.initial_state,
"steps": scenario.steps,
"iterations": scenario.iterations,
"parameters": scenario.parameters,
"final_state": {},
"state_history": [],
"metrics": {},
"insights": [],
"anomalies": [],
"recommendations": []
}
Explanation: Create initial state dict from scenario parameters. This is the starting state for the workflow.
# Run workflow
final_state = self.workflow.invoke(initial_state)
Explanation: Execute the workflow graph. This runs through all nodes: initialize → run → metrics → anomalies → insights.
execution_time = time.time() - start_time
Explanation: Calculate total execution time.
# Create result object
metrics = SimulationMetrics(
execution_time=execution_time,
total_steps=scenario.steps,
total_iterations=scenario.iterations,
convergence_score=final_state["metrics"].get("convergence_score", 0.0),
variance=final_state["metrics"].get("variance", {}),
stability_index=final_state["metrics"].get("stability_index", 0.0)
)
Explanation: Create metrics dataclass from workflow output and timing data.
result = SimulationResult(
scenario=scenario,
final_state=final_state["final_state"],
state_history=final_state["state_history"],
metrics=metrics,
insights=final_state["insights"],
anomalies=final_state["anomalies"],
recommendations=final_state["recommendations"]
)
Explanation: Create complete result object with all simulation data.
# Store result
self.results[scenario.scenario_id] = result
# Update statistics
self.stats["total_simulations"] += 1
self.stats["simulations_by_type"][scenario.simulation_type.value] += 1
self.stats["total_steps"] += scenario.steps
self.stats["total_iterations"] += scenario.iterations
self.stats["avg_execution_time"] = (
(self.stats["avg_execution_time"] * (self.stats["total_simulations"] - 1) + execution_time)
/ self.stats["total_simulations"]
)
Explanation: Store result and update running statistics, including rolling average execution time.
Explanation: Return complete simulation result.
Lines 433-475: SimulationPattern Coordinator
class SimulationPattern:
"""
Comprehensive Simulation Pattern coordinator with LangChain/LangGraph.
"""
def __init__(self, model_name: str = "gpt-4"):
self.simulator = IntelligentSimulator(model_name)
Explanation: Coordinator class that wraps the simulator. Simplifies API for end users.
def create_scenario(
self,
scenario_id: str,
name: str,
simulation_type: SimulationType,
initial_state: Dict[str, float],
steps: int = 100,
iterations: int = 10,
parameters: Optional[Dict[str, Any]] = None
) -> SimulationScenario:
"""Create a simulation scenario."""
return SimulationScenario(
scenario_id=scenario_id,
name=name,
simulation_type=simulation_type,
initial_state=initial_state,
steps=steps,
iterations=iterations,
parameters=parameters or {}
)
Explanation: Factory method to create scenario objects with proper defaults.
def run_simulation(self, scenario: SimulationScenario) -> SimulationResult:
"""Run a simulation."""
return self.simulator.simulate(scenario)
Explanation: Delegate to simulator’s simulate method.
def get_statistics(self) -> Dict[str, Any]:
"""Get simulation statistics."""
return self.simulator.get_statistics()
Explanation: Expose simulator’s statistics.
Lines 477-592: Demo Code
if __name__ == "__main__":
print("=== Simulation Pattern with LangChain/LangGraph Demo ===\n")
Explanation: Demo section only runs when file is executed directly (not imported).
print("Scenario 1: Monte Carlo Simulation")
print("-" * 50)
pattern = SimulationPattern(model_name="gpt-4")
Explanation: Create pattern instance with GPT-4 model.
# Create Monte Carlo scenario
mc_scenario = pattern.create_scenario(
scenario_id="mc-001",
name="Market Price Simulation",
simulation_type=SimulationType.MONTE_CARLO,
initial_state={"price": 100.0, "volume": 1000.0, "volatility": 0.2},
steps=50,
iterations=100
)
Explanation: Create Monte Carlo scenario simulating market prices with initial price $100, volume 1000, volatility 0.2. Run 50 steps, 100 iterations.
print(f"Running {mc_scenario.name}...")
result1 = pattern.run_simulation(mc_scenario)
Explanation: Execute simulation and get results.
print(f"\nFinal State:")
for key, value in result1.final_state.items():
print(f" {key}: {value:.2f}")
Explanation: Display final state values formatted to 2 decimal places.
print(f"\nMetrics:")
print(f" Execution Time: {result1.metrics.execution_time:.3f}s")
print(f" Convergence Score: {result1.metrics.convergence_score:.3f}")
print(f" Stability Index: {result1.metrics.stability_index:.3f}")
Explanation: Display key metrics.
if result1.anomalies:
print(f"\nAnomalies Detected:")
for anomaly in result1.anomalies:
print(f" - {anomaly}")
Explanation: Display any detected anomalies.
if result1.insights:
print(f"\nLLM Insights:")
for insight in result1.insights:
print(f" - {insight}")
Explanation: Display LLM-generated insights.
The demo continues with two more scenarios (agent-based opinion dynamics and discrete event queue management), demonstrating different simulation types, then displays aggregate statistics showing total simulations run, steps simulated, and average execution time.
Key Design Patterns
- State Machine Pattern: Workflow progresses through defined states
- Strategy Pattern: Different simulation algorithms selected at runtime
- Chain of Responsibility: Workflow nodes process state sequentially
- Observer Pattern: Statistics tracking observes all simulations
- Template Method: Common workflow structure with variable implementations
Performance Considerations
- Monte Carlo: O(iterations × steps × variables) complexity
- Agent-Based: O(steps × agents²) due to pairwise interactions
- Discrete Event: O(steps × events) – most efficient for sparse events
- Memory: Stores full history – O(steps × variables)
Extension Points
- Add new simulation types by implementing
_run_<type> method
- Customize metrics in
_calculate_metrics_node
- Enhance anomaly detection in
_detect_anomalies_node
- Modify LLM prompts for domain-specific insights
- Add visualization nodes to workflow