Supervisor Pattern

Supervisor (receives goal)
  ├─ Tool call: Research Agent → result 1
  ├─ Tool call: Analyst Agent → result 2
  └─ Tool call: Writer Agent → final output

import { Agent } from "@mastra/core/agent"
import { openai } from "@ai-sdk/openai"

// --- Sub-agents ---

const researchAgent = new Agent({
  name: "Researcher",
  instructions: `You are a research specialist. Given a topic or question,
    search for relevant information and return a structured summary with:
    - key facts
    - relevant sources
    - confidence level (high/medium/low)
    Return only factual content; do not interpret or recommend.`,
  model: openai("gpt-4o-mini"),
  tools: { webSearch, fetchUrl, extractText }
})

const analystAgent = new Agent({
  name: "Analyst",
  instructions: `You are a data analysis specialist. Given research summaries
    and data, identify patterns, draw conclusions, and flag uncertainties.
    Return structured analysis with explicit confidence bounds.`,
  model: openai("gpt-4o"),
  tools: { runQuery, generateChart, calculateStats }
})

const writerAgent = new Agent({
  name: "Writer",
  instructions: `You are a technical writer. Given research and analysis,
    produce clear, structured reports for a technical audience.
    Follow house style: active voice, concrete examples, no jargon.`,
  model: openai("gpt-4o"),
  tools: { formatMarkdown, checkGrammar }
})

// --- Supervisor ---

const supervisorAgent = new Agent({
  name: "Supervisor",
  instructions: `You coordinate research, analysis, and writing tasks.
    
    For any user request:
    1. Determine which specialists are needed
    2. Delegate to Researcher first to gather facts
    3. Delegate to Analyst with the research results
    4. Delegate to Writer with the analysis to produce the final output
    5. Review the output and return it to the user
    
    Always delegate — do not attempt research, analysis, or writing yourself.
    If a specialist returns an error, report it clearly and suggest a retry or alternative.`,
  model: openai("gpt-4o"),
  agents: [researchAgent, analystAgent, writerAgent]
})

// Usage
const response = await supervisorAgent.generate("Analyze the competitive landscape for MCP servers")

const supervisorAgent = new Agent({
  name: "Supervisor",
  instructions: "...",
  model: openai("gpt-4o"),
  agents: [researchAgent, analystAgent, writerAgent],
  onGenerateText: async ({ messages, tools }) => {
    // Log which agent is being invoked for observability
    const delegationCalls = tools?.filter(t => t.type === "agent")
    if (delegationCalls?.length) {
      console.log(`Supervisor delegating to: ${delegationCalls.map(t => t.name).join(", ")}`)
    }
  }
})

import { generateText, tool } from "ai"
import { openai } from "@ai-sdk/openai"
import { z } from "zod"

// --- Sub-agent tools ---

const researchTool = tool({
  description: `Research a topic and return structured findings.
    Use when you need factual information about a specific subject.
    Returns sources, key facts, and confidence level.`,
  parameters: z.object({
    topic: z.string().describe("The topic or question to research"),
    depth: z.enum(["quick", "thorough"]).describe("How deep to search")
  }),
  execute: async ({ topic, depth }) => {
    // The sub-agent is invoked here as a regular function call
    const { text } = await generateText({
      model: openai("gpt-4o-mini"),
      system: `You are a research specialist. Search thoroughly and return structured JSON:
        { "facts": [], "sources": [], "confidence": "high|medium|low" }`,
      prompt: `Research: ${topic} (depth: ${depth})`
    })
    return text
  }
})

const analysisTool = tool({
  description: `Analyze data or research findings and return structured conclusions.
    Use after research is complete to interpret findings and identify patterns.
    Returns conclusions with explicit confidence bounds.`,
  parameters: z.object({
    data: z.string().describe("The research findings or data to analyze"),
    question: z.string().describe("The specific analytical question to answer")
  }),
  execute: async ({ data, question }) => {
    const { text } = await generateText({
      model: openai("gpt-4o"),
      system: "You are a data analysis specialist. Return structured JSON analysis.",
      prompt: `Analyze this data:\n${data}\n\nQuestion: ${question}`
    })
    return text
  }
})

// --- Supervisor ---

const { text } = await generateText({
  model: openai("gpt-4o"),
  system: `You coordinate research and analysis tasks.
    Always use the research tool before the analysis tool.
    Synthesize the final result into a clear, actionable summary.`,
  prompt: userRequest,
  tools: { research: researchTool, analyze: analysisTool },
  maxSteps: 10 // prevent runaway loops
})

const researchTool = tool({
  description: "Research a topic...",
  parameters: z.object({ topic: z.string() }),
  execute: async ({ topic }) => {
    // Returns a large object with full research data
    return await runResearchAgent(topic)
  },
  // Only pass a compact summary to the supervisor's context
  toModelOutput: (result) => ({
    type: "text",
    text: `Research complete. Key facts: ${result.facts.slice(0, 3).join("; ")}. Confidence: ${result.confidence}.`
  })
})

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import create_supervisor
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# --- Sub-agents as LangGraph nodes ---

def research_node(state: dict) -> dict:
    """Research sub-agent node."""
    model = ChatOpenAI(model="gpt-4o-mini")
    result = model.invoke([
        HumanMessage(content=f"Research: {state['task']}")
    ])
    return {"research": result.content, "messages": [result]}

