building-agents-patterns

Original source / Raw SKILL.md

---
name: building-agents-patterns
description: Best practices, patterns, and examples for building goal-driven agents. Includes pause/resume architecture, hybrid workflows, anti-patterns, and handoff to testing. Use when optimizing agent design.
license: Apache-2.0
metadata:
  author: hive
  version: "1.0"
  type: reference
  part_of: building-agents
---

# Building Agents - Patterns & Best Practices

Design patterns, examples, and best practices for building robust goal-driven agents.

**Prerequisites:** Complete agent structure using `building-agents-construction`.

## Practical Example: Hybrid Workflow

How to build a node using both direct file writes and optional MCP validation:

```python
# 1. WRITE TO FILE FIRST (Primary - makes it visible)
node_code = '''
search_node = NodeSpec(
    id="search-web",
    node_type="llm_tool_use",
    input_keys=["query"],
    output_keys=["search_results"],
    system_prompt="Search the web for: {query}",
    tools=["web_search"],
)
'''

Edit(
    file_path="exports/research_agent/nodes/__init__.py",
    old_string="# Nodes will be added here",
    new_string=node_code
)

print("✅ Added search_node to nodes/__init__.py")
print("📁 Open exports/research_agent/nodes/__init__.py to see it!")

# 2. OPTIONALLY VALIDATE WITH MCP (Secondary - bookkeeping)
validation = mcp__agent-builder__test_node(
    node_id="search-web",
    test_input='{"query": "python tutorials"}',
    mock_llm_response='{"search_results": [...mock results...]}'
)

print(f"✓ Validation: {validation['success']}")
```

**User experience:**

- Immediately sees node in their editor (from step 1)
- Gets validation feedback (from step 2)
- Can edit the file directly if needed

This combines visibility (files) with validation (MCP tools).

## Pause/Resume Architecture

For agents needing multi-turn conversations with user interaction:

### Basic Pause/Resume Flow

```python
# Define pause nodes - execution stops at these nodes
pause_nodes = ["request-clarification", "await-approval"]

# Define entry points - where to resume from each pause
entry_points = {
    "start": "analyze-request",  # Initial entry
    "request-clarification_resume": "process-clarification",  # Resume from clarification
    "await-approval_resume": "execute-action",  # Resume from approval
}
```

### Example: Multi-Turn Research Agent

```python
# Nodes
nodes = [
    NodeSpec(id="analyze-request", ...),
    NodeSpec(id="request-clarification", ...),  # PAUSE NODE
    NodeSpec(id="process-clarification", ...),
    NodeSpec(id="generate-results", ...),
    NodeSpec(id="await-approval", ...),  # PAUSE NODE
    NodeSpec(id="execute-action", ...),
]

# Edges with resume flows
edges = [
    EdgeSpec(
        id="analyze-to-clarify",
        source="analyze-request",
        target="request-clarification",
        condition=EdgeCondition.CONDITIONAL,
        condition_expr="needs_clarification == true",
    ),
    # When resumed, goes to process-clarification
    EdgeSpec(
        id="clarify-to-process",
        source="request-clarification",
        target="process-clarification",
        condition=EdgeCondition.ALWAYS,
    ),
    EdgeSpec(
        id="results-to-approval",
        source="generate-results",
        target="await-approval",
        condition=EdgeCondition.ALWAYS,
    ),
    # When resumed, goes to execute-action
    EdgeSpec(
        id="approval-to-execute",
        source="await-approval",
        target="execute-action",
        condition=EdgeCondition.ALWAYS,
    ),
]

# Configuration
pause_nodes = ["request-clarification", "await-approval"]
entry_points = {
    "start": "analyze-request",
    "request-clarification_resume": "process-clarification",
    "await-approval_resume": "execute-action",
}
```

### Running Pause/Resume Agents

```python
# Initial run - will pause at first pause node
result1 = await agent.run(
    context={"query": "research topic"},
    session_state=None
)

# Check if paused
if result1.paused_at:
    print(f"Paused at: {result1.paused_at}")

    # Resume with user input
    result2 = await agent.run(
        context={"user_response": "clarification details"},
        session_state=result1.session_state  # Pass previous state
    )
```

## Anti-Patterns

### What NOT to Do

