agentic-workflow

name	agentic-workflow
description	Design and implement LLM-powered agentic workflows with tool use, planning, memory, and multi-agent coordination. Outputs agent architecture, tool definitions, evaluation framework, and safety controls.
argument-hint	["task type","tools available","autonomy level","human-in-the-loop requirements"]
allowed-tools	Read, Write, Bash

Agentic Workflow Design

Agentic workflows let LLMs take sequences of actions — calling tools, searching the web, writing code, coordinating with other agents — to complete complex tasks autonomously. The design challenge is balancing autonomy (getting things done without hand-holding) with reliability (not taking wrong or harmful actions).

Process

1. Define the task and scope. What does the agent accomplish? What is explicitly out of scope? What can it never do?
2. Identify required tools. Each tool does one thing. Tools are the agent's interface to the world.
3. Choose the architecture. Single agent with tools, multi-agent with orchestrator, or hierarchical. Start simple.
4. Design the planning loop. ReAct (Reason + Act), plan-then-execute, or reflection loops.
5. Implement memory. Working memory (context window), episodic memory (conversation history), semantic memory (vector store).
6. Define human-in-the-loop checkpoints. What actions require approval? What is irreversible?
7. Build evaluation. Automated tests for tool calling accuracy, task completion, and safety.
8. Add guardrails. Input/output filters, action allowlists, cost limits, iteration caps.

Architecture Patterns

Pattern 1: Single Agent + Tools (start here)
  User → Agent → [search, code_exec, file_write, api_call] → Result
  Best for: Well-defined tasks, single domain, <10 tool calls

Pattern 2: Orchestrator + Specialist Agents  
  User → Orchestrator → [ResearchAgent, WriterAgent, ReviewerAgent] → Result
  Best for: Complex multi-step tasks, parallel work streams

Pattern 3: Hierarchical (Manager → Workers)
  Manager Agent
    ├── SubAgent A (research)
    ├── SubAgent B (analysis)
    └── SubAgent C (drafting)
  Best for: Large parallelisable tasks

Pattern 4: Reflection / Critique Loop
  Agent → Draft → Critic Agent → Revise → Final
  Best for: High-quality output requirements (code, reports, plans)

Tool Definition

from anthropic import Anthropic
from typing import Any
import json

client = Anthropic()

# Tool schema — clear name, description, and typed parameters
TOOLS = [
    {
        "name": "search_web",
        "description": "Search the web for current information. Use when you need facts, news, or data not in your training.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query. Be specific and use key terms."
                },
                "max_results": {
                    "type": "integer",
                    "description": "Number of results to return (1-10)",
                    "default": 5
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "run_python",
        "description": "Execute Python code in a sandboxed environment. Use for calculations, data processing, or generating outputs. Returns stdout and any errors.",
        "input_schema": {
            "type": "object",
            "properties": {
                "code": {
                    "type": "string",
                    "description": 
                },
                : {
                    : ,
                    : ,
                    : 
                }
            },
            : []
        }
    },
    {
        : ,
        : ,
        : {
            : ,
            : {
                : {: , : },
                : {: , : },
                : {: , : [, ], : }
            },
            : [, ]
        }
    }
]

ReAct Agent Loop

import subprocess
import tempfile
import os
from typing import Optional

def execute_tool(tool_name: str, tool_input: dict) -> str:
    """Execute a tool and return the result as a string."""
    if tool_name == "search_web":
        # Integration with search API
        results = search_api.search(tool_input["query"], 
                                    n=tool_input.get("max_results", 5))
        return json.dumps(results)
    
    elif tool_name == "run_python":
        code = tool_input["code"]
        timeout = tool_input.get("timeout_seconds", 10)
        
        # Sandboxed execution
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            f.write(code)
            f.flush()
            try:
                result = subprocess.run(
                    ["python3", f.name],
                    capture_output=True, text=True, timeout=timeout,
                    cwd="/tmp"  # Restricted working directory
                )
                output = result.stdout
                if result.stderr: output += f"\nSTDERR: {result.stderr}"
                return output or "(no output)"
             subprocess.TimeoutExpired:
                 
            :
                os.unlink(f.name)
    
     tool_name == :
        
        path = os.path.basename(tool_input[])
        safe_path = os.path.join(, path)
        os.makedirs(os.path.dirname(safe_path), exist_ok=)
        mode =   tool_input.get() ==   
         (safe_path, mode)  f:
            f.write(tool_input[])
         
    
     


 () -> :
    
    messages = [{: , : task}]
    
    system_prompt = 
    
     iteration  (max_iterations):
        response = client.messages.create(
            model=,
            max_tokens=,
            system=system_prompt,
            tools=TOOLS,
            messages=messages,
        )
        
        
         response.stop_reason == :
            text_blocks = [b.text  b  response.content  (b, )]
             .join(text_blocks)
        
        
        messages.append({: , : response.content})
        tool_results = []
        
         block  response.content:
             block. == :
                ()
                result = execute_tool(block.name, block.)
                tool_results.append({
                    : ,
                    : block.,
                    : result,
                })
        
        messages.append({: , : tool_results})
    
     

Multi-Agent Orchestration

