AI Agent Orchestration: Designing Multi-Step Workflows with Tools, Memory, and Handoffs

Posted on Oct 19 • Originally published at veduis.com on Oct 19

Single-prompt AI interactions have a ceiling. Ask a model to summarize a document, straightforward. Ask it to research a competitor, extract key data, compare it against your product specs, draft a report, and email it to your sales team. That chain of steps requires something fundamentally different from a single LLM call.

AI agents bridge that gap. An agent is an LLM that can decide which actions to take, execute them using tools, observe the results, and decide what to do next, looping until it completes a goal or determines it needs human input. The jump from single-prompt to agentic systems is substantial in capability and equally substantial in complexity. Systems that work in demos break in production when tool calls fail, loops run indefinitely, or the model loses track of what it has already done.

I've built agent systems for research automation, customer onboarding pipelines, data extraction workflows, and competitive intelligence. The architectural decisions you make early (how to structure memory, how to handle failures, when to split work across multiple agents) determine whether these systems perform reliably or erratically. This guide covers those decisions in detail.

The first architectural decision is whether to use one agent or many.

A single agent receives a task, has access to a set of tools, and works toward completion by calling tools iteratively. Simple, observable, and sufficient for many use cases. A customer support agent that can query a knowledge base, check order status, and draft responses is a single agent. No coordination overhead, no handoff logic, easier to debug.

A multi-agent system decomposes work across specialized agents coordinated by an orchestrator. The orchestrator receives a high-level goal, breaks it into subtasks, delegates each subtask to a specialized agent, collects results, and synthesizes a final output. A research pipeline that uses one agent to search the web, another to read and extract key points from each page, and a third to write a synthesis is a multi-agent system.

The temptation to decompose everything into multiple specialized agents is strong but often counterproductive. Each handoff between agents is a potential failure point. I default to single-agent and only add agents when the single-agent approach demonstrably fails.

Tools are functions an agent can call. The quality of your tool design determines how well the agent can complete tasks. Poorly designed tools are the most common reason agent systems fail.

Every tool needs three things the model can reason about:

from langchain_core.tools import tool from pydantic import BaseModel, Field from typing import Optional import httpx class WebSearchInput ( BaseModel ): query: str = Field ( description = " The search query. Be specific and use relevant keywords. " ) max_results: int = Field ( default = 5, ge = 1, le = 20, description = " Number of results to return (1-20) " ) @tool ( " search_web ", args_schema = WebSearchInput ) def search_web ( query: str, max_results: int = 5 ) -> str: """ Search the web for current information.

Use this when you need facts, recent news, or data that may not be in your training data. Returns a list of search results with titles, URLs, and snippets. """ # Implementation using a search API (Brave, Serper, Tavily, etc.) response = httpx. get ( " https://api.search-provider.com/search ", params = { " q ": query, " count ": max_results }, headers = { " Authorization ": f " Bearer { SEARCH_API_KEY } " } ) results = response. json ()[ " results " ] formatted = [] for r in results: formatted. append ( f " Title: { r [ ' title ' ] } \n URL: { r [ ' url ' ] } \n Snippet: { r [ ' snippet ' ] } " ) return " \n\n ". join ( formatted ) if formatted else " No results found.

" Enter fullscreen mode Exit fullscreen mode Tool design principles Return human-readable strings. Agents process tool output as text in their context window. JSON blobs work but add tokens and make the model work harder to extract meaning.

Format results as readable summaries.

Return error messages, not exceptions. When a tool fails, catch the exception and return a descriptive string like "Error: Could not retrieve data. The requested resource returned 404." The agent can then decide whether to retry, try a different approach, or ask for help. Unhandled exceptions crash the agent loop entirely.

Make tools idempotent where possible. Agents retry tools when they're unsure the call succeeded. Idempotent tools (where calling twice produces the same result as calling once) are safe to retry. Non-idempotent tools (sending emails, creating database records, charging payments) need idempotency key deduplication logic.

Keep tools atomic. A tool that does one thing is easier to reason about than one that does several things conditionally. If an agent needs to read a file and then process its contents, give it two tools rather than one combined tool with branching logic.

