"""
Pattern 85: Mobile Agent Pattern with LangChain/LangGraph
Purpose: Agent migration between execution environments with intelligent routing.
Enables agents to move across environments while preserving state and optimizing placement.
"""
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 json
import time
class AgentStatus(Enum):
"""Agent execution status."""
IDLE = "idle"
EXECUTING = "executing"
MIGRATING = "migrating"
SUSPENDED = "suspended"
TERMINATED = "terminated"
class MigrationType(Enum):
"""Types of migration."""
MANUAL = "manual"
LOAD_BALANCING = "load_balancing"
RESOURCE_OPTIMIZATION = "resource_optimization"
FAULT_TOLERANCE = "fault_tolerance"
STRATEGIC = "strategic"
class EnvironmentCapability(Enum):
"""Environment capabilities."""
COMPUTE = "compute"
STORAGE = "storage"
NETWORK = "network"
GPU = "gpu"
MEMORY = "memory"
@dataclass
class ExecutionEnvironment:
"""Execution environment definition."""
env_id: str
name: str
capabilities: List[EnvironmentCapability]
current_load: float = 0.0
max_load: float = 100.0
latency_ms: float = 10.0
available: bool = True
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class MigrationRecord:
"""Record of agent migration."""
migration_id: str
agent_id: str
from_env: str
to_env: str
migration_type: MigrationType
reason: str
timestamp: datetime
duration_ms: float
success: bool
state_size_bytes: int
@dataclass
class AgentTask:
"""Task executed by agent."""
task_id: str
description: str
required_capabilities: List[EnvironmentCapability]
execution_time: float
result: Optional[str] = None
class MigrationState(TypedDict):
"""State for migration workflow."""
agent_id: str
current_env: str
target_env: Optional[str]
migration_reason: str
agent_state: Dict[str, Any]
task_history: List[Dict[str, Any]]
environments: Dict[str, Dict[str, Any]]
migration_score: float
recommendations: List[str]
approved: bool
class MobileAgent:
"""
Mobile Agent with state preservation and migration capabilities.
"""
def __init__(self, agent_id: str, name: str, initial_env: str):
self.agent_id = agent_id
self.name = name
self.current_env = initial_env
self.status = AgentStatus.IDLE
self.state_data: Dict[str, Any] = {}
self.task_history: List[AgentTask] = []
self.migration_history: List[MigrationRecord] = []
self.created_at = datetime.now()
def serialize_state(self) -> Dict[str, Any]:
"""Serialize agent state for migration."""
return {
"agent_id": self.agent_id,
"name": self.name,
"current_env": self.current_env,
"status": self.status.value,
"state_data": self.state_data,
"task_history": [
{
"task_id": t.task_id,
"description": t.description,
"result": t.result
} for t in self.task_history
],
"created_at": self.created_at.isoformat()
}
def deserialize_state(self, state: Dict[str, Any], new_env: str):
"""Restore agent state after migration."""
self.current_env = new_env
self.state_data = state.get("state_data", {})
# Task history restored from serialized data
def execute_task(self, task: AgentTask) -> str:
"""Execute task in current environment."""
self.status = AgentStatus.EXECUTING
# Simulate task execution
time.sleep(0.01) # Simulate work
result = f"Task {task.task_id} completed in {self.current_env}"
task.result = result
self.task_history.append(task)
self.status = AgentStatus.IDLE
return result
class IntelligentMigrationManager:
"""
Intelligent Migration Manager with LangGraph workflow.
Features:
- Environment analysis
- Load balancing
- Intelligent routing
- State preservation
- Migration optimization
"""
def __init__(self, model_name: str = "gpt-4", temperature: float = 0.3):
self.llm = ChatOpenAI(model=model_name, temperature=temperature)
self.agents: Dict[str, MobileAgent] = {}
self.environments: Dict[str, ExecutionEnvironment] = {}
self.workflow = self._build_workflow()
# Statistics
self.stats = {
"total_migrations": 0,
"migrations_by_type": {t.value: 0 for t in MigrationType},
"successful_migrations": 0,
"failed_migrations": 0,
"total_agents": 0,
"avg_migration_time_ms": 0.0
}
def _build_workflow(self) -> StateGraph:
"""Build LangGraph workflow for intelligent migration."""
workflow = StateGraph(MigrationState)
workflow.add_node("analyze_environments", self._analyze_environments_node)
workflow.add_node("evaluate_migration", self._evaluate_migration_node)
workflow.add_node("select_target", self._select_target_node)
workflow.add_node("validate_migration", self._validate_migration_node)
workflow.add_node("generate_recommendations", self._generate_recommendations_node)
workflow.set_entry_point("analyze_environments")
workflow.add_edge("analyze_environments", "evaluate_migration")
workflow.add_edge("evaluate_migration", "select_target")
workflow.add_edge("select_target", "validate_migration")
workflow.add_edge("validate_migration", "generate_recommendations")
workflow.add_edge("generate_recommendations", END)
return workflow.compile()
def _analyze_environments_node(self, state: MigrationState) -> MigrationState:
"""Analyze available environments."""
env_analysis = {}
for env_id, env in self.environments.items():
env_analysis[env_id] = {
"available": env.available,
"load": env.current_load,
"capacity": env.max_load - env.current_load,
"latency_ms": env.latency_ms,
"capabilities": [c.value for c in env.capabilities]
}
state["environments"] = env_analysis
return state
def _evaluate_migration_node(self, state: MigrationState) -> MigrationState:
"""Evaluate need for migration using LLM."""
current_env = state["current_env"]
reason = state["migration_reason"]
environments = state["environments"]
# Use LLM to evaluate migration necessity
prompt = ChatPromptTemplate.from_template(
"Evaluate if agent migration is beneficial:\n"
"Current Environment: {current_env}\n"
"Reason: {reason}\n"
"Available Environments: {envs}\n\n"
"Provide migration score (0.0-1.0) and brief analysis."
)
chain = prompt | self.llm
response = chain.invoke({
"current_env": current_env,
"reason": reason,
"envs": str(environments)
})
# Extract score (simplified - in production, parse LLM response)
analysis = str(response.content)
# Calculate score based on environment conditions
if current_env in environments:
current_load = environments[current_env]["load"]
capacity = environments[current_env]["capacity"]
if current_load > 80:
score = 0.9 # High load, strong migration candidate
elif capacity MigrationState:
"""Select optimal target environment."""
current_env = state["current_env"]
environments = state["environments"]
migration_score = state["migration_score"]
if migration_score best_score:
best_score = total_score
best_env = env_id
state["target_env"] = best_env
return state
def _validate_migration_node(self, state: MigrationState) -> MigrationState:
"""Validate migration is safe and beneficial."""
target_env = state["target_env"]
if target_env is None:
state["approved"] = False
else:
# Check if target environment can accept agent
target_data = state["environments"].get(target_env, {})
if target_data.get("available") and target_data.get("capacity", 0) > 10:
state["approved"] = True
else:
state["approved"] = False
state["target_env"] = None
return state
def _generate_recommendations_node(self, state: MigrationState) -> MigrationState:
"""Generate migration recommendations."""
recommendations = []
if state["approved"]:
recommendations.append(f"Migrate to {state['target_env']} - optimal environment identified")
recommendations.append("State preservation verified")
recommendations.append("Target environment has sufficient capacity")
else:
recommendations.append("Migration not recommended at this time")
if state["migration_score"] Optional[MigrationRecord]:
"""Migrate agent with intelligent routing."""
if agent_id not in self.agents:
return None
agent = self.agents[agent_id]
start_time = time.time()
# Prepare workflow state
initial_state: MigrationState = {
"agent_id": agent_id,
"current_env": agent.current_env,
"target_env": target_env,
"migration_reason": reason,
"agent_state": agent.serialize_state(),
"task_history": [
{"task_id": t.task_id, "result": t.result}
for t in agent.task_history
],
"environments": {},
"migration_score": 0.0,
"recommendations": [],
"approved": False
}
# Run workflow
result_state = self.workflow.invoke(initial_state)
# Check if migration approved
if not result_state["approved"]:
return None
final_target = result_state["target_env"]
if not final_target or final_target not in self.environments:
return None
# Perform migration
agent.status = AgentStatus.MIGRATING
# Serialize state
serialized_state = agent.serialize_state()
state_size = len(json.dumps(serialized_state))
# Update environment loads
old_env = self.environments.get(agent.current_env)
if old_env:
old_env.current_load = max(0, old_env.current_load - 10)
new_env = self.environments[final_target]
new_env.current_load = min(new_env.max_load, new_env.current_load + 10)
# Migrate agent
agent.deserialize_state(serialized_state, final_target)
agent.status = AgentStatus.IDLE
duration = (time.time() - start_time) * 1000 # ms
# Record migration
record = MigrationRecord(
migration_id=f"mig-{len(agent.migration_history) + 1}",
agent_id=agent_id,
from_env=result_state["current_env"],
to_env=final_target,
migration_type=migration_type,
reason=reason,
timestamp=datetime.now(),
duration_ms=duration,
success=True,
state_size_bytes=state_size
)
agent.migration_history.append(record)
# Update stats
self.stats["total_migrations"] += 1
self.stats["migrations_by_type"][migration_type.value] += 1
self.stats["successful_migrations"] += 1
total_time = self.stats["avg_migration_time_ms"] * (self.stats["total_migrations"] - 1) + duration
self.stats["avg_migration_time_ms"] = total_time / self.stats["total_migrations"]
return record
def auto_balance(self):
"""Automatically balance agents across environments."""
# Find overloaded environments
overloaded = [
env_id for env_id, env in self.environments.items()
if env.current_load > 80
]
# Migrate agents from overloaded environments
for agent in self.agents.values():
if agent.current_env in overloaded:
self.migrate_agent(
agent.agent_id,
reason="load_balancing",
migration_type=MigrationType.LOAD_BALANCING
)
def get_statistics(self) -> Dict[str, Any]:
"""Get migration statistics."""
return {
**self.stats,
"total_environments": len(self.environments),
"active_agents": sum(1 for a in self.agents.values() if a.status != AgentStatus.TERMINATED)
}
class MobileAgentPattern:
"""
Comprehensive Mobile Agent Pattern coordinator with LangChain/LangGraph.
"""
def __init__(self, model_name: str = "gpt-4"):
self.manager = IntelligentMigrationManager(model_name=model_name)
def create_environment(
self,
env_id: str,
name: str,
capabilities: List[EnvironmentCapability],
max_load: float = 100.0
) -> ExecutionEnvironment:
"""Create execution environment."""
env = ExecutionEnvironment(
env_id=env_id,
name=name,
capabilities=capabilities,
max_load=max_load
)
self.manager.register_environment(env)
return env
def create_agent(
self,
agent_id: str,
name: str,
initial_env: str
) -> MobileAgent:
"""Create mobile agent."""
agent = MobileAgent(agent_id, name, initial_env)
self.manager.register_agent(agent)
return agent
def migrate(
self,
agent_id: str,
target_env: Optional[str] = None,
reason: str = "manual"
) -> Optional[MigrationRecord]:
"""Migrate agent."""
return self.manager.migrate_agent(agent_id, target_env, reason)
def auto_balance(self):
"""Auto-balance agents."""
self.manager.auto_balance()
def get_statistics(self) -> Dict[str, Any]:
"""Get statistics."""
return self.manager.get_statistics()
if __name__ == "__main__":
print("=== Mobile Agent Pattern with LangChain/LangGraph Demo ===\n")
print("Scenario 1: Environment Setup and Agent Creation")
print("-" * 50)
pattern = MobileAgentPattern(model_name="gpt-4")
# Create environments
env1 = pattern.create_environment(
"env-001",
"Production Server",
[EnvironmentCapability.COMPUTE, EnvironmentCapability.STORAGE, EnvironmentCapability.NETWORK],
max_load=100.0
)
print(f"✓ Created environment: {env1.name}")
env2 = pattern.create_environment(
"env-002",
"Edge Node",
[EnvironmentCapability.COMPUTE, EnvironmentCapability.NETWORK],
max_load=80.0
)
print(f"✓ Created environment: {env2.name}")
env3 = pattern.create_environment(
"env-003",
"GPU Cluster",
[EnvironmentCapability.COMPUTE, EnvironmentCapability.GPU, EnvironmentCapability.MEMORY],
max_load=120.0
)
print(f"✓ Created environment: {env3.name}")
# Create agents
agent1 = pattern.create_agent("agent-001", "DataProcessor", "env-001")
print(f"\n✓ Created agent: {agent1.name} in {agent1.current_env}")
agent2 = pattern.create_agent("agent-002", "APIWorker", "env-001")
print(f"✓ Created agent: {agent2.name} in {agent2.current_env}")
agent3 = pattern.create_agent("agent-003", "MLAgent", "env-002")
print(f"✓ Created agent: {agent3.name} in {agent3.current_env}")
print("\n" + "=" * 50)
print("\nScenario 2: Task Execution")
print("-" * 50)
# Execute tasks
task1 = AgentTask(
task_id="task-001",
description="Process data batch",
required_capabilities=[EnvironmentCapability.COMPUTE],
execution_time=0.1
)
result = agent1.execute_task(task1)
print(f"Agent {agent1.name}: {result}")
task2 = AgentTask(
task_id="task-002",
description="Handle API request",
required_capabilities=[EnvironmentCapability.NETWORK],
execution_time=0.05
)
result = agent2.execute_task(task2)
print(f"Agent {agent2.name}: {result}")
print("\n" + "=" * 50)
print("\nScenario 3: Intelligent Migration")
print("-" * 50)
# Simulate high load on env-001
pattern.manager.environments["env-001"].current_load = 85.0
print(f"Simulated high load on {env1.name}: {env1.current_load}%")
# Migrate agent1 with intelligent routing
print(f"\nMigrating {agent1.name} with intelligent routing...")
record = pattern.migrate(agent1.agent_id, reason="high_load_detected")
if record:
print(f"✓ Migration successful!")
print(f" From: {record.from_env}")
print(f" To: {record.to_env}")
print(f" Duration: {record.duration_ms:.2f}ms")
print(f" State Size: {record.state_size_bytes} bytes")
else:
print("✗ Migration not recommended")
print("\n" + "=" * 50)
print("\nScenario 4: Manual Migration")
print("-" * 50)
# Manual migration to GPU cluster
print(f"Migrating {agent3.name} to GPU cluster...")
record = pattern.migrate(agent3.agent_id, target_env="env-003", reason="gpu_required")
if record:
print(f"✓ Migration successful!")
print(f" From: {record.from_env}")
print(f" To: {record.to_env}")
print(f" Reason: {record.reason}")
else:
print("✗ Migration failed or not approved")
print("\n" + "=" * 50)
print("\nScenario 5: Auto Load Balancing")
print("-" * 50)
print("Running auto load balancing...")
pattern.auto_balance()
print("✓ Load balancing complete")
print("\n" + "=" * 50)
print("\nScenario 6: Migration History")
print("-" * 50)
for agent in pattern.manager.agents.values():
if agent.migration_history:
print(f"\n{agent.name} migration history:")
for i, mig in enumerate(agent.migration_history, 1):
print(f" {i}. {mig.from_env} → {mig.to_env} ({mig.migration_type.value})")
print("\n" + "=" * 50)
print("\nStatistics")
print("-" * 50)
stats = pattern.get_statistics()
print(f"Total Agents: {stats['total_agents']}")
print(f"Active Agents: {stats['active_agents']}")
print(f"Total Migrations: {stats['total_migrations']}")
print(f"Successful Migrations: {stats['successful_migrations']}")
print(f"Average Migration Time: {stats['avg_migration_time_ms']:.2f}ms")
print(f"Total Environments: {stats['total_environments']}")
print(f"\nMigrations by Type:")
for mig_type, count in stats['migrations_by_type'].items():
if count > 0:
print(f" {mig_type}: {count}")
print("\n" + "=" * 50)
print("Mobile agent pattern demonstration complete!")
print("=" * 50)
Agentic Architecture Pattern: Mobile Agent
Pattern 85: Mobile Agent Pattern – Line-by-Line Explanation
Overview
This pattern implements mobile agents that can migrate between execution environments, carrying their state and logic. It includes intelligent routing, load balancing, state serialization, and LLM-powered migration decisions.
Code Explanation
Lines 1-17: Imports
"""
Pattern 85: Mobile Agent Pattern with LangChain/LangGraph
Purpose: Mobile agents with migration, state transfer, and intelligent routing.
Provides agent mobility with LLM-powered migration decisions and load balancing.
"""
Explanation: Module docstring – pattern for agents that move between environments.
from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import pickle
import base64
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json
Explanation: Includes pickle for object serialization and base64 for encoding serialized state.
Lines 19-34: Enum Definitions
class AgentStatus(Enum):
"""Status of mobile agent."""
IDLE = "idle"
MIGRATING = "migrating"
EXECUTING = "executing"
WAITING = "waiting"
TERMINATED = "terminated"
Explanation: Five agent lifecycle states.
class MigrationStrategy(Enum):
"""Strategy for agent migration."""
LOAD_BALANCED = "load_balanced" # Migrate to least loaded
PROXIMITY = "proximity" # Migrate to closest
CAPABILITY = "capability" # Migrate to most capable
COST = "cost" # Migrate to cheapest
INTELLIGENT = "intelligent" # LLM decides
Explanation: Five migration strategies: – LOAD_BALANCED: Choose environment with lowest load (CPU, memory, queue size) – PROXIMITY: Choose geographically closest (minimize latency) – CAPABILITY: Choose environment with best resources (GPU, special hardware) – COST: Choose cheapest environment (cloud cost optimization) – INTELLIGENT: Use LLM to analyze requirements and choose best environment
Lines 36-92: Dataclasses
@dataclass
class Environment:
"""Execution environment for mobile agents."""
id: str
name: str
capabilities: List[str]
load: float = 0.0 # 0.0 to 1.0
cost_per_hour: float = 0.0
location: str = "unknown"
active_agents: List[str] = field(default_factory=list)
Explanation: Represents an execution environment: – capabilities: What this environment can do (e.g., [“gpu”, “high_memory”, “python3.11”]) – load: Current utilization (0.0 = idle, 1.0 = maxed out) – cost_per_hour: For cloud environments (e.g., $0.05/hour) – location: Geographic location (e.g., “us-east-1”, “europe-west”) – active_agents: List of agent IDs currently running
@dataclass
class AgentTask:
"""Task for mobile agent."""
id: str
code: Callable
data: Dict[str, Any]
requirements: List[str] = field(default_factory=list)
Explanation: Task agent will execute: – code: Actual function to run – data: Input data – requirements: Required capabilities (e.g., [“gpu”, “tensorflow”])
@dataclass
class MigrationDecision:
"""Decision about agent migration."""
should_migrate: bool
target_environment: Optional[str]
reasoning: str
strategy: MigrationStrategy
confidence: float
Explanation: Migration decision made by system or LLM: – should_migrate: Yes/no decision – target_environment: Destination environment ID – reasoning: Why this decision was made – strategy: Which strategy was used – confidence: 0.0-1.0 confidence in decision
@dataclass
class MigrationRecord:
"""Record of agent migration."""
agent_id: str
from_environment: str
to_environment: str
timestamp: datetime = field(default_factory=datetime.now)
state_size_bytes: int = 0
migration_time_ms: float = 0.0
success: bool = True
Explanation: Historical record of migration event. Used for analysis and optimization.
Lines 94-110: TypedDict State
class MobileAgentState(TypedDict):
"""State for mobile agent workflow."""
agent_id: str
current_environment: str
task: AgentTask
agent_state: Dict[str, Any]
serialized_state: Optional[str]
migration_decision: Optional[MigrationDecision]
target_environment: Optional[str]
status: AgentStatus
execution_result: Any
insights: List[str]
Explanation: Workflow state tracking agent identity, location, state, and migration decisions.
Lines 112-160: MobileAgent Initialization
class MobileAgent:
"""
Mobile Agent with LangGraph workflow integration.
Features:
- Agent migration between environments
- State serialization and transfer
- Intelligent routing decisions
- Load balancing
- Capability-based placement
- Cost optimization
- Proximity awareness
- Migration history tracking
"""
Explanation: Main mobile agent class with intelligent migration.
def __init__(
self,
agent_id: str,
initial_environment: str,
model_name: str = "gpt-4",
temperature: float = 0.5
):
self.agent_id = agent_id
self.current_environment = initial_environment
self.llm = ChatOpenAI(model=model_name, temperature=temperature)
# Agent state
self.state: Dict[str, Any] = {}
self.status = AgentStatus.IDLE
# Migration history
self.migration_history: List[MigrationRecord] = []
# Build workflow
self.workflow = self._build_workflow()
# Statistics
self.stats = {
"total_migrations": 0,
"successful_migrations": 0,
"failed_migrations": 0,
"total_execution_time_ms": 0.0
}
Explanation: Initialize agent with ID, starting environment, LLM for decisions, state, and tracking.
Lines 162-180: Workflow Building
def _build_workflow(self) -> StateGraph:
"""Build LangGraph workflow."""
workflow = StateGraph(MobileAgentState)
workflow.add_node("analyze_requirements", self._analyze_requirements_node)
workflow.add_node("decide_migration", self._decide_migration_node)
workflow.add_node("serialize_state", self._serialize_state_node)
workflow.add_node("migrate", self._migrate_node)
workflow.add_node("execute_task", self._execute_task_node)
Explanation: Create 5-node workflow: 1. Analyze requirements: Check what task needs 2. Decide migration: Determine if/where to migrate 3. Serialize state: Package agent state for transfer 4. Migrate: Move agent to new environment 5. Execute task: Run the actual task
workflow.set_entry_point("analyze_requirements")
workflow.add_edge("analyze_requirements", "decide_migration")
workflow.add_edge("decide_migration", "serialize_state")
workflow.add_edge("serialize_state", "migrate")
workflow.add_edge("migrate", "execute_task")
workflow.add_edge("execute_task", END)
return workflow.compile()
Explanation: Linear flow through migration and execution.
Lines 182-204: Analyze Requirements Node
def _analyze_requirements_node(self, state: MobileAgentState) -> MobileAgentState:
"""Analyze task requirements."""
task = state["task"]
state["insights"] = []
state["insights"].append(f"Analyzing requirements for task {task.id}")
if task.requirements:
state["insights"].append(f"Task requires: {', '.join(task.requirements)}")
else:
state["insights"].append("Task has no specific requirements")
return state
Explanation: Extract and document task requirements. This informs migration decision.
Lines 206-312: Decide Migration Node (LLM-Powered)
def _decide_migration_node(
self,
state: MobileAgentState,
environments: Dict[str, Environment],
strategy: MigrationStrategy = MigrationStrategy.INTELLIGENT
) -> MobileAgentState:
"""Decide whether and where to migrate."""
current_env_id = state["current_environment"]
task = state["task"]
current_env = environments.get(current_env_id)
Explanation: Extract current environment and task information.
if strategy == MigrationStrategy.LOAD_BALANCED:
# Choose least loaded environment
candidates = [(env_id, env) for env_id, env in environments.items()]
candidates.sort(key=lambda x: x[1].load)
if candidates and candidates[0][1].load < current_env.load:
state["migration_decision"] = MigrationDecision(
should_migrate=True,
target_environment=candidates[0][0],
reasoning=f"Target has lower load ({candidates[0][1].load:.2f} vs {current_env.load:.2f})",
strategy=strategy,
confidence=0.9
)
else:
state["migration_decision"] = MigrationDecision(
should_migrate=False,
target_environment=None,
reasoning="Current environment has best load",
strategy=strategy,
confidence=0.9
)
Explanation: LOAD_BALANCED strategy – choose environment with lowest load. Migrate only if better option exists.
elif strategy == MigrationStrategy.CAPABILITY:
# Choose environment with required capabilities
suitable = []
for env_id, env in environments.items():
if all(req in env.capabilities for req in task.requirements):
suitable.append((env_id, env))
if suitable:
# Prefer environment with most capabilities
suitable.sort(key=lambda x: len(x[1].capabilities), reverse=True)
if suitable[0][0] != current_env_id:
state["migration_decision"] = MigrationDecision(
should_migrate=True,
target_environment=suitable[0][0],
reasoning=f"Target has required capabilities: {suitable[0][1].capabilities}",
strategy=strategy,
confidence=0.95
)
else:
state["migration_decision"] = MigrationDecision(
should_migrate=False,
target_environment=None,
reasoning="Current environment has required capabilities",
strategy=strategy,
confidence=0.95
)
else:
state["migration_decision"] = MigrationDecision(
should_migrate=False,
target_environment=None,
reasoning="No suitable environments found",
strategy=strategy,
confidence=0.5
)
Explanation: CAPABILITY strategy – find environments that meet task requirements. Choose one with most capabilities.
elif strategy == MigrationStrategy.INTELLIGENT:
# Use LLM to decide
prompt = ChatPromptTemplate.from_messages([
("system", "You are an expert in distributed systems and agent placement optimization."),
("user", """Analyze this mobile agent migration scenario:
Agent ID: {agent_id}
Current Environment: {current_env}
- Capabilities: {current_capabilities}
- Load: {current_load:.2f}
- Cost: ${current_cost:.2f}/hour
Task Requirements: {requirements}
Available Environments:
{environments}
Provide migration decision as JSON with:
- should_migrate: true/false
- target_environment: ID of target (or null)
- reasoning: Detailed explanation
- confidence: 0.0-1.0 confidence score
Consider: capabilities, load, cost, and migration overhead.""")
])
# Format environments
env_list = []
for env_id, env in environments.items():
if env_id != current_env_id:
env_list.append(
f" - {env_id}: {env.name}\n"
f" Capabilities: {env.capabilities}\n"
f" Load: {env.load:.2f}\n"
f" Cost: ${env.cost_per_hour:.2f}/hour"
)
Explanation: INTELLIGENT strategy – prepare comprehensive context for LLM.
chain = prompt | self.llm
response = chain.invoke({
"agent_id": state["agent_id"],
"current_env": current_env.name,
"current_capabilities": current_env.capabilities,
"current_load": current_env.load,
"current_cost": current_env.cost_per_hour,
"requirements": task.requirements,
"environments": "\n".join(env_list)
})
Explanation: Invoke LLM with full context about current and available environments.
try:
decision_data = json.loads(response.content)
state["migration_decision"] = MigrationDecision(
should_migrate=decision_data.get("should_migrate", False),
target_environment=decision_data.get("target_environment"),
reasoning=decision_data.get("reasoning", "LLM decision"),
strategy=strategy,
confidence=decision_data.get("confidence", 0.5)
)
except Exception as e:
# Fallback to staying
state["migration_decision"] = MigrationDecision(
should_migrate=False,
target_environment=None,
reasoning=f"LLM decision failed: {str(e)}",
strategy=strategy,
confidence=0.0
)
Explanation: Parse LLM response. Fallback to staying if parsing fails.
state["target_environment"] = state["migration_decision"].target_environment
return state
Lines 314-355: Serialize State Node
def _serialize_state_node(self, state: MobileAgentState) -> MobileAgentState:
"""Serialize agent state for migration."""
decision = state["migration_decision"]
if not decision or not decision.should_migrate:
state["serialized_state"] = None
state["insights"].append("No migration needed, skipping serialization")
return state
Explanation: Skip serialization if not migrating.
try:
# Serialize agent state
agent_state = state["agent_state"]
serialized = pickle.dumps(agent_state)
encoded = base64.b64encode(serialized).decode('utf-8')
state["serialized_state"] = encoded
state["insights"].append(f"State serialized: {len(encoded)} bytes")
except Exception as e:
state["insights"].append(f"Serialization failed: {str(e)}")
state["serialized_state"] = None
# Cancel migration
state["migration_decision"].should_migrate = False
return state
Explanation: Serialize state using pickle, encode as base64 string for transfer. If serialization fails, cancel migration.
Why pickle + base64? – Pickle: Python object serialization (preserves types, nested structures) – Base64: Encode binary as text (safe for JSON, network transfer)
Lines 357-420: Migrate Node
def _migrate_node(
self,
state: MobileAgentState,
environments: Dict[str, Environment]
) -> MobileAgentState:
"""Migrate agent to new environment."""
decision = state["migration_decision"]
if not decision or not decision.should_migrate:
state["insights"].append("Staying in current environment")
return state
Explanation: Skip migration if decision says stay.
target_env_id = decision.target_environment
current_env_id = state["current_environment"]
if target_env_id not in environments:
state["insights"].append(f"Target environment {target_env_id} not found")
return state
Explanation: Validate target environment exists.
try:
start_time = datetime.now()
# Update environment tracking
current_env = environments[current_env_id]
target_env = environments[target_env_id]
# Remove from current
if state["agent_id"] in current_env.active_agents:
current_env.active_agents.remove(state["agent_id"])
# Add to target
target_env.active_agents.append(state["agent_id"])
# Update state
state["current_environment"] = target_env_id
state["status"] = AgentStatus.IDLE
# Deserialize state in new environment
if state["serialized_state"]:
decoded = base64.b64decode(state["serialized_state"])
deserialized_state = pickle.loads(decoded)
state["agent_state"] = deserialized_state
Explanation: Perform migration: 1. Remove agent from current environment’s active list 2. Add agent to target environment’s active list 3. Update agent’s current_environment 4. Deserialize state in new environment
# Record migration
end_time = datetime.now()
migration_time = (end_time - start_time).total_seconds() * 1000
record = MigrationRecord(
agent_id=state["agent_id"],
from_environment=current_env_id,
to_environment=target_env_id,
state_size_bytes=len(state["serialized_state"]) if state["serialized_state"] else 0,
migration_time_ms=migration_time,
success=True
)
self.migration_history.append(record)
self.stats["total_migrations"] += 1
self.stats["successful_migrations"] += 1
state["insights"].append(
f"Migrated from {current_env.name} to {target_env.name} in {migration_time:.2f}ms"
)
except Exception as e:
state["insights"].append(f"Migration failed: {str(e)}")
self.stats["failed_migrations"] += 1
return state
Explanation: Record migration metrics, update statistics, handle failures.
Lines 422-457: Execute Task Node
def _execute_task_node(self, state: MobileAgentState) -> MobileAgentState:
"""Execute agent task in current environment."""
task = state["task"]
state["status"] = AgentStatus.EXECUTING
try:
start_time = datetime.now()
# Execute task code
result = task.code(task.data)
end_time = datetime.now()
execution_time = (end_time - start_time).total_seconds() * 1000
state["execution_result"] = result
state["status"] = AgentStatus.IDLE
self.stats["total_execution_time_ms"] += execution_time
state["insights"].append(f"Task executed in {execution_time:.2f}ms")
except Exception as e:
state["execution_result"] = {"error": str(e)}
state["status"] = AgentStatus.IDLE
state["insights"].append(f"Execution failed: {str(e)}")
return state
Explanation: Execute task code with data, measure execution time, handle errors.
Lines 482-640: Demo Code
Scenario 1: Setup Environments
# Create environments
environments = {
"env1": Environment(
id="env1",
name="Standard Compute",
capabilities=["python", "basic"],
load=0.3,
cost_per_hour=0.05,
location="us-east-1"
),
"env2": Environment(
id="env2",
name="GPU Compute",
capabilities=["python", "gpu", "tensorflow"],
load=0.6,
cost_per_hour=0.50,
location="us-west-2"
),
"env3": Environment(
id="env3",
name="High Memory",
capabilities=["python", "high_memory"],
load=0.2,
cost_per_hour=0.15,
location="europe-west"
)
}
Explanation: Define three diverse environments with different capabilities, loads, costs, and locations.
Scenario 2: Load Balanced Migration
agent1 = pattern.create_agent("agent1", "env1")
def simple_task(data: Dict) -> Dict:
return {"result": data.get("value", 0) * 2}
task1 = AgentTask(
id="task1",
code=simple_task,
data={"value": 10},
requirements=[]
)
result1 = pattern.execute_with_migration(
"agent1",
task1,
strategy=MigrationStrategy.LOAD_BALANCED
)
Explanation: Execute task with load balancing. Agent migrates from env1 (load=0.3) to env3 (load=0.2) since it’s less loaded.
Scenario 3: Capability-Based Migration
agent2 = pattern.create_agent("agent2", "env1")
def gpu_task(data: Dict) -> Dict:
# Simulated GPU computation
return {"processed": True, "gpu_accelerated": True}
task2 = AgentTask(
id="task2",
code=gpu_task,
data={"matrix": "large"},
requirements=["gpu", "tensorflow"]
)
result2 = pattern.execute_with_migration(
"agent2",
task2,
strategy=MigrationStrategy.CAPABILITY
)
Explanation: Task requires GPU. Agent migrates from env1 (no GPU) to env2 (has GPU + TensorFlow).
Scenario 4: Intelligent LLM-Based Migration
agent3 = pattern.create_agent("agent3", "env2")
def memory_intensive_task(data: Dict) -> Dict:
return {"memory_used": "high", "completed": True}
task3 = AgentTask(
id="task3",
code=memory_intensive_task,
data={"large_dataset": True},
requirements=["high_memory"]
)
result3 = pattern.execute_with_migration(
"agent3",
task3,
strategy=MigrationStrategy.INTELLIGENT
)
Explanation: LLM analyzes: – Current: env2 (GPU, expensive $0.50/hour, high load 0.6) – Task needs: high memory – env3 has high_memory, lower cost ($0.15/hour), lower load (0.2) – Decision: Migrate to env3 (better match, cheaper, less loaded)
Migration Strategy Comparison
| Strategy | Decision Basis | Use Case |
|---|---|---|
| LOAD_BALANCED | Current load | Distribute work evenly |
| PROXIMITY | Geographic location | Minimize network latency |
| CAPABILITY | Required features | Match task to hardware |
| COST | Price per hour | Optimize cloud spending |
| INTELLIGENT | LLM analysis | Complex multi-factor decisions |
Key Design Patterns
- Strategy Pattern: Different migration strategies
- State Pattern: Agent status management
- Memento Pattern: State serialization for migration
- Proxy Pattern: Agent acts as proxy in remote environment
- Observer Pattern: Migration history tracking
Performance Considerations
- Serialization: O(state_size) – pickle is fast but grows with state
- Migration: O(network_latency + deserialization) – typically 10-100ms
- Decision (Rule-Based): O(environments) – scan all options
- Decision (LLM): O(LLM) – ~1-2 seconds
- Execution: O(task_complexity) – depends on task
Best Practices: – Minimize agent state size (only essentials) – Use compression for large states (gzip before base64) – Batch migrate multiple agents to same destination – Cache LLM decisions for similar scenarios – Consider migration cost vs execution time savings – Use load balancing for homogeneous tasks – Use capability matching for specialized tasks
When to Migrate: – Migration time << execution time savings – Task requires unavailable capabilities – Current environment overloaded – Cost savings justify overhead
Extension Points
- Add checkpointing during long-running tasks
- Implement secure migration with encryption
- Add fault tolerance (retry, rollback)
- Support partial migration (leave state, move code)
- Add network-aware routing (bandwidth, latency)
- Implement agent cloning for parallel execution
- Add environment auto-scaling based on agent demand
- Support heterogeneous code (not just Python)
Leave a Reply