❌ **Don't rely on `export_graph`** - Write files immediately, not at end
```python
# BAD: Building in session state, exporting at end
mcp__agent-builder__add_node(...)
mcp__agent-builder__add_node(...)
mcp__agent-builder__export_graph()  # Files appear only now

# GOOD: Writing files immediately
Write(file_path="...", content=node_code)  # File visible now
Write(file_path="...", content=node_code)  # File visible now
```

❌ **Don't hide code in session** - Write to files as components approved
```python
# BAD: Accumulating changes invisibly
session.add_component(component1)
session.add_component(component2)
# User can't see anything yet

# GOOD: Incremental visibility
Edit(file_path="...", ...)  # User sees change 1
Edit(file_path="...", ...)  # User sees change 2
```

❌ **Don't wait to write files** - Agent visible from first step
```python
# BAD: Building everything before writing
design_all_nodes()
design_all_edges()
write_everything_at_once()

# GOOD: Write as you go
write_package_structure()  # Visible
write_goal()  # Visible
write_node_1()  # Visible
write_node_2()  # Visible
```

❌ **Don't batch everything** - Write incrementally
```python
# BAD: Batching all nodes
nodes = [design_node_1(), design_node_2(), ...]
write_all_nodes(nodes)

# GOOD: One at a time with user feedback
write_node_1()  # User approves
write_node_2()  # User approves
write_node_3()  # User approves
```

### MCP Tools - Correct Usage

**MCP tools OK for:**
✅ `test_node` - Validate node configuration with mock inputs
✅ `validate_graph` - Check graph structure
✅ `create_session` - Track session state for bookkeeping
✅ Other validation tools

**Just don't:** Use MCP as the primary construction method or rely on export_graph

## Best Practices

### 1. Show Progress After Each Write

```python
# After writing a node
print("✅ Added analyze_request_node to nodes/__init__.py")
print("📊 Progress: 1/6 nodes added")
print("📁 Open exports/my_agent/nodes/__init__.py to see it!")
```

### 2. Let User Open Files During Build

```python
# Encourage file inspection
print("✅ Goal written to agent.py")
print("")
print("💡 Tip: Open exports/my_agent/agent.py in your editor to see the goal!")
```

### 3. Write Incrementally - One Component at a Time

```python
# Good flow
write_package_structure()
show_user("Package created")

write_goal()
show_user("Goal written")

for node in nodes:
    get_approval(node)
    write_node(node)
    show_user(f"Node {node.id} written")
```

### 4. Test As You Build

```python
# After adding several nodes
print("💡 You can test current state with:")
print("  PYTHONPATH=core:exports python -m my_agent validate")
print("  PYTHONPATH=core:exports python -m my_agent info")
```

### 5. Keep User Informed

```python
# Clear status updates
print("🔨 Creating package structure...")
print("✅ Package created: exports/my_agent/")
print("")
print("📝 Next: Define agent goal")
```

## Continuous Monitoring Agents

For agents that run continuously without terminal nodes:

```python
# No terminal nodes - loops forever
terminal_nodes = []

# Workflow loops back to start
edges = [
    EdgeSpec(id="monitor-to-check", source="monitor", target="check-condition"),
    EdgeSpec(id="check-to-wait", source="check-condition", target="wait"),
    EdgeSpec(id="wait-to-monitor", source="wait", target="monitor"),  # Loop
]

# Entry node only
entry_node = "monitor"
entry_points = {"start": "monitor"}
pause_nodes = []
```

**Example: File Monitor**

```python
nodes = [
    NodeSpec(id="list-files", ...),
    NodeSpec(id="check-new-files", node_type="router", ...),
    NodeSpec(id="process-files", ...),
    NodeSpec(id="wait-interval", node_type="function", ...),
]

edges = [
    EdgeSpec(id="list-to-check", source="list-files", target="check-new-files"),
    EdgeSpec(
        id="check-to-process",
        source="check-new-files",
        target="process-files",
        condition=EdgeCondition.CONDITIONAL,
        condition_expr="new_files_count > 0",
    ),
    EdgeSpec(
        id="check-to-wait",
        source="check-new-files",
        target="wait-interval",
        condition=EdgeCondition.CONDITIONAL,
        condition_expr="new_files_count == 0",
    ),
    EdgeSpec(id="process-to-wait", source="process-files", target="wait-interval"),
    EdgeSpec(id="wait-to-list", source="wait-interval", target="list-files"),  # Loop back
]

terminal_nodes = []  # No terminal - runs forever
```