# Orchestrator that coordinates specialist agents
class ResearchOrchestrator:
    def __init__(self):
        self.client = Anthropic()
    
    def run(self, research_question: str) -> dict:
        # Phase 1: Plan
        plan = self._plan(research_question)
        
        # Phase 2: Parallel research (multiple sub-agents)
        import asyncio
        research_results = asyncio.run(
            self._parallel_research(plan["subtopics"])
        )
        
        # Phase 3: Synthesise
        synthesis = self._synthesise(research_question, research_results)
        
        # Phase 4: Review (critic agent)
        review = self._critique(synthesis)
        
        return {"synthesis": synthesis, "critique": review, "sources": research_results}
    
    def _plan(self, question: str) -> dict:
        response = self.client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=1024,
            messages=[{"role": "user", "content": f"""
                Break this research question into 3-5 focused subtopics to investigate:
                {question}
                
                Respond with JSON: {{"subtopics": ["subtopic1", "subtopic2", ...]}}
            """}]
        )
         json.loads(response.content[].text)
    
      () -> :
         asyncio
        tasks = [._research_subtopic(t)  t  subtopics]
          asyncio.gather(*tasks)
    
      () -> :
        
        result = run_agent()
         {: subtopic, : result}
    
     () -> :
        research_text = .join(
              r  research
        )
        response = .client.messages.create(
            model=,
            max_tokens=,
            messages=[{: , : }]
        )
         response.content[].text

Safety and Guardrails

class AgentGuardrails:
    """Enforce safety limits on agent execution."""
    
    FORBIDDEN_PATTERNS = [
        r'rm\s+-rf',          # Dangerous shell commands
        r'DROP\s+TABLE',       # Database destruction
        r'os\.system',         # Shell injection
        r'eval\(',             # Code injection
        r'__import__',         # Dynamic imports
    ]
    
    # Tools that require human approval before execution
    APPROVAL_REQUIRED = {"write_file", "send_email", "delete_record"}
    
    def __init__(self, max_cost_usd: float = 1.0, max_iterations: int = 20):
        self.max_cost = max_cost_usd
        self.max_iterations = max_iterations
        self.total_cost = 0.0
        self.iteration_count = 0
    
    def check_tool_call(self, tool_name: str, tool_input: dict) -> tuple[bool, str]:
        """Returns (allowed, reason). Called before each tool execution."""
        # Iteration cap
        self.iteration_count += 1
        if self.iteration_count > .max_iterations:
             , 
        
        
         .total_cost > .max_cost:
             , 
        
        
         tool_name == :
            code = tool_input.get(, )
             re
             pattern  .FORBIDDEN_PATTERNS:
                 re.search(pattern, code, re.IGNORECASE):
                     , 
        
        
         tool_name  .APPROVAL_REQUIRED:
            approved = ._request_human_approval(tool_name, tool_input)
              approved:
                 , 
        
         , 
    
     () -> :
        ()
        ()
        ()
        response = ()
         response.lower() == 

Evaluation Framework

import json
from dataclasses import dataclass

@dataclass
class AgentEvalCase:
    task: str
    expected_tools: list[str]    # Tools that should be called
    forbidden_tools: list[str]   # Tools that must NOT be called
    output_contains: list[str]   # Strings that must appear in output
    max_iterations: int = 10

EVAL_SUITE = [
    AgentEvalCase(
        task="What is 2^32?",
        expected_tools=["run_python"],
        forbidden_tools=["search_web"],
        output_contains=["4294967296"],
    ),
    AgentEvalCase(
        task="What was the closing price of AAPL yesterday?",
        expected_tools=["search_web"],
        forbidden_tools=[],
        output_contains=["AAPL", "$"],
        max_iterations=5,
    ),
]

def evaluate_agent(eval_cases: list[AgentEvalCase]) -> dict:
    results = []
    for case in eval_cases:
        tools_called = []
        
        # Patch tool executor to track calls
        original_execute = execute_tool
        def tracking_execute(name, inp):
            tools_called.append(name)
            return original_execute(name, inp)
        
        output = run_agent(case.task, max_iterations=case.max_iterations)
        
        score = {
            : .task,
            : (t  tools_called  t  .expected_tools),
            :  (t  tools_called  t  .forbidden_tools),
            : (s  output  s  .output_contains),
        }
        score[] = (score.values())
        results.append(score)
    
    pass_rate = (  r  results  r[]) / (results)
     {: pass_rate, : results}

Anti-Patterns to Avoid

Anti-Pattern	Problem	Fix
No iteration cap	Agent loops forever on hard problems	Always set max_iterations; default 20
Tools that do too much	Hard to control, debug, test	One tool = one action; compose at agent level
No human-in-the-loop for irreversible actions	Agent deletes data, sends emails without oversight	Approval gates for destructive/external actions
Unbounded cost	Agent spins up 50 searches on one task	Per-run cost cap with hard stop
No evaluation suite	Can't measure if agent improvements help or hurt	Eval suite from day one
Agentic for simple tasks	Using 20 tool calls for something a single prompt handles	Single prompt first; agents only when necessary
No sandboxing for code execution	Agent runs arbitrary code on host system	Isolated container or subprocess with limited permissions

10 Rules

1. Start with the simplest architecture — single agent + tools — before building multi-agent systems.
2. Every agent run has an iteration cap and cost cap. No exceptions.
3. Tools do one thing. Composition happens at the agent level, not within tools.
4. Irreversible actions (delete, send, write to production) require human approval gates.
5. Code execution is always sandboxed — never run agent-generated code on the host system.
6. Build an evaluation suite before optimising — you need to measure to improve.
7. Prompts are configuration — version-control them, treat changes as deployments.
8. Log every tool call with inputs and outputs — you need the trace to debug failures.
9. Agents fail in long tail ways — test adversarial inputs, edge cases, and ambiguous tasks.
10. The best agent is the one that completes the task with the fewest tool calls — efficiency is a quality signal.