Managing context windows effectively is crucial for Ralph Orchestrator's success. AI agents have limited context windows, and exceeding them can cause failures or degraded performance.
# Bad - Too verboseCreate a comprehensive Python application that implements a calculator
with extensive error handling, logging capabilities, user-friendly
interface, and support for basic arithmetic operations including
addition, subtraction, multiplication, and division...
# Good - Concise and clearCreate a Python calculator with:
-Basic operations: +, -, *, /
-Error handling for division by zero
-Simple CLI interface
# .agent/config.json{"exclude_patterns":["*.pyc","__pycache__","*.log","test_*.py",# Exclude during implementation"docs/",# Exclude documentation".git/"# Never include git directory]}
# Instead of one large task"Build a complete web application"
# Break into phasesPhase 1: Create project structure
Phase 2: Implement data models
Phase 3: Add API endpoints
Phase 4: Build frontend
Phase 5: Add tests
defcreate_context_aware_checkpoint(iteration,context_usage):"""Create checkpoint when context is getting full"""ifcontext_usage['percentage_used']['current_agent']>70:# Reset context by creating checkpointcreate_checkpoint(iteration)# Clear working memoryclear_agent_memory()# Summarize progresscreate_progress_summary()
classContextManager:def__init__(self,max_history=5):self.history=[]self.max_history=max_historydefadd_iteration(self,prompt,response):"""Add iteration to history with sliding window"""self.history.append({'prompt':prompt,'response':response,'timestamp':time.time()})# Keep only recent historyiflen(self.history)>self.max_history:self.history.pop(0)defget_context(self):"""Get current context for agent"""# Include only recent iterationsreturn'\n'.join([f"Iteration {i+1}:\n{h['response'][:500]}..."fori,hinenumerate(self.history[-3:])])
defcompress_context(text,max_length=1000):"""Compress text while preserving key information"""iflen(text)<=max_length:returntext# Extract key sectionslines=text.split('\n')important_lines=[]forlineinlines:# Keep headers, errors, and key codeifany(markerinlineformarkerin['#','def ','class ','ERROR','TODO']):important_lines.append(line)compressed='\n'.join(important_lines)# If still too long, truncate with summaryiflen(compressed)>max_length:returncompressed[:max_length-20]+"\n... (truncated)"returncompressed
defmonitor_context_usage():"""Monitor and log context usage"""usage=estimate_context_usage('PROMPT.md',glob.glob('*.py'))# Log warning if approaching limitsforagent,percentageinusage['percentage_used'].items():ifpercentage>80:logging.warning(f"Context usage for {agent}: {percentage:.1f}% - "f"Consider optimization")# Save metricswithopen('.agent/metrics/context_usage.json','w')asf:json.dump(usage,f,indent=2)returnusage
# Instead of including full codeSee `calculator.py` for implementation details
# Reference specific sectionsRefer to lines 45-67 in `utils.py` for error handling
defcreate_iteration_summary(iteration_num):"""Create summary every N iterations"""ifiteration_num%10==0:summary={'completed':[],'in_progress':[],'pending':[],'issues':[]}# ... populate summarywithopen(f'.agent/summaries/summary_{iteration_num}.md','w')asf:f.write(format_summary(summary))
defrecover_from_context_overflow():"""Recover when context limits exceeded"""# 1. Save current statesave_state()# 2. Create summary of work donesummary=create_work_summary()# 3. Reset with minimal contextnew_prompt=f""" Continue from checkpoint. Previous work summary:{summary} Current task: {get_current_task()} """# 4. Resume with fresh contextreturnnew_prompt