## Complex Routing Patterns

### Multi-Condition Router

```python
router_node = NodeSpec(
    id="decision-router",
    node_type="router",
    input_keys=["analysis_result"],
    output_keys=["decision"],
    system_prompt="""
    Based on the analysis result, decide the next action:
    - If confidence > 0.9: route to "execute"
    - If 0.5 <= confidence <= 0.9: route to "review"
    - If confidence < 0.5: route to "clarify"

    Return: {"decision": "execute|review|clarify"}
    """,
)

# Edges for each route
edges = [
    EdgeSpec(
        id="router-to-execute",
        source="decision-router",
        target="execute-action",
        condition=EdgeCondition.CONDITIONAL,
        condition_expr="decision == 'execute'",
        priority=1,
    ),
    EdgeSpec(
        id="router-to-review",
        source="decision-router",
        target="human-review",
        condition=EdgeCondition.CONDITIONAL,
        condition_expr="decision == 'review'",
        priority=2,
    ),
    EdgeSpec(
        id="router-to-clarify",
        source="decision-router",
        target="request-clarification",
        condition=EdgeCondition.CONDITIONAL,
        condition_expr="decision == 'clarify'",
        priority=3,
    ),
]
```

## Error Handling Patterns

### Graceful Failure with Fallback

```python
# Primary node with error handling
nodes = [
    NodeSpec(id="api-call", max_retries=3, ...),
    NodeSpec(id="fallback-cache", ...),
    NodeSpec(id="report-error", ...),
]

edges = [
    # Success path
    EdgeSpec(
        id="api-success",
        source="api-call",
        target="process-results",
        condition=EdgeCondition.ON_SUCCESS,
    ),
    # Fallback on failure
    EdgeSpec(
        id="api-to-fallback",
        source="api-call",
        target="fallback-cache",
        condition=EdgeCondition.ON_FAILURE,
        priority=1,
    ),
    # Report if fallback also fails
    EdgeSpec(
        id="fallback-to-error",
        source="fallback-cache",
        target="report-error",
        condition=EdgeCondition.ON_FAILURE,
        priority=1,
    ),
]
```

## Performance Optimization

### Parallel Node Execution

```python
# Use multiple edges from same source for parallel execution
edges = [
    EdgeSpec(
        id="start-to-search1",
        source="start",
        target="search-source-1",
        condition=EdgeCondition.ALWAYS,
    ),
    EdgeSpec(
        id="start-to-search2",
        source="start",
        target="search-source-2",
        condition=EdgeCondition.ALWAYS,
    ),
    EdgeSpec(
        id="start-to-search3",
        source="start",
        target="search-source-3",
        condition=EdgeCondition.ALWAYS,
    ),
    # Converge results
    EdgeSpec(
        id="search1-to-merge",
        source="search-source-1",
        target="merge-results",
    ),
    EdgeSpec(
        id="search2-to-merge",
        source="search-source-2",
        target="merge-results",
    ),
    EdgeSpec(
        id="search3-to-merge",
        source="search-source-3",
        target="merge-results",
    ),
]
```

## Handoff to Testing

When agent is complete, transition to testing phase:

```python
print("""
✅ Agent complete: exports/my_agent/

Next steps:
1. Switch to testing-agent skill
2. Generate and approve tests
3. Run evaluation
4. Debug any failures

Command: "Test the agent at exports/my_agent/"
""")
```

### Pre-Testing Checklist

Before handing off to testing-agent:

- [ ] Agent structure validates: `python -m agent_name validate`
- [ ] All nodes defined in nodes/__init__.py
- [ ] All edges connect valid nodes
- [ ] Entry node specified
- [ ] Agent can be imported: `from exports.agent_name import default_agent`
- [ ] README.md with usage instructions
- [ ] CLI commands work (info, validate)

## Related Skills

- **building-agents-core** - Fundamental concepts
- **building-agents-construction** - Step-by-step building
- **testing-agent** - Test and validate agents
- **agent-workflow** - Complete workflow orchestrator

---

**Remember: Agent is actively constructed, visible the whole time. No hidden state. No surprise exports. Just transparent, incremental file building.**