aetherlang-karpathy-upgrades

# AetherLang Karpathy Agent Upgrades A battle-tested skill for implementing 10 advanced AI agent node types that transform any simple LLM pipeline into a full autonomous AI agent framework. Based on Andrej Karpathy's vision of AI systems that think, compute, reflect, and act autonomously. Built and validated in production on the AetherLang Omega platform (neurodoc.app), these upgrades have been tested with real API calls, Docker deployments, and live traffic. ## The 10 Karpathy Upgrades | # | Node Type | Capability | What It Does | |---|-----------|-----------|--------------| | 1 | `plan` | Self-Programming | LLM generates sub-flows dynamically, then runtime compiles and executes them | | 2 | `code_interpreter` | Real Computation | Sandboxed Python execution with safe imports, no more hallucinated math | | 3 | `critique` | Self-Improvement | Evaluates output quality (0-10), retries up to 3x with feedback until threshold met | | 4 | `router` | Intelligent Branching | LLM picks optimal processing path, skips unselected routes (10x speedup) | | 5 | `ensemble` | Multi-Agent Synthesis | Runs multiple AI personas in parallel, synthesizes superior combined response | | 6 | `memory` | Persistent State | Store/recall/search data across executions with namespace isolation | | 7 | `tool` | External API Access | Call any REST API (GET/POST/PUT/DELETE) with security limits | | 8 | `loop` | Iterative Execution | Repeat any node over multiple items with collect or chain modes | | 9 | `transform` | Data Reshaping | Template, extract, format, or LLM-powered data transformation between nodes | | 10 | `parallel` | Concurrent Execution | Run multiple nodes simultaneously (3 API calls in 0.2s instead of 3s) | ## Architecture Pattern Every upgrade follows the same 3-step implementation pattern. This is universal and works for any Python-based DSL runtime. ### Step 1: Parser — Add Node Type to Enum ```python class NodeType(Enum): LLM = "llm" # ... existing types ... PLAN = "plan" CODE_INTERPRETER = "code_interpreter" CRITIQUE = "critique" ROUTER = "router" ENSEMBLE = "ensemble" MEMORY = "memory" TOOL = "tool" LOOP = "loop" TRANSFORM = "transform" PARALLEL = "parallel" ``` ### Step 2: Runtime Dispatch — Route to Handler Add an `elif` branch in the main node execution dispatcher: ```python async def _execute_node(self, ctx, node): if node.node_type == NodeType.LLM: result = await self._execute_llm(ctx, node, data) elif node.node_type == NodeType.PLAN: result = await self._execute_plan(ctx, node, data) elif node.node_type == NodeType.CODE_INTERPRETER: result = await self._execute_code_interpreter(ctx, node, data) # ... etc for all 10 types ``` ### Step 3: Runtime Method — Implement Logic Add the `async def _execute_<type>()` method to the runtime class. **CRITICAL WARNING**: In Python, if a method name appears twice in a class, the LAST definition silently wins. Always check for and remove old stubs before inserting new methods: ```bash grep -c "async def _execute_plan" runtime.py # Must be exactly 1 ``` ## Implementation Order and Dependencies Implement in this order due to dependencies: 1. **PLAN** — No dependencies, foundational self-programming 2. **CODE_INTERPRETER** — No dependencies, standalone sandboxed execution 3. **CRITIQUE** — Uses `_execute_node()` for retries, needs existing node execution 4. **ROUTER** — Uses `_execute_node()` for selected routes, needs skip logic in main loop 5. **ENSEMBLE** — Uses `asyncio.gather()` for parallel agent execution 6. **MEMORY** — File-based JSON persistence, fully independent 7. **TOOL** — Needs HTTP client (httpx preferred over aiohttp in Docker) 8. **LOOP** — Uses `_execute_node()` for iterating target nodes 9. **TRANSFORM** — Independent, JSON parsing and LLM reformatting 10. **PARALLEL** — Uses `asyncio.gather()`, needs skip logic like ROUTER --- ## Upgrade #1: PLAN — Self-Programming Compiler The PLAN node lets the AI write its own processing pipeline dynamically. **Parameters**: - `steps` (int): Number of sub-steps to generate, default 3 **How It Works**: 1. Send query to LLM asking it to break the task into N sequential steps 2. LLM returns JSON: `[{"step": 1, "action": "description", "type": "llm|code|search"}]` 3. For each step, create a temporary node and execute it 4. Collect all step results and combine into final response **System Prompt for Plan Generation**: ``` Break this task into exactly {steps} sequential steps. Return ONLY a JSON array: [{"step": 1, "action": "what to do", "type": "llm"}] Valid types: llm (text generation), code (calculation), search (information lookup) ``` **Implementation**: ```python async def _execute_plan(self, ctx, node, data): query = data["inputs"].get("query", "") steps = int(node.params.get("steps", "3")) # 1. Generate plan via LLM plan_response = await self.openai_client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": plan_system_prompt}, {"role": "user", "content": query} ], temperature=0.3 ) plan = json.loads(plan_response.choices[0].message.content) # 2. Execute each step results = [] for step in plan: step_result = await self._execute_plan_step(ctx, step, query, results) results.append(step_result) # 3. Combine results combined = "\n\n".join([f"Step {r['step']}: {r['result']}" for r in results]) return {"response": combined, "plan_steps": plan, "node_type": "plan"} ``` --- ## Upgrade #2: CODE_INTERPRETER — Sandboxed Python Executes real Python code for accurate calculations instead of LLM hallucinations. **Safety Configuration**: ```python FORBIDDEN_OPERATIONS = ['__import__', 'exec', 'eval', 'compile', 'open', 'subprocess', 'os', 'sys', 'socket', 'shutil'] ALLOWED_IMPORTS = {'math': math, 'statistics': statistics} SAFE_BUILTINS = {'abs': abs, 'round': round, 'min': min, 'max': max, 'sum': sum, 'len': len, 'range': range, 'int': int, 'float': float, 'str': str, 'bool': bool, 'list': list, 'dict': dict, 'print': print, 'isinstance': isinstance} CODE_TIMEOUT = 5 # seconds ``` **Implementation**: ```python async def _execute_code_interpreter(self, ctx, node, data): query = data["inputs"].get("query", "") # 1. LLM generates code code_response = await self.openai_client.chat.completions.create( model=node.params.get("model", "gpt-4o-mini"), messages=[ {"role": "system", "content": "Write Python code. Set variable `result` at the end. ONLY code, no markdown."}, {"role": "user", "content": query} ], temperature=0.2 ) code = code_response.choices[0].message.content.strip() code = code.replace("```python", "").replace("```", "").strip() # 2. Security check for forbidden in FORBIDDEN_OPERATIONS: if forbidden in code: return {"response": f"Blocked: {forbidden}", "node_type": "code_interpreter"} # 3. Sandboxed execution safe_globals = {"__builtins__": SAFE_BUILTINS, "math": math, "statistics": statistics} local_vars = {} with ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(exec, code, safe_globals, local_vars) future.result(timeout=CODE_TIMEOUT) result = local_vars.get("result", "No result variable set") return {"response": str(result), "code": code, "node_type": "code_interpreter"} ``` --- ## Upgrade #3: CRITIQUE — Self-Improvement Loop Evaluates output quality and retries with feedback until quality threshold is met. **Parameters**: - `threshold` (float): Minimum score 0-10, default 7 - `max_retries` (int): Max attempts, capped at hard limit of 3 - `criteria` (str): Evaluation criteria, default "accuracy, completeness, clarity" **Implementation Logic**: ``` 1. Get upstream output 2. For each attempt (0 to max_retries): a. Send output + criteria to evaluator LLM b. Parse: {score: 8.5, passed: true, feedback: "...", strengths: "..."} c. If score >= threshold → PASS, return output d. If max retries hit → return best attempt e. Else → re-execute source node with enhanced query containing feedback ``` **Enhanced Query Format for Retries**: ``` [Original query] [IMPROVEMENT FEEDBACK - attempt 2/3]: Score: 5.5/10 (need 7.0/10) Issues: Missing specific examples Keep: Good structure and flow Please improve your response addressing the feedback above. ``` --- ## Upgrade #4: ROUTER — Intelligent Branching LLM analyzes query and selects the optimal downstream node to execute, skipping all others. **Parameters**: - `routes` (str): Comma-separated "alias:description" pairs - `strategy` (str): "single" or "multi", default "single" **Critical Skip Logic** (required for performance — 10x speedup): Two parts must be implemented: **Part 1 — Main execution loop check**: ```python for node_alias in execution_order: if node_alias in ctx.node_outputs: ctx.log("SYSTEM", "INFO", f"Skip {node_alias} (already executed)") continue # ... execute node normally ``` **Part 2 — Router marks unselected routes**: ```python # After executing selected route(s): for route_alias in route_map: if route_alias not in selected: ctx.node_outputs[route_alias] = {"skipped_by_router": True, "response": ""} ``` **Result**: Query "What is 15% tip on $85.50?" routes to calculator node in 2.3s instead of executing all 3 routes in 21s. --- ## Upgrade #5: ENSEMBLE — Mixture of Agents Runs the same query through multiple AI personas in parallel, then synthesizes the best elements into a superior response. **Parameters**: - `agents` (str): Comma-separated "alias:persona" pairs - `synthesize` (str): "true" (default) or "false" to return all - `model` (str): Agent model, default "gpt-4o-mini" - `synth_model` (str): Synthesizer model **Safety**: MAX_ENSEMBLE_AGENTS = 7 **Implementation Pattern**: ```python async def _execute_ensemble(self, ctx, node, data): # 1. Parse agents agents = parse_agents(node.params.get("agents", "")) # 2. Run ALL agents in parallel async def run_agent(alias, persona): response = await self.openai_client.chat.completions.create( model=agent_model, messages=[ {"role": "system", "content": f"You are: {persona}. Respond from YOUR unique perspective."}, {"role": "user", "content": query} ], temperature=0.8 # High for diversity ) return {"alias": alias, "response": response.choices[0].message.content} results = await asyncio.gather(*[run_agent(a, p) for a, p in agents]) # 3. Synthesize synthesis = await self.openai_client.chat.completions.create( model=synth_model, messages=[{"role": "user", "content": f"Combine best insights from: {results}"}], temperature=0.5 ) return {"response": synthesis.choices[0].message.content, "node_type": "ensemble"} ``` **Real Test**: Moussaka recipe with French Chef + Greek Grandmother + Molecular Scientist = superior unified recipe. --- ## Upgrade #6: MEMORY — Persistent State File-based JSON storage for data that persists across flow executions. **Parameters**: - `namespace` (str): Memory bucket, default "default" - `action` (str): "store", "recall", "search", "clear", "auto" - `key` (str): Storage key, or LLM auto-extracts one - `ttl` (str): Hours until expiry, "0" = forever **Storage Format** (`memory_store/{namespace}.json`): ```json { "diet": { "value": "I am vegan and allergic to nuts", "_ts": 1708502400.0, "_query": "I am vegan and allergic to nuts" } } ``` **Actions**: - `store` — Save upstream response or query with key - `recall` — Return specific key or all entries as context - `search` — LLM semantic search over stored entries - `clear` — Delete key or entire namespace - `auto` — LLM decides whether to store or recall based on context --- ## Upgrade #7: TOOL — External API Access Call any REST API from within a flow with safety limits.