def analysis_node(state: dict) -> dict:
    """Analysis sub-agent node."""
    model = ChatOpenAI(model="gpt-4o")
    result = model.invoke([
        HumanMessage(content=f"Analyze this research:\n{state['research']}")
    ])
    return {"analysis": result.content, "messages": [result]}

# --- Using create_supervisor prebuilt ---

from langgraph.prebuilt import create_supervisor

supervisor_graph = create_supervisor(
    agents=[research_agent, analysis_agent, writer_agent],
    model=ChatOpenAI(model="gpt-4o"),
    prompt="""You coordinate research, analysis, and writing.
    
    Route to 'research' for fact-gathering tasks.
    Route to 'analysis' after research is complete and needs interpretation.
    Route to 'writer' to produce the final document.
    Route to FINISH when the output is complete and reviewed."""
)

# --- Manual supervisor implementation for full control ---

def supervisor_node(state: dict) -> dict:
    """Supervisor determines the next step."""
    next_step = supervisor_model.invoke(state["messages"])
    return {"next": next_step.content}

def route(state: dict) -> str:
    if state["next"] == "FINISH":
        return END
    return state["next"]

builder = StateGraph(dict)
builder.add_node("supervisor", supervisor_node)
builder.add_node("research", research_node)
builder.add_node("analysis", analysis_node)
builder.add_edge("__start__", "supervisor")
builder.add_conditional_edges("supervisor", route)
builder.add_edge("research", "supervisor")
builder.add_edge("analysis", "supervisor")

graph = builder.compile()

from langgraph.prebuilt import create_react_agent

research_agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4o-mini"),
    tools=[web_search, fetch_url],
    name="researcher",
    prompt="You are a research specialist..."
)

# Expose the agent as a tool for the supervisor
research_tool = research_agent.as_tool(
    name="research",
    description="Research a topic and return structured findings. "
                "Use before analysis tasks."
)

supervisor = create_react_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=[research_tool, analysis_tool, write_tool],
    prompt="You coordinate research, analysis, and writing..."
)

from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

# --- Workers ---

researcher = Agent(
    role="Research Specialist",
    goal="Find accurate, relevant information for any given topic",
    backstory="""You have deep expertise in research methodology and source evaluation.
    You are rigorous, skeptical, and precise in your findings.""",
    tools=[web_search, document_reader],
    llm=ChatOpenAI(model="gpt-4o-mini"),
    verbose=True
)

analyst = Agent(
    role="Data Analyst",
    goal="Derive meaningful insights from research and data",
    backstory="""You specialize in pattern recognition, statistical reasoning,
    and translating complex data into clear conclusions.""",
    tools=[run_calculation, generate_chart],
    llm=ChatOpenAI(model="gpt-4o"),
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Produce clear, accurate reports from research and analysis",
    backstory="""You write for technical audiences. Your reports are precise,
    structured, and free of jargon.""",
    tools=[format_document],
    llm=ChatOpenAI(model="gpt-4o"),
    verbose=True
)

# --- Tasks ---

research_task = Task(
    description="Research the following topic thoroughly: {topic}",
    expected_output="Structured research summary with sources and confidence levels",
    agent=researcher
)

analysis_task = Task(
    description="Analyze the research findings and identify key insights",
    expected_output="Analysis with conclusions and evidence",
    agent=analyst,
    context=[research_task]  # This task depends on research_task
)

report_task = Task(
    description="Write a comprehensive report based on the research and analysis",
    expected_output="Complete technical report in Markdown",
    agent=writer,
    context=[research_task, analysis_task]
)

# --- Crew with hierarchical process ---

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4o"),  # auto-created manager
    verbose=True
)

result = crew.kickoff(inputs={"topic": "MCP server adoption patterns"})

from openai.agents import Agent, Runner

research_agent = Agent(
    name="Researcher",
    instructions="""Research topics and return structured findings.
    When done, hand off back to the supervisor.""",
    tools=[web_search, fetch_url]
)

analysis_agent = Agent(
    name="Analyst",
    instructions="""Analyze research findings and return structured analysis.
    When done, hand off back to the supervisor.""",
    tools=[run_calculation]
)

supervisor = Agent(
    name="Supervisor",
    instructions="""Coordinate research and analysis to fulfill user requests.
    
    1. Hand off to Researcher to gather facts
    2. Hand off to Analyst to interpret findings
    3. Synthesize the final response
    
    Note: When you hand off, execution transfers to that agent.
    They will hand back to you when done.""",
    handoffs=[research_agent, analysis_agent]
)

result = await Runner.run(supervisor, "Analyze the MCP ecosystem")
print(result.final_output)

from openai.agents import Agent, Runner

research_agent = Agent(
    name="Researcher",
    instructions="Research topics thoroughly and return structured findings.",
    tools=[web_search, fetch_url]
)

# Expose as a tool — supervisor receives the output as a tool result
research_tool = research_agent.as_tool(
    tool_name="research",
    tool_description="Research a topic and return structured findings. "
                     "Use before analysis tasks. Returns facts, sources, confidence."
)

supervisor = Agent(
    name="Supervisor",
    instructions="""Use your tools to complete research and analysis tasks.
    Always research before analyzing. Synthesize the final response yourself.""",
    tools=[research_tool, analysis_tool]  # sub-agents appear as regular tools
)

result = await Runner.run(supervisor, "Analyze the MCP ecosystem")