Python pseudocode for each major component of the agentic loop.
class AgenticLoop:
def __init__(self, llm, tools, memory):
self.llm = llm
self.tools = tools
self.memory = memory
def run(self, task: str) -> dict:
"""Run the core agentic loop."""
# 1. Prompt
prompt = self.build_prompt(task)
# 2. Context
context = self.gather_context(task)
# 3. Plan
plan = self.create_plan(task, context)
# 4-7. Execute loop
for cycle in range(plan.max_cycles):
# 4. Reason
decision = self.reason(plan, context)
# 5. Act
result = self.act(decision)
# 6. Observe
observation = self.observe(result)
# Check if done
if observation.status == "success":
break
# Update context
context.update(observation)
# 8. Store
self.store(task, plan, result)
return resultclass PermissionGate:
def __init__(self, rules: list):
self.rules = rules
def evaluate(self, action: dict) -> dict:
"""Evaluate action against permission rules."""
result = {
"allowed": False,
"reason": "",
"risk_level": "unknown"
}
# Check scope
if not self.check_scope(action):
result["reason"] = "Action outside scope"
return result
# Check policy
if not self.check_policy(action):
result["reason"] = "Action violates policy"
return result
# Check blast radius
risk = self.assess_blast_radius(action)
result["risk_level"] = risk
if risk == "critical":
result["reason"] = "Critical risk - requires human approval"
return result
# Check reversibility
if not self.check_reversibility(action):
result["reason"] = "Irreversible action - requires human approval"
return result
result["allowed"] = True
result["reason"] = "Action permitted"
return result
def check_scope(self, action: dict) -> bool:
"""Check if action is within agent's scope."""
allowed_patterns = ["read_file", "write_file", "run_tests"]
return action["type"] in allowed_patterns
def check_policy(self, action: dict) -> bool:
"""Check if action violates any policies."""
deny_patterns = ["rm -rf", "drop table", "delete user"]
return not any(pattern in str(action) for pattern in deny_patterns)
def assess_blast_radius(self, action: dict) -> str:
"""Assess the blast radius of an action."""
if action.get("target_count", 1) > 100:
return "critical"
elif action.get("target_count", 1) > 10:
return "high"
elif action.get("target_count", 1) > 1:
return "medium"
return "low"
def check_reversibility(self, action: dict) -> bool:
"""Check if action can be undone."""
reversible_actions = ["write_file", "edit_file"]
return action["type"] in reversible_actionsclass GoalCheck:
def __init__(self, goal: str, max_cycles: int = 10, max_tokens: int = 50000):
self.goal = goal
self.max_cycles = max_cycles
self.max_tokens = max_tokens
def evaluate(self, state: dict, cycle: int, tokens_used: int) -> dict:
"""Evaluate if goal is met."""
result = {
"done": False,
"reason": "",
"status": "continue"
}
# Check if goal is met
if self.is_goal_met(state):
result["done"] = True
result["reason"] = "Goal achieved"
result["status"] = "success"
return result
# Check cycle limit
if cycle >= self.max_cycles:
result["reason"] = "Max cycles reached"
result["status"] = "max_cycles"
return result
# Check token budget
if tokens_used >= self.max_tokens:
result["reason"] = "Token budget exhausted"
result["status"] = "budget_exhausted"
return result
# Check for diminishing returns
if self.has_diminishing_returns(state):
result["reason"] = "Diminishing returns detected"
result["status"] = "diminishing_returns"
return result
return result
def is_goal_met(self, state: dict) -> bool:
"""Check if the goal is met."""
# Custom logic per goal type
if self.goal.startswith("fix"):
return state.get("tests_passing", False)
elif self.goal.startswith("deploy"):
return state.get("deployed", False)
return False
def has_diminishing_returns(self, state: dict) -> bool:
"""Check if progress has stalled."""
recent_progress = state.get("progress_history", [])[-3:]
if len(recent_progress) < 3:
return False
# Check if progress is flat
return all(
abs(recent_progress[i] - recent_progress[i-1]) < 0.01
for i in range(1, len(recent_progress))
)class SelfHealing:
def __init__(self, healing_rules: dict):
self.healing_rules = healing_rules
def diagnose(self, error: str) -> dict:
"""Diagnose error and determine if self-healing is possible."""
for pattern, rule in self.healing_rules.items():
if pattern in error:
return {
"diagnosable": True,
"pattern": pattern,
"fix": rule["fix"],
"max_attempts": rule["max_attempts"]
}
return {"diagnosable": False}
def apply_fix(self, diagnosis: dict, context: dict) -> dict:
"""Apply the fix for a diagnosed error."""
fix_type = diagnosis["fix"]
if fix_type == "refresh_token":
return self.refresh_token(context)
elif fix_type == "retry_with_backoff":
return self.retry_with_backoff(context)
elif fix_type == "find_alternative":
return self.find_alternative(context)
return {"success": False, "reason": "Unknown fix type"}
def refresh_token(self, context: dict) -> dict:
"""Refresh authentication token."""
new_token = auth.refresh_token(context["refresh_token"])
return {"success": True, "new_token": new_token}
def retry_with_backoff(self, context: dict) -> dict:
"""Retry with exponential backoff."""
import time
for attempt in range(3):
time.sleep(2 ** attempt)
result = context["action"]()
if result["success"]:
return result
return {"success": False, "reason": "Retries exhausted"}class AdaptivePlanner:
def __init__(self, strategy_library: dict):
self.strategy_library = strategy_library
def create_plan(self, task: str, history: list) -> dict:
"""Create plan based on task type and historical success."""
# Identify task type
task_type = self.classify_task(task)
# Get historical success rates for this task type
success_rates = self.get_success_rates(task_type, history)
# Select best strategy
best_strategy = max(success_rates, key=success_rates.get)
# Customize plan
plan = self.strategy_library[best_strategy]
plan = self.customize_plan(plan, task)
return plan
def classify_task(self, task: str) -> str:
"""Classify task type."""
keywords = {
"bug_fix": ["fix", "bug", "error", "failing"],
"feature": ["add", "implement", "create", "new"],
"refactor": ["refactor", "clean", "improve"],
"investigate": ["find", "debug", "investigate"],
}
for task_type, words in keywords.items():
if any(word in task.lower() for word in words):
return task_type
return "general"
def get_success_rates(self, task_type: str, history: list) -> dict:
"""Get success rates for strategies on this task type."""
rates = {}
for entry in history:
if entry["task_type"] == task_type:
strategy = entry["strategy"]
success = entry["success"]
if strategy not in rates:
rates[strategy] = {"success": 0, "total": 0}
rates[strategy]["total"] += 1
if success:
rates[strategy]["success"] += 1
return {
s: r["success"] / r["total"]
for s, r in rates.items()
}class CostOptimizer:
def __init__(self, model_tiers: dict, budget: float):
self.model_tiers = model_tiers
self.budget = budget
self.spent = 0.0
def select_model(self, task_complexity: str) -> str:
"""Select cheapest model that can handle the task."""
return self.model_tiers.get(task_complexity, "gpt-4o-mini")
def track_cost(self, model: str, tokens: int):
"""Track cost of LLM call."""
cost = self.calculate_cost(model, tokens)
self.spent += cost
if self.spent > self.budget * 0.8:
self.alert_budget_warning()
def calculate_cost(self, model: str, tokens: int) -> float:
"""Calculate cost of LLM call."""
rates = {
"gpt-4o-mini": 0.15 / 1_000_000,
"gpt-4o": 2.50 / 1_000_000,
"claude-3-haiku": 0.25 / 1_000_000,
"claude-3-sonnet": 3.00 / 1_000_000,
}
return tokens * rates.get(model, 2.50 / 1_000_000)class MemoryManager:
def __init__(self, storage_path: str):
self.storage_path = storage_path
self.memory = self.load_memory()
def store(self, key: str, value: any, metadata: dict):
"""Store memory with metadata."""
self.memory[key] = {
"value": value,
"metadata": metadata,
"timestamp": datetime.now(),
"access_count": 0
}
self.save_memory()
def retrieve(self, key: str) -> any:
"""Retrieve memory by key."""
if key in self.memory:
self.memory[key]["access_count"] += 1
self.memory[key]["last_accessed"] = datetime.now()
return self.memory[key]["value"]
return None
def get_relevant(self, task: str, top_k: int = 5) -> list:
"""Get most relevant memories for a task."""
scored = []
for key, entry in self.memory.items():
score = self.relevance_score(task, entry)
scored.append((key, entry, score))
scored.sort(key=lambda x: x[2], reverse=True)
return scored[:top_k]
def prune(self, max_age_days: int = 30, min_access_count: int = 2):
"""Remove old, unused memories."""
cutoff = datetime.now() - timedelta(days=max_age_days)
to_remove = []
for key, entry in self.memory.items():
if (entry["timestamp"] < cutoff and
entry["access_count"] < min_access_count):
to_remove.append(key)
for key in to_remove:
del self.memory[key]
self.save_memory()