07242683bf
Complete collection of AI agent skills including: - Frontend Development (Vue, React, Next.js, Three.js) - Backend Development (NestJS, FastAPI, Node.js) - Mobile Development (React Native, Expo) - Testing (E2E, frontend, webapp) - DevOps (GitHub Actions, CI/CD) - Marketing (SEO, copywriting, analytics) - Security (binary analysis, vulnerability scanning) - And many more... Synchronized from: https://skills.sh/ Co-Authored-By: Claude <noreply@anthropic.com>
388 lines
12 KiB
Python
388 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Ralph Integration for Brainstorming Skill
|
|
|
|
Automatically delegates complex tasks to RalphLoop for autonomous iteration.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional, Dict, Any
|
|
|
|
# Configuration
|
|
RALPHLOOP_CMD = Path(__file__).parent.parent.parent.parent / "obsidian-web-interface" / "bin" / "ralphloop"
|
|
COMPLEXITY_THRESHOLD = 5 # Minimum estimated steps to trigger Ralph
|
|
POLL_INTERVAL = 2 # Seconds between state checks
|
|
TIMEOUT = 3600 # Max wait time (1 hour) for complex tasks
|
|
|
|
|
|
def analyze_complexity(task_description: str, context: str = "") -> int:
|
|
"""
|
|
Analyze task complexity and return estimated number of steps.
|
|
|
|
Heuristics:
|
|
- Keyword detection for complex patterns
|
|
- Phrases indicating multiple phases
|
|
- Technical scope indicators
|
|
"""
|
|
task_lower = task_description.lower()
|
|
context_lower = context.lower()
|
|
|
|
complexity = 1 # Base complexity
|
|
|
|
# Keywords that increase complexity
|
|
complexity_keywords = {
|
|
# Architecture/System level (+3 each)
|
|
"architecture": 3, "system": 3, "platform": 3, "framework": 2,
|
|
"multi-tenant": 4, "distributed": 3, "microservices": 3,
|
|
|
|
# Data/Processing (+2 each)
|
|
"database": 2, "api": 2, "integration": 3, "pipeline": 3,
|
|
"real-time": 2, "async": 2, "streaming": 2, "monitoring": 2,
|
|
|
|
# Features (+1 each)
|
|
"authentication": 2, "authorization": 2, "security": 2,
|
|
"billing": 3, "payment": 2, "notifications": 1,
|
|
"dashboard": 1, "admin": 1, "reporting": 1,
|
|
|
|
# Phrases indicating complexity
|
|
"multi-step": 3, "end-to-end": 3, "full stack": 3,
|
|
"from scratch": 2, "complete": 2, "production": 2,
|
|
|
|
# Complete Pipeline Flow indicators (+4 each)
|
|
"complete chain": 4, "complete pipeline": 4, "real time logger": 4,
|
|
"real-time logger": 4, "automated qa": 4, "monitoring agent": 4,
|
|
"ai engineer second opinion": 4, "trigger ai assistant": 4,
|
|
}
|
|
|
|
# Count keywords
|
|
for keyword, weight in complexity_keywords.items():
|
|
if keyword in task_lower or keyword in context_lower:
|
|
complexity += weight
|
|
|
|
# Detect explicit complexity indicators
|
|
if "complex" in task_lower or "large scale" in task_lower:
|
|
complexity += 5
|
|
|
|
# Detect multiple requirements (lists, "and", "plus", "also")
|
|
if task_lower.count(',') > 2 or task_lower.count(' and ') > 1:
|
|
complexity += 2
|
|
|
|
# Detect implementation phases
|
|
phase_words = ["then", "after", "next", "finally", "subsequently"]
|
|
if sum(1 for word in phase_words if word in task_lower) > 1:
|
|
complexity += 2
|
|
|
|
return max(1, complexity)
|
|
|
|
|
|
def should_use_ralph(task_description: str, context: str = "") -> bool:
|
|
"""
|
|
Determine if task is complex enough to warrant RalphLoop.
|
|
|
|
Returns True if complexity exceeds threshold or user explicitly opts in.
|
|
"""
|
|
# Check for explicit opt-in via environment
|
|
if os.getenv("RALPH_AUTO", "").lower() in ("true", "1", "yes"):
|
|
return True
|
|
|
|
if os.getenv("BRAINSTORMING_USE_RALPH", "").lower() in ("true", "1", "yes"):
|
|
return True
|
|
|
|
# Check complexity
|
|
complexity = analyze_complexity(task_description, context)
|
|
return complexity >= COMPLEXITY_THRESHOLD
|
|
|
|
|
|
def create_ralph_task(task_description: str, context: str = "") -> str:
|
|
"""
|
|
Create a Ralph-formatted task prompt.
|
|
|
|
Returns the path to the created PROMPT.md file.
|
|
"""
|
|
ralph_dir = Path(".ralph")
|
|
ralph_dir.mkdir(exist_ok=True)
|
|
|
|
prompt_file = ralph_dir / "PROMPT.md"
|
|
|
|
# Format the task for Ralph with Complete Pipeline Flow
|
|
prompt_content = f"""# Task: {task_description}
|
|
|
|
## Context
|
|
{context}
|
|
|
|
## Complete Pipeline Flow
|
|
|
|
### Phase 1: Investigation & Analysis
|
|
- Thoroughly investigate the issue/codebase
|
|
- Identify all root causes
|
|
- Document findings with evidence
|
|
|
|
### Phase 2: Design with AI Engineer Review
|
|
- Propose comprehensive solution
|
|
- **MANDATORY**: Get AI Engineer's second opinion before coding
|
|
- Address all concerns raised
|
|
- Only proceed after design approval
|
|
|
|
### Phase 3: Implementation
|
|
- Follow approved design precisely
|
|
- Integrate real-time logging
|
|
- Monitor for errors during implementation
|
|
|
|
### Phase 4: Automated QA
|
|
- Use test-writer-fixer agent with:
|
|
- backend-architect review
|
|
- frontend-developer review
|
|
- ai-engineer double-check
|
|
- Fix any issues found
|
|
|
|
### Phase 5: Real-Time Monitoring
|
|
- Activate monitoring agent
|
|
- Catch issues in real-time
|
|
- Auto-trigger fixes to prevent repeating errors
|
|
|
|
## Success Criteria
|
|
|
|
The task is complete when:
|
|
- [ ] All requirements are understood and documented
|
|
- [ ] Root causes are identified with evidence
|
|
- [ ] Design/architecture is fully specified
|
|
- [ ] AI Engineer has reviewed and APPROVED the design
|
|
- [ ] Components and data flow are defined
|
|
- [ ] Error handling and edge cases are addressed
|
|
- [ ] Real-time logger is integrated
|
|
- [ ] Automated QA passes (all 3 agents)
|
|
- [ ] Testing strategy is outlined
|
|
- [ ] Implementation considerations are documented
|
|
- [ ] Monitoring agent is active
|
|
|
|
## Critical Rules
|
|
|
|
1. **AI Engineer Review REQUIRED**: Before ANY coding/execution, the AI Engineer agent MUST review and approve the design/approach. This is NON-NEGOTIABLE.
|
|
|
|
2. **Real-Time Logger**: Integrate comprehensive logging that:
|
|
- Logs all state transitions
|
|
- Tracks API calls and responses
|
|
- Monitors EventBus traffic
|
|
- Alerts on error patterns
|
|
- Provides live debugging capability
|
|
|
|
3. **Automated QA Pipeline**: After implementation completion:
|
|
- Run test-writer-fixer with backend-architect
|
|
- Run test-writer-fixer with frontend-developer
|
|
- Run test-writer-fixer with ai-engineer for double-check
|
|
- Fix ALL issues found before marking complete
|
|
|
|
4. **Real-Time Monitoring**: Activate monitoring that:
|
|
- Catches errors in real-time
|
|
- Auto-triggers AI assistant agent on failures
|
|
- Detects and solves issues immediately
|
|
- Prevents repeating the same errors
|
|
|
|
## Brainstorming Mode
|
|
|
|
You are in autonomous brainstorming mode. Your role is to:
|
|
1. Ask clarifying questions one at a time (simulate by making reasonable assumptions)
|
|
2. Explore 2-3 different approaches with trade-offs
|
|
3. Present the design in sections (200-300 words each)
|
|
4. Cover: architecture, components, data flow, error handling, testing
|
|
5. Validate the design against success criteria
|
|
|
|
## Instructions
|
|
|
|
- Follow the COMPLETE PIPELINE FLOW in order
|
|
- **NEVER skip AI Engineer review before coding**
|
|
- Iterate continuously until all success criteria are met
|
|
- When complete, add <!-- COMPLETE --> marker to this file
|
|
- Output the final validated design as markdown in iterations/final.md
|
|
"""
|
|
prompt_file.write_text(prompt_content)
|
|
|
|
return str(prompt_file)
|
|
|
|
|
|
def run_ralphloop(task_description: str, context: str = "",
|
|
max_iterations: Optional[int] = None,
|
|
max_runtime: Optional[int] = None) -> Dict[str, Any]:
|
|
"""
|
|
Run RalphLoop for autonomous task completion.
|
|
|
|
Returns a dict with:
|
|
- success: bool
|
|
- iterations: int
|
|
- output: str (final output)
|
|
- state: dict (Ralph's final state)
|
|
- error: str (if failed)
|
|
"""
|
|
print("🔄 Delegating to RalphLoop 'Tackle Until Solved' for autonomous iteration...")
|
|
print(f" Complexity: {analyze_complexity(task_description, context)} steps estimated")
|
|
print()
|
|
|
|
# Create Ralph task
|
|
prompt_path = create_ralph_task(task_description, context)
|
|
print(f"✅ Ralph task initialized: {prompt_path}")
|
|
print()
|
|
|
|
# Check if ralphloop exists
|
|
if not RALPHLOOP_CMD.exists():
|
|
return {
|
|
"success": False,
|
|
"error": f"RalphLoop not found at {RALPHLOOP_CMD}",
|
|
"iterations": 0,
|
|
"output": "",
|
|
"state": {}
|
|
}
|
|
|
|
# Build command
|
|
cmd = [str(RALPHLOOP_CMD)]
|
|
|
|
# Add inline task
|
|
cmd.append(task_description)
|
|
|
|
# Add optional parameters
|
|
if max_iterations:
|
|
cmd.extend(["--max-iterations", str(max_iterations)])
|
|
|
|
if max_runtime:
|
|
cmd.extend(["--max-runtime", str(max_runtime)])
|
|
|
|
# Environment variables
|
|
env = os.environ.copy()
|
|
env.setdefault("RALPH_AGENT", "claude")
|
|
env.setdefault("RALPH_MAX_ITERATIONS", str(max_iterations or 100))
|
|
|
|
print(f"Command: {' '.join(cmd)}")
|
|
print("=" * 60)
|
|
print()
|
|
|
|
# Run RalphLoop (synchronous for now)
|
|
try:
|
|
process = subprocess.Popen(
|
|
cmd,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
bufsize=1,
|
|
env=env
|
|
)
|
|
|
|
# Stream output
|
|
output_lines = []
|
|
for line in process.stdout:
|
|
print(line, end='', flush=True)
|
|
output_lines.append(line)
|
|
|
|
process.wait()
|
|
returncode = process.returncode
|
|
|
|
print()
|
|
print("=" * 60)
|
|
|
|
if returncode == 0:
|
|
# Read final state
|
|
state_file = Path(".ralph/state.json")
|
|
final_file = Path(".ralph/iterations/final.md")
|
|
|
|
state = {}
|
|
if state_file.exists():
|
|
state = json.loads(state_file.read_text())
|
|
|
|
final_output = ""
|
|
if final_file.exists():
|
|
final_output = final_file.read_text()
|
|
|
|
iterations = state.get("iteration", 0)
|
|
|
|
print(f"✅ Ralph completed in {iterations} iterations")
|
|
print()
|
|
|
|
return {
|
|
"success": True,
|
|
"iterations": iterations,
|
|
"output": final_output,
|
|
"state": state,
|
|
"error": None
|
|
}
|
|
else:
|
|
return {
|
|
"success": False,
|
|
"error": f"RalphLoop exited with code {returncode}",
|
|
"iterations": 0,
|
|
"output": "".join(output_lines),
|
|
"state": {}
|
|
}
|
|
|
|
except KeyboardInterrupt:
|
|
print()
|
|
print("⚠️ RalphLoop interrupted by user")
|
|
return {
|
|
"success": False,
|
|
"error": "Interrupted by user",
|
|
"iterations": 0,
|
|
"output": "",
|
|
"state": {}
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
"success": False,
|
|
"error": str(e),
|
|
"iterations": 0,
|
|
"output": "",
|
|
"state": {}
|
|
}
|
|
|
|
|
|
def delegate_to_ralph(task_description: str, context: str = "") -> Optional[str]:
|
|
"""
|
|
Main entry point: Delegate task to Ralph if complex, return None if should run directly.
|
|
|
|
If Ralph is used, returns the final output as a string.
|
|
If task is simple, returns None (caller should run directly).
|
|
"""
|
|
if not should_use_ralph(task_description, context):
|
|
return None
|
|
|
|
result = run_ralphloop(task_description, context)
|
|
|
|
if result["success"]:
|
|
return result["output"]
|
|
else:
|
|
print(f"❌ RalphLoop failed: {result.get('error', 'Unknown error')}")
|
|
print("Falling back to direct brainstorming mode...")
|
|
return None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Test the integration
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="Test Ralph integration")
|
|
parser.add_argument("task", help="Task description")
|
|
parser.add_argument("--context", default="", help="Additional context")
|
|
parser.add_argument("--force", action="store_true", help="Force Ralph mode")
|
|
parser.add_argument("--test-complexity", action="store_true", help="Only test complexity")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.test_complexity:
|
|
complexity = analyze_complexity(args.task, args.context)
|
|
print(f"Complexity: {complexity} steps")
|
|
print(f"Should use Ralph: {complexity >= COMPLEXITY_THRESHOLD}")
|
|
else:
|
|
if args.force:
|
|
os.environ["RALPH_AUTO"] = "true"
|
|
|
|
result = delegate_to_ralph(args.task, args.context)
|
|
|
|
if result:
|
|
print("\n" + "=" * 60)
|
|
print("FINAL OUTPUT:")
|
|
print("=" * 60)
|
|
print(result)
|
|
else:
|
|
print("\nTask not complex enough for Ralph. Running directly...")
|