Memory determines what information an agent has access to across steps. Without deliberate memory design, agents either repeat work they've already done, lose track of earlier results, or fill their context window with irrelevant history.

Everything in the active context window is short-term memory. The model can directly reference anything that appears in the conversation history. For simple, focused tasks, this is the only memory you need.

The practical limit is context length. For gpt-4o with a 128K token window, you can fit a substantial amount of agent state in context. For longer-running workflows, you need to manage what stays in context and what gets summarized or stored externally.

I use a rolling window with summarization for long-running agents:

from langchain_openai import ChatOpenAI from langchain.memory import ConversationSummaryBufferMemory memory = ConversationSummaryBufferMemory ( llm = ChatOpenAI ( model = " gpt-4o-mini " ), max_token_limit = 4000, # Keep recent turns verbatim return_messages = True ) Enter fullscreen mode Exit fullscreen mode When the conversation history exceeds max_token_limit, older turns get summarized into a compressed representation that preserves key facts without every individual message.

For agents that need to remember information across separate sessions (user preferences, facts learned in previous runs, knowledge accumulated over time), that information must be persisted to an external store and retrieved selectively.

import json from datetime import datetime class AgentMemoryStore: """ Simple persistent memory using a JSON file or database. """ def __init__ ( self, agent_id: str, storage_backend ): self. agent_id = agent_id self. db = storage_backend def remember ( self, key: str, value: str, category: str = " general " ): """ Store a fact for future retrieval. """ self. db. upsert ({ " agent_id ": self. agent_id, " key ": key, " value ": value, " category ": category, " stored_at ": datetime. utcnow (). isoformat () }) def recall ( self, query: str, category: Optional [ str ] = None, limit: int = 5 ) -> list: """ Retrieve relevant memories using semantic search.

""" filters = { " agent_id ": self. agent_id } if category: filters [ " category " ] = category return self. db. semantic_search ( query, filters = filters, limit = limit ) def forget ( self, key: str ): """ Remove a stored memory. """ self. db. delete ({ " agent_id ": self. agent_id, " key ": key }) Enter fullscreen mode Exit fullscreen mode For production systems, I store agent memory in a vector database (for semantic retrieval) or a key-value store (for exact lookups), depending on whether the agent needs to find "memories related to this topic" or "the specific value stored under this key."

For multi-step tasks, you need to track what has been attempted, what succeeded, and what failed within the current run. I maintain explicit task state rather than relying on the model to reconstruct it from conversation history:

from dataclasses import dataclass, field from enum import Enum from typing import Any class StepStatus ( Enum ): PENDING = " pending " IN_PROGRESS = " in_progress " COMPLETED = " completed " FAILED = " failed " SKIPPED = " skipped " @dataclass class TaskStep: name: str description: str status: StepStatus = StepStatus. PENDING result: Any = None error: Optional [ str ] = None attempts: int = 0 @dataclass class TaskState: goal: str steps: list [ TaskStep ] = field ( default_factory = list ) context: dict = field ( default_factory = dict ) def pending_steps ( self ) -> list [ TaskStep ]: return [ s for s in self. steps if s. status == StepStatus. PENDING ] def completed_steps ( self ) -> list [ TaskStep ]: return [ s for s in self. steps if s. status == StepStatus.

COMPLETED ] def summary ( self ) -> str: lines = [ f " Goal: { self. goal } ", " Steps: " ] for step in self. steps: lines. append ( f " [ { step. status. value } ] { step. name } " ) if step. result: lines. append ( f " Result: { str ( step. result )[: 200 ] } " ) return " \n ". join ( lines ) Enter fullscreen mode Exit fullscreen mode Injecting a concise task state summary into the agent's context at each step prevents the model from losing track of where it is in a complex workflow.

LangGraph is the framework I reach for when building production agent systems. It models agent logic as a directed graph where nodes are processing steps and edges define control flow. This graph structure makes it possible to implement loops, conditional branching, parallel execution, and recovery flows that are difficult to express in linear chain architectures.

The ReAct (Reasoning + Acting) pattern is the foundation of most practical agents: reason about what to do, take an action, observe the result, reason about the next step.

from langgraph.prebuilt import create_react_agent from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage # Define tools tools = [ search_web, read_url, extract_data, write_file ] # Create agent llm = ChatOpenAI ( model = " gpt-4o ", temperature = 0 ) agent = create_react_agent ( llm, tools ) # Run agent result = agent. invoke ({ " messages ": [ HumanMessage ( content = " Research the top 5 cloud storage providers and create a comparison table of their pricing " )] }) print ( result [ " messages " ][ - 1 ]. content ) Enter fullscreen mode Exit fullscreen mode For more control over the execution graph:

from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from typing import TypedDict, Annotated class AgentState ( TypedDict ): messages: Annotated [ list, add_messages ] task_state: TaskState iteration_count: int def should_continue ( state: AgentState ) -> str: """ Determine whether to continue the agent loop or finish. """ messages = state [ " messages " ] last_message = messages [ - 1 ] # If the last AI message has no tool calls, we're done if not last_message. tool_calls: return " end " # Prevent infinite loops if state [ " iteration_count " ] >= 15: return " end " return " continue " def call_model ( state: AgentState ) -> AgentState: """ Call the LLM with current state. """ response = llm_with_tools. invoke ( state [ " messages " ]) return { " messages ": [ response ], " iteration_count ": state [ " iteration_count " ] + 1 } def call_tools ( state: AgentState ) -> AgentState: """ Execute tool calls from the last model response.

""" tool_executor = ToolExecutor ( tools ) last_message = state [ " messages " ][ - 1 ] tool_results = tool_executor. batch ( last_message. tool_calls ) return { " messages ": tool_results } # Build graph graph = StateGraph ( AgentState ) graph. add_node ( " agent ", call_model ) graph. add_node ( " tools ", call_tools ) graph. set_entry_point ( " agent " ) graph. add_conditional_edges ( " agent ", should_continue, { " continue ": " tools ", " end ": END }) graph. add_edge ( " tools ", " agent " ) runnable = graph. compile () Enter fullscreen mode Exit fullscreen mode The should_continue function is where you implement loop termination logic. Hard iteration limits are key; without them, a confused agent can loop indefinitely on a failing tool call.

When an agent needs to call multiple independent tools, executing them in parallel rather than sequentially reduces latency significantly:

import asyncio from langchain_core.tools import BaseTool async def execute_tools_parallel ( tool_calls: list [ dict ], tools: dict [ str, BaseTool ]) -> list: """ Execute multiple tool calls concurrently. """ tasks = [] for call in tool_calls: tool = tools [ call [ " name " ]] tasks. append ( tool. ainvoke ( call [ " args " ])) results = await asyncio. gather ( * tasks, return_exceptions = True ) # Convert exceptions to error strings so the agent can recover processed = [] for i, result in enumerate ( results ): if isinstance ( result, Exception ): processed. append ( f " Error in { tool_calls [ i ][ ' name ' ] }: { str ( result ) } " ) else: processed. append ( result ) return processed Enter fullscreen mode Exit fullscreen mode A research agent that needs to fetch five URLs runs them all concurrently rather than one after another, reducing a 15-second sequential operation to 3-4 seconds.

CrewAI provides a higher-level abstraction for multi-agent systems organized around roles and crews. Each agent has a defined role, goal, and backstory that shapes its behavior, and a crew coordinates their collaborative work.

from crewai import Agent, Task, Crew, Process from crewai_tools import SerperDevTool, WebsiteSearchTool search_tool = SerperDevTool () web_tool = WebsiteSearchTool () # Define specialized agents researcher = Agent ( role = " Market Research Specialist ", goal = " Find comprehensive and accurate information about competitors ", backstory = """ You are an expert market researcher with years of experience analyzing SaaS companies. You know how to find pricing pages, feature lists, customer reviews, and recent news efficiently. """, tools = [ search_tool, web_tool ], verbose = True, max_iter = 5 ) analyst = Agent ( role = " Competitive Intelligence Analyst ", goal = " Extract structured data and identify strategic insights from research ", backstory = """ You specialize in distilling raw research into actionable competitive intelligence.

You identify differentiation opportunities and market gaps. """, tools = [], # Pure reasoning, no external tools needed verbose = True ) writer = Agent ( role = " Technical Writer ", goal = " Produce clear, well-structured reports from analysis ", backstory = """ You write executive-level reports that communicate complex competitive landscapes clearly and concisely. """, tools = [], verbose = True ) # Define tasks with explicit dependencies research_task = Task ( description = """ Research {competitor_name}.

Find: 1. Current pricing (all tiers) 2. Key features and capabilities 3.

Target customer segments 4. Recent product announcements (last 6 months) 5. Customer reviews highlighting strengths and complaints """, expected_output = " Comprehensive research notes with sources cited ", agent = researcher ) analysis_task = Task ( description = """ Using the research on {competitor_name}, produce: 1.

Feature comparison table vs our product 2. Pricing position analysis 3. Three key differentiation opportunities for us 4.

Three areas where they outperform us """, expected_output = " Structured competitive analysis with actionable insights ", agent = analyst, context = [ research_task ] # Depends on research completion ) report_task = Task ( description = " Write a one-page executive summary of the competitive analysis ", expected_output = " Professional report in markdown format, under 600 words ", agent = writer, context = [ research_task, analysis_task ] ) # Assemble crew crew = Crew ( agents = [ researcher, analyst, writer ], tasks = [ research_task, analysis_task, report_task ], process = Process. sequential, verbose = True ) result = crew. kickoff ( inputs = { " competitor_name ": " Competitor Corp " }) Enter fullscreen mode Exit fullscreen mode The context parameter on each task passes completed task output to subsequent tasks, creating explicit data flow dependencies.

When work moves between agents, the quality of the handoff determines whether context is preserved or lost.

Never pass raw LLM text between agents. Structure handoff data:

from pydantic import BaseModel class ResearchHandoff ( BaseModel ): """ Structured output from research agent to analysis agent. """ subject: str sources_consulted: list [ str ] key_findings: list [ str ] raw_data: dict confidence_level: str # "high", "medium", "low" gaps_identified: list [ str ] # What could not be found class AnalysisHandoff ( BaseModel ): """ Structured output from analysis agent to writing agent. """ subject: str key_insights: list [ str ] data_tables: list [ dict ] recommended_actions: list [ str ] supporting_evidence: dict [ str, str ] Enter fullscreen mode Exit fullscreen mode Structured handoffs prevent the receiving agent from misinterpreting unstructured text, allow the orchestrator to validate completeness before proceeding, and create a clear audit trail.

def orchestrate_research_pipeline ( subject: str ) -> str: """ Orchestrate a multi-agent research and analysis pipeline. """ # Step 1: Research research_result = researcher_agent. invoke ({ " task ": f " Research { subject } ", " output_format ": ResearchHandoff. schema () }) # Validate handoff data try: research_data = ResearchHandoff. model_validate_json ( research_result ) except Exception as e: return f " Research agent produced invalid output: { e } " if research_data. confidence_level == " low ": # Request human review before proceeding return request_human_review ( research_data, reason = " Low confidence research " ) # Step 2: Analysis analysis_result = analyst_agent. invoke ({ " research ": research_data. model_dump (), " task ": " Analyze the research and identify key insights " }) analysis_data = AnalysisHandoff. model_validate_json ( analysis_result ) # Step 3: Write report report = writer_agent. invoke ({ " analysis ": analysis_data. model_dump (), " task ": " Write an executive summary report " }) return report Enter fullscreen mode Exit fullscreen mode Failure Recovery Production agent systems fail. Tools time out, APIs return unexpected responses, models produce malformed output.

Failure recovery design determines whether those failures cascade into complete system failure or get handled gracefully.

import time from functools import wraps def retry_with_backoff ( max_retries: int = 3, base_delay: float = 1.0 ): """ Decorator for automatic retry with exponential backoff. """ def decorator ( func ): @wraps ( func ) def wrapper ( * args, ** kwargs ): last_exception = None for attempt in range ( max_retries ): try: return func ( * args, ** kwargs ) except Exception as e: last_exception = e if attempt < max_retries - 1: delay = base_delay * ( 2 ** attempt ) print ( f " Attempt { attempt + 1 } failed: { e }. Retrying in { delay } s...

" ) time. sleep ( delay ) raise last_exception return wrapper return decorator @retry_with_backoff ( max_retries = 3 ) def call_external_api ( url: str ) -> dict: response = httpx. get ( url, timeout = 10.0 ) response. raise_for_status () return response. json () Enter fullscreen mode Exit fullscreen mode Fallback tools For critical capabilities, define fallback tools the agent can use when the primary tool fails:

TOOL_FALLBACKS = { " search_web_primary ": " search_web_fallback ", " read_url_primary ": " extract_url_text_cached ", } def get_tool_with_fallback ( tool_name: str, tools: dict ) -> BaseTool: if tool_name in tools: return tools [ tool_name ] fallback_name = TOOL_FALLBACKS. get ( tool_name ) if fallback_name and fallback_name in tools: print ( f " Primary tool ' { tool_name } ' unavailable, using fallback ' { fallback_name } '" ) return tools [ fallback_name ] raise ValueError ( f " No tool available for: { tool_name } " ) Enter fullscreen mode Exit fullscreen mode Human-in-the-loop checkpoints Not every decision should be automated. High-stakes actions (sending emails, making purchases, deleting data) require explicit human approval before execution:

from langgraph.checkpoint.memory import MemorySaver from langgraph.types import interrupt def execute_high_stakes_action ( state: AgentState ) -> AgentState: """ Node that requires human approval before proceeding. """ proposed_action = state [ " proposed_action " ] # Interrupt the graph and wait for human input human_response = interrupt ({ " question ": f " Agent wants to: { proposed_action [ ' description ' ] } \n\n Approve? (yes/no) ", " proposed_action ": proposed_action }) if human_response. lower ()!= " yes ": return { " messages ": [ ToolMessage ( content = " Action cancelled by human reviewer. ", tool_call_id = " human_review " )], " action_approved ": False } # Proceed with action result = execute_action ( proposed_action ) return { " action_approved ": True, " action_result ": result } # Enable checkpointing for interrupt support memory = MemorySaver () graph = graph. compile ( checkpointer = memory, interrupt_before = [ " execute_high_stakes_action " ]) Enter fullscreen mode Exit fullscreen mode LangGraph's interrupt mechanism pauses graph execution, persists state, and resumes from the same point after human input, no lost context.

Agent systems are notoriously difficult to debug without thorough tracing. Every tool call, model invocation, and decision point should be logged with enough context to reconstruct what happened:

from langsmith import Client # LangSmith traces all LangChain/LangGraph calls automatically when configured os. environ [ " LANGCHAIN_TRACING_V2 " ] = " true " os. environ [ " LANGCHAIN_PROJECT " ] = " production-research-agent " Enter fullscreen mode Exit fullscreen mode LangSmith is the standard tracing tool for LangChain/LangGraph systems. It captures full execution traces including model inputs/outputs, tool calls, latency per step, and token usage.

class AgentBudget: def __init__ ( self, max_tokens: int = 100_000, max_time_seconds: int = 120 ): self. max_tokens = max_tokens self. max_time_seconds = max_time_seconds self. tokens_used = 0 self. start_time = time. time () def check ( self ) -> bool: """ Returns True if budget allows continued execution. """ elapsed = time. time () - self. start_time if elapsed > self. max_time_seconds: raise TimeoutError ( f " Agent exceeded { self. max_time_seconds } s time budget " ) if self. tokens_used > self. max_tokens: raise BudgetExceededError ( f " Agent exceeded { self. max_tokens } token budget " ) return True Enter fullscreen mode Exit fullscreen mode An agent that loops on a failing tool will burn tokens indefinitely without hard limits.

Design agent pipelines so that re-running a partially completed job produces the same final result as a successful first run. Cache tool results keyed by input, skip completed steps, and use transaction IDs to prevent duplicate side effects.

Agent orchestration is the infrastructure layer that makes AI capabilities compound. A well-designed orchestration layer turns individual model calls into reliable business processes. The key is treating agent systems with the same engineering discipline you'd apply to any distributed system: explicit state management, graceful failure handling, thorough observability, and hard limits on autonomous execution.