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

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']}\nURL: {r['url']}\nSnippet: {r['snippet']}")

    return "\n\n".join(formatted) if formatted else "No results found."

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
)

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})

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)

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)

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()

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

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"})

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]

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

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()

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}")

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\nApprove? (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"])

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"

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