adu2021 Building AI Coding Agents for the Terminal: Compound Model Architecture and Context Management
Terminal-based AI coding agents must overcome three challenges: managing context across sessions exceeding token budgets, preventing destructive operations when executing arbitrary shell commands, and extending capabilities without prompt bloat. OpenDev solves these through a compound AI architecture: rather than a single LLM, use specialized model routing where each cognitive task (planning, thinking, critique, execution) binds to potentially different models.
Core Concept
Instead of one LLM handling everything, decompose into five specialized roles:
- Planner: Strategy and high-level reasoning
- Thinker: Chain-of-thought analysis when stuck
- Critic: Verification and reflection
- Executor: Tool calling and command generation
- Vision: Image/screenshot understanding
Each role independently configured, lazily initialized, and optimized for its workload. This enables cost, latency, and capability optimization per task. A complex reasoning step uses an expensive model; a simple command uses a fast model.
Architecture Overview
- Workload-Specialized Routing: Router dispatches tasks to appropriate models
- Extended ReAct Cycle: Pre-check, thinking, critique, action, execution, post-processing
- Context Engineering: Compaction, event-driven reminders, dual memory (episodic + working)
- Safety Gates: Approval checks for destructive operations
- Lazy Initialization: Models loaded only when needed
- Token Budget Management: Adaptive message drainage and older observation reduction
Implementation Steps
Implement a compound AI agent system with specialized model routing and extended ReAct execution.
Model Router and Lazy Loading
import torch
from typing import Dict, Optional, Any
from dataclasses import dataclass
@dataclass
class ModelConfig:
"""Configuration for a specialized model."""
name: str
task: str # "planning", "thinking", "critique", "execution", "vision"
model_id: str
max_tokens: int
temperature: float
cost_per_1k_tokens: float
latency_ms_estimate: int
class CompoundAgentRouter:
"""Routes tasks to specialized models based on workload."""
def __init__(self):
self.models: Dict[str, Any] = {}
self.model_configs = {
'planner': ModelConfig(
name='planner',
task='planning',
model_id='claude-3-opus', # High capability for strategy
max_tokens=4096,
temperature=0.7,
cost_per_1k_tokens=15.0,
latency_ms_estimate=2000
),
'thinker': ModelConfig(
name='thinker',
task='thinking',
model_id='claude-3-sonnet', # Medium capability, good speed
max_tokens=2048,
temperature=0.5,
cost_per_1k_tokens=3.0,
latency_ms_estimate=800
),
'critic': ModelConfig(
name='critic',
task='critique',
model_id='claude-3-sonnet', # Verify outputs
max_tokens=1024,
temperature=0.3,
cost_per_1k_tokens=3.0,
latency_ms_estimate=600
),
'executor': ModelConfig(
name='executor',
task='execution',
model_id='claude-3-haiku', # Fast tool calling
max_tokens=512,
temperature=0.2,
cost_per_1k_tokens=0.8,
latency_ms_estimate=300
),
'vision': ModelConfig(
name='vision',
task='vision',
model_id='claude-3-opus-vision', # Multimodal
max_tokens=2048,
temperature=0.5,
cost_per_1k_tokens=15.0,
latency_ms_estimate=1500
)
}
self.loaded_models = set()
def get_model(self, task: str):
"""Lazily load and return model for task."""
if task not in self.model_configs:
raise ValueError(f"Unknown task: {task}")
config = self.model_configs[task]
# Lazy load
if config.name not in self.loaded_models:
self.models[config.name] = self._load_model(config)
self.loaded_models.add(config.name)
return self.models[config.name], config
def _load_model(self, config: ModelConfig):
"""Load model from HuggingFace or API."""
# Simplified; real version uses API clients or local models
print(f"[Loading] {config.name} ({config.model_id})")
return f"MockModel({config.model_id})"
def route_task(self, task_type: str, prompt: str, context: Dict[str, Any]) -> str:
"""Route task to appropriate model."""
model, config = self.get_model(task_type)
# Add task-specific system prompt
system_prompts = {
'planning': "You are a strategic planner for a coding agent. Break down the task into actionable steps.",
'thinking': "You are a reasoning engine. Think carefully through the problem step by step.",
'critique': "You are a quality reviewer. Check the proposed action for errors or improvements.",
'execution': "You are a tool executor. Generate precise tool calls based on the plan.",
'vision': "You analyze screenshots and code. Provide accurate visual/code understanding."
}
# Invoke model
response = self._invoke_model(
model,
system_prompts.get(task_type, ""),
prompt,
max_tokens=config.max_tokens,
temperature=config.temperature
)
return response
def _invoke_model(self, model, system_prompt, prompt, max_tokens, temperature):
"""Invoke model API."""
# Simplified interface
return f"Response from {model}"
Extended ReAct Execution Loop
class ExtendedReActExecutor:
"""Extended Reason-Act-Execute-Observe loop with six phases."""
def __init__(self, router: CompoundAgentRouter, max_iterations=10):
self.router = router
self.max_iterations = max_iterations
self.iteration = 0
def execute(self, user_request: str, context: Dict[str, Any]) -> str:
"""Execute extended ReAct loop."""
memory = {
'episodic': [], # Long-term: important facts
'working': [] # Short-term: current task context
}
for iteration in range(self.max_iterations):
self.iteration = iteration
# Phase 1: Pre-check and Context Compaction
memory, continue_flag = self._phase_precheck_compaction(memory, context)
if not continue_flag:
break
# Phase 2: Optional Thinking
if self._should_think(context):
thought = self._phase_thinking(user_request, memory)
memory['working'].append(f"[THOUGHT] {thought}")
# Phase 3: Optional Self-Critique
if self._should_critique(context):
critique = self._phase_critique(memory)
memory['working'].append(f"[CRITIQUE] {critique}")
# Phase 4: Action Planning
action_plan = self._phase_action(user_request, memory, context)
memory['working'].append(f"[ACTION] {action_plan}")
# Phase 5: Tool Execution
result, is_safe = self._phase_execution(action_plan, context)
if not is_safe:
memory['working'].append(f"[BLOCKED] {result}")
continue
memory['working'].append(f"[RESULT] {result}")
# Phase 6: Post-Processing
done, summary = self._phase_postprocessing(result, memory)
if done:
return summary
return self._final_summary(memory)
def _phase_precheck_compaction(self, memory, context):
"""Phase 1: Drain injected messages and compact context."""
# Check if context is overflowing token budget
token_estimate = self._estimate_tokens(memory)
budget = context.get('token_budget', 8000)
if token_estimate > 0.8 * budget:
# Compaction needed
memory['episodic'] = self._compress_memory(memory['episodic'], keep_top_k=10)
memory['working'] = memory['working'][-5:] # Keep recent 5 items
return memory, True # Continue flag
def _phase_thinking(self, request: str, memory: Dict) -> str:
"""Phase 2: Optional chain-of-thought reasoning."""
prompt = f"""Given the request: {request}
Context: {memory['working'][-3:] if memory['working'] else 'None'}
Think through this carefully:"""
thought = self.router.route_task('thinking', prompt, {})
return thought
def _phase_critique(self, memory: Dict) -> str:
"""Phase 3: Optional self-critique."""
recent_action = memory['working'][-1] if memory['working'] else ""
prompt = f"Review this action: {recent_action}. Any issues or improvements?"
critique = self.router.route_task('critique', prompt, {})
return critique
def _phase_action(self, request: str, memory: Dict, context: Dict) -> str:
"""Phase 4: Determine next action via planning."""
prompt = f"""Task: {request}
Recent progress: {memory['working'][-3:]}
Next action (call tool or analyze):"""
action = self.router.route_task('planning', prompt, context)
return action
def _phase_execution(self, action_plan: str, context: Dict) -> tuple:
"""Phase 5: Execute action with safety gates."""
# Parse action (tool call, etc.)
tool_calls = self._extract_tool_calls(action_plan)
for tool_call in tool_calls:
# Safety check: is this destructive?
if self._is_destructive_operation(tool_call):
# Require approval
approved = self._request_approval(tool_call)
if not approved:
return f"Action blocked: {tool_call}", False
# Execute safely
results = []
for tool_call in tool_calls:
result = self._execute_tool(tool_call)
results.append(result)
return "\n".join(results), True
def _phase_postprocessing(self, result: str, memory: Dict) -> tuple:
"""Phase 6: Decide termination or continuation."""
# Check if task completed
if "completed" in result.lower() or "done" in result.lower():
return True, result
return False, result
def _is_destructive_operation(self, tool_call: str) -> bool:
"""Check if tool call is destructive."""
destructive_patterns = ['rm ', 'delete ', 'rmdir', 'drop ', 'truncate']
return any(pattern in tool_call.lower() for pattern in destructive_patterns)
def _request_approval(self, tool_call: str) -> bool:
"""Request user approval for destructive operation."""
print(f"\n[APPROVAL NEEDED] Execute this command?\n{tool_call}\n(y/n): ", end="")
response = input().strip().lower()
return response == 'y'
def _extract_tool_calls(self, action_plan: str) -> list:
"""Parse tool calls from action plan."""
# Simplified; real version uses structured output
return [action_plan]
def _execute_tool(self, tool_call: str) -> str:
"""Execute tool (shell command, API call, etc.)."""
# Simplified; real version uses subprocess, requests, etc.
return f"Executed: {tool_call}"
def _should_think(self, context: Dict) -> bool:
"""Decide whether to use thinking phase."""
complexity = context.get('complexity_score', 0.5)
return complexity > 0.7
def _should_critique(self, context: Dict) -> bool:
"""Decide whether to use critique phase."""
return context.get('enable_critique', True)
def _estimate_tokens(self, memory: Dict) -> int:
"""Estimate tokens used by memory."""
return sum(len(item.split()) for item in memory['working']) * 1.3
def _compress_memory(self, episodic: list, keep_top_k: int) -> list:
"""Compress long-term memory."""
return episodic[-keep_top_k:]
def _final_summary(self, memory: Dict) -> str:
"""Generate final summary."""
return f"Completed after {self.iteration} iterations."
Safety and Context Management
class ContextManager:
"""Manages token budget and context lifecycle."""
def __init__(self, max_tokens=8000):
self.max_tokens = max_tokens
self.current_tokens = 0
self.system_reminders = [
"You are a helpful AI coding assistant",
"Prioritize code quality and safety",
"Explain complex steps"
]
def add_system_reminder(self, reminder: str):
"""Add event-driven system reminder to counter instruction fade."""
if self._should_add_reminder():
self.system_reminders.append(reminder)
def _should_add_reminder(self) -> bool:
"""Decide if reminder should be injected."""
# Every 5 actions or when tokens > 70% budget
return self.current_tokens > 0.7 * self.max_tokens
def get_available_tokens(self) -> int:
"""Get remaining token budget."""
return max(0, self.max_tokens - self.current_tokens)
def drain_messages(self, messages: list) -> list:
"""Remove non-essential messages under budget pressure."""
if self.current_tokens > 0.9 * self.max_tokens:
# Keep system + last 10 messages
return messages[:1] + messages[-10:]
return messages
Practical Guidance
Hyperparameters:
- Token budget: 8000-16000 depending on model
- Context compaction threshold: 80% of budget
- Memory retention: keep last 10 actions + top-k important facts
- Thinking complexity threshold: 0.7
When to Apply:
- Complex terminal-based coding tasks
- Multi-step problems requiring reasoning
- Scenarios where cost optimization matters
- Tasks needing safety guarantees
When NOT to Apply:
- Simple single-command tasks (overhead not justified)
- Real-time latency-critical applications
- Scenarios with strict model availability
Key Pitfalls:
- Approval gates too strict—blocks legitimate operations
- Context compaction too aggressive—loses important facts
- Model mismatches—executor model too weak for complex tool calls
- Not tracking which operations are truly destructive
Integration Notes: Designed for terminal use; works with any LLM API; requires tool definitions and execution environments; approval gates can be automated with high-confidence predictions.
Evidence: Enables complex multi-step coding tasks with 60% fewer token costs than single-model approach; safety gates prevent 100% of destructive mistakes; 40% faster execution through model specialization.
Reference: https://arxiv.org/abs/2603.05344