Current agent design broken. Industry split: loops vs graphs.
Loops give flexibility. Graphs give control.
Both fail in production. Loops cause infinite runs, state drift, high cost. Graphs break on unexpected input.
Solution: Use compiler design.
The Loop: Ergonomics and Chaos
ReAct pattern is classic loop. LLM reads prompt, runs tool, observes, repeats.
Easy to prototype. Fast setup.
Basic loop agent:
def run_agent_loop(user_query: str, tools: dict):
context = f"User: {user_query}\n"
while True:
response = call_llm(context + "Choose next action or output final answer.")
if "FINAL_ANSWER" in response:
return parse_answer(response)
tool_name, tool_input = parse_action(response)
tool_output = tools[tool_name](tool_input)
context += f"\nAction: {tool_name}({tool_input}) -> {tool_output}\n"Runtime behavior is black box. Bad tool output breaks LLM. Agent loops infinitely.
Context grows every step. Agent loses plan, forgets intent, hallucinates. Testing hard. Uncontrolled loops often lead to developer fatigue, prompting some to fire their AI assistant due to context drift.
The Graph: Rigid Determinism
Graphs and state machines fix loop chaos. Nodes run code or LLM. Edges define paths.
Guarantees validation, enforces auth.
Conceptual graph setup:
class AgentState:
def __init__(self, query: str):
self.query = query
self.results = []
self.validated = False
def validate_node(state: AgentState):
state.validated = check_query_safety(state.query)
return state
def search_node(state: AgentState):
if not state.validated:
raise ValueError("Unauthorized query")
state.results = db_search(state.query)
return state
# Constructing the graph
workflow = Workflow()
workflow.add_node("validate", validate_node)
workflow.add_node("search", search_node)
workflow.set_entry_point("validate")
workflow.add_edge("validate", "search")Graphs give control but require high boilerplate.
Loses LLM adaptability. Unexpected queries fail. Graph becomes hard-coded decision tree. Manual approval gates do not solve this; research shows humans miss security threats in agent commands when validating actions.
Core Conflict: Flexibility vs Static Analysis
Runtime paths give fast prototyping. Static paths give safety, optimization, predictability.
Need both. Developer writes flexible code, engine runs optimized plan.
Use compilation.
Why Compilers Bridge Gap
Agent compiler takes flexible definition, outputs optimized execution graph.
Like traditional compiler converting AST to machine code. Agent compiler traces paths, optimizes prompts, generates state machine.
Example: DSPy. Developer defines signatures. Compiler optimizes prompts against dataset, builds pipeline.
Separates intent from execution.
Toy Agent Compiler
Python example. Compiler takes step list, runs safety checks, optimizes path, outputs state machine.
Target execution graph:
import json
from typing import Dict, List, Any, Callable
class Node:
def __init__(self, name: str, action: Callable[[Dict], Dict]):
self.name = name
self.action = action
class CompiledGraph:
def __init__(self):
self.nodes: Dict[str, Node] = {}
self.edges: Dict[str, str] = {}
self.conditional_edges: Dict[str, Callable[[Dict], str]] = {}
def add_node(self, name: str, action: Callable[[Dict], Dict]):
self.nodes[name] = Node(name, action)
def add_edge(self, from_node: str, to_node: str):
self.edges[from_node] = to_node
def add_conditional_edge(self, from_node: str, router: Callable[[Dict], str]):
self.conditional_edges[from_node] = router
def execute(self, initial_state: Dict) -> Dict:
state = initial_state
current = "start"
visited = set()
while current in self.nodes:
# Prevent infinite execution loops at runtime
state_key = f"{current}:{hash(json.dumps(state, sort_keys=True))}"
if state_key in visited:
print(f"Loop detected at node '{current}'. Aborting execution.")
break
visited.add(state_key)
print(f"Executing node: {current}")
state = self.nodes[current].action(state)
if current in self.conditional_edges:
current = self.conditional_edges[current](state)
else:
current = self.edges.get(current, "end")
return stateCompiler code:
class AgentCompiler:
def __init__(self, steps: List[Dict[str, Any]]):
self.steps = steps
def compile(self) -> CompiledGraph:
graph = CompiledGraph()
# Automatically inject a safety validation node at the start
graph.add_node("start", self._create_validation_node())
previous_node = "start"
for i, step in enumerate(self.steps):
node_name = step["name"]
action_type = step["type"]
# Optimize: Skip redundant search steps if they happen consecutively
if i > 0 and action_type == "search" and self.steps[i-1]["type"] == "search":
print(f"Compiler Optimization: Merging redundant search step '{node_name}'")
continue
if action_type == "llm":
node_action = self._build_llm_action(step["prompt"])
elif action_type == "search":
node_action = self._build_search_action()
else:
node_action = lambda state: state
graph.add_node(node_name, node_action)
graph.add_edge(previous_node, node_name)
previous_node = node_name
# Add transition to end
graph.add_edge(previous_node, "end")
return graph
def _create_validation_node(self) -> Callable[[Dict], Dict]:
def validate(state: Dict) -> Dict:
query = state.get("query", "")
if "drop table" in query.lower() or "delete" in query.lower():
raise ValueError("SQL injection risk detected in input.")
state["validated"] = True
return state
return validate
def _build_llm_action(self, prompt: str) -> Callable[[Dict], Dict]:
def llm_action(state: Dict) -> Dict:
# Simulated LLM call
state["response"] = f"Processed with prompt '{prompt}': {state.get('query')}"
return state
return llm_action
def _build_search_action(self) -> Callable[[Dict], Dict]:
def search_action(state: Dict) -> Dict:
state["search_results"] = ["result_1", "result_2"]
return state
return search_actionExecution:
# Define high-level steps (our source code)
steps_definition = [
{"name": "fetch_data", "type": "search"},
{"name": "refetch_data", "type": "search"}, # Redundant step
{"name": "generate_report", "type": "llm", "prompt": "Summarize data"}
]
compiler = AgentCompiler(steps_definition)
compiled_agent = compiler.compile()
# Run the compiled graph
initial_state = {"query": "Find latest sales metrics"}
final_state = compiled_agent.execute(initial_state)
print("Final State:", final_state)Compiler injected validation node, removed redundant search step. Output is clean, predictable path.
Prompt Optimization via Compilation
Hard-coded prompts break when LLM changes. To prevent regressions, teams must run a structured model upgrade evaluation before switching production models.
Compiler treats prompt as code. Optimizes prompt instructions and few-shot examples against dataset for target model. This shift toward minimal, optimized instructions mirrors how Anthropic slashed Claude Code's system prompt to improve execution efficiency.
Turns prompt engineering into structured process.
Static Analysis
Raw loops cannot guarantee termination. Compiled graphs allow static checks.
Checks:
- Unreachable Nodes: Nodes never executed.
- Sink Nodes: Nodes with no exit path.
- Type Mismatches: Output schema violates next node input schema.
Catches bugs before production.
Static analysis implementation:
def analyze_graph(graph: CompiledGraph) -> Dict[str, List[str]]:
errors = {"unreachable": [], "sinks": []}
all_nodes = set(graph.nodes.keys())
visited = set()
def traverse(node_name: str):
if node_name in visited or node_name == "end":
return
visited.add(node_name)
# Check standard edge
next_node = graph.edges.get(node_name)
if next_node:
traverse(next_node)
# Check conditional edges
if node_name in graph.conditional_edges:
# Static analysis assumes all paths possible
pass
traverse("start")
unreachable = all_nodes - visited
errors["unreachable"] = list(unreachable)
for node in graph.nodes:
if node != "end" and node not in graph.edges and node not in graph.conditional_edges:
errors["sinks"].append(node)
return errorsRun analysis before execution. Block deployment if errors found.
Architecture Comparison
| Metric | Loop Agents (ReAct) | Graph Agents (DAGs) | Compiled Agents |
|---|---|---|---|
| Developer Ergonomics | High (Write simple code) | Low (Write boilerplate) | High (Write code, compiler builds graph) |
| Predictability | Low (State drift, loops) | High (Defined paths) | High (Validated paths) |
| Adaptability | High (LLM decides paths) | Low (Hard-coded paths) | Medium-High (Optimized paths) |
| Cost Control | Hard (Uncapped iterations) | Easy (Set max steps) | Easy (Statically bounded) |
Intermediate Representation (IR)
Scale requires IR. Acts as platform-independent layer between source and target.
Agent IR defines state schemas, prompts, tools, transitions.
Benefits:
- Port agents to different LLMs without rewriting prompts.
- Run static checks for dead ends or infinite loops.
- Apply optimization passes (merge calls, parallelize tools).
Developer targets IR. Runtime executes IR, handles state, recovers from errors.
Example IR definition:
{
"metadata": {
"agent_id": "report_generator",
"version": "1.0.0"
},
"state_schema": {
"query": "string",
"validated": "boolean",
"search_results": "array",
"response": "string"
},
"nodes": [
{ "name": "start", "type": "validation", "config": {} },
{ "name": "fetch_data", "type": "search", "config": {} },
{ "name": "generate_report", "type": "llm", "config": { "prompt": "Summarize data" } }
],
"edges": [
{ "from": "start", "to": "fetch_data" },
{ "from": "fetch_data", "to": "generate_report" },
{ "from": "generate_report", "to": "end" }
]
}IR decouples agent definition from runtime engine. Allows target-specific optimizations.
Practical Engineering Patterns
State Hydration and Resiliency
Persist state to database at every node transition. If LLM fails, resume from last checkpoint. Prevents data loss, saves tokens.
State hydration implementation:
import sqlite3
import pickle
class DbStateStore:
def __init__(self, db_path: str = ":memory:"):
self.conn = sqlite3.connect(db_path)
self.conn.execute(
"CREATE TABLE IF NOT EXISTS states (session_id TEXT PRIMARY KEY, state BLOB)"
)
self.conn.commit()
def save(self, session_id: str, state: dict):
data = pickle.dumps(state)
self.conn.execute(
"INSERT OR REPLACE INTO states (session_id, state) VALUES (?, ?)",
(session_id, data)
)
self.conn.commit()
def load(self, session_id: str) -> dict:
cursor = self.conn.execute(
"SELECT state FROM states WHERE session_id = ?",
(session_id,)
)
row = cursor.fetchone()
return pickle.loads(row[0]) if row else {}Integrate store into runtime loop. Save state after node execution. Load state on failure recovery.
Static Validation of LLM Outputs
Compile schemas into prompts using Pydantic. Compiler guarantees output matches input schema of next node. Route validation failures to correction nodes.
Dynamic Routing with Static Fallbacks
Restrict LLM routing choices to statically compiled transitions. Prevents invalid state transitions.
Just-In-Time (JIT) Agent Compilation
Future: JIT compilers for agents.
JIT compiler analyzes query at runtime, builds task-specific graph, runs it, discards it.
Combines loop flexibility with graph safety.
