How OpenMAIC Implements Multi-Agent Orchestration Using LangGraph

OpenMAIC implements multi-agent orchestration using LangGraph through a lightweight StateGraph architecture where a "director" node decides which agent speaks next and an "agent_generate" node streams that agent's response.

The THU-MAIC/OpenMAIC repository provides a production-ready pattern for coordinating multiple AI agents in conversational workflows. Unlike complex hierarchical systems, OpenMAIC uses a deliberately simple topology—START → director → agent_generate → END—that keeps the server stateless while enabling dynamic speaker selection via LLM-driven decisions.

Core Architecture Components

OrchestratorState: The Shared Memory

All mutable state flows through OrchestratorState, defined in [lib/orchestration/types.ts](https://github.com/THU-MAIC/OpenMAIC/blob/main/lib/orchestration/types.ts#L9-L27) using LangChain's Annotation.Root:

// lib/orchestration/types.ts#L9-L27
const OrchestratorStateAnnotation = Annotation.Root({
  messages: Annotation<BaseMessage[]>({
    reducer: messagesStateReducer,
    default: () => [],
  }),
  availableAgentIds: Annotation<string[]>({
    default: () => [],
  }),
  currentAgentId: Annotation<string>({
    default: () => undefined,
  }),
  turnCount: Annotation<number>({
    default: () => 0,
  }),
  // ... additional fields for whiteboard state, summaries, etc.
});

This centralized state enables the director and agent nodes to communicate without tight coupling.

Director Node: The Decision Engine

The director node in [lib/orchestration/director-graph.ts](https://github.com/THU-MAIC/OpenMAIC/blob/main/lib/orchestration/director-graph.ts#L90-L123) implements three decision paths:

• Single-agent fast path — When availableAgentIds.length <= 1, the director uses pure code logic with no LLM overhead.
• Trigger agent shortcut — On turn 0, if a triggerAgentId is specified, it bypasses the LLM.
• LLM-driven selection — For multi-agent scenarios, it calls the adapter and parses structured output to determine the next speaker, user cue, or termination.

The LLM integration appears in lines 150-190:

// lib/orchestration/director-graph.ts#L150-190 (simplified structure)
async function directorNode(state: typeof OrchestratorStateAnnotation.State, config: RunnableConfig) {
  if (state.availableAgentIds.length <= 1) {
    // Fast path: no LLM call needed
    return { currentAgentId: state.availableAgentIds[0] };
  }

  if (state.turnCount === 0 && state.triggerAgentId) {
    return { currentAgentId: state.triggerAgentId };
  }

  // Multi-agent: query LLM for next speaker
  const adapter = config.configurable?.adapter as AISdkLangGraphAdapter;
  const systemPrompt = buildDirectorPrompt(state);
  
  const result = await adapter._generate(
    [new SystemMessage(systemPrompt), ...state.messages],
    { signal: config.signal }
  );
  
  const decision = parseDirectorDecision(result.generations[0]?.text);
  
  if (decision.shouldEnd || decision.nextAgentId === 'USER') {
    return { shouldEnd: true };
  }
  
  return { currentAgentId: decision.nextAgentId };
}

Agent Generate Node: Streaming Responses

The agent_generate node (lines 31-68) handles the actual agent execution:

• Constructs system prompts with agent-specific configuration
• Converts internal messages to LLM-compatible format
• Streams structured chunks (text deltas, whiteboard actions, errors)
• Emits events via config.writer for Server-Sent Events (SSE)

// lib/orchestration/director-graph.ts#L31-68 (streaming implementation pattern)
async function agentGenerateNode(state: typeof OrchestratorStateAnnotation.State, config: RunnableConfig) {
  const adapter = config.configurable?.adapter as AISdkLangGraphAdapter;
  const agentId = state.currentAgentId;
  const agent = state.agents[agentId];
  
  const systemPrompt = buildAgentSystemPrompt(agent, state.whiteboardState);
  const messages = convertMessagesToOpenAI(state.messages, agentId);
  
  const stream = adapter.streamGenerate(
    [new SystemMessage(systemPrompt), ...messages],
    { signal: config.signal }
  );
  
  const summary: AgentTurnSummary = {
    agentId,
    text: '',
    actions: [],
  };
  
  for await (const chunk of stream) {
    if (chunk.type === 'delta') {
      const parsed = parseStructuredChunk(chunk.content, parserState);
      
      if (parsed.text) {
        config.writer({ type: 'text_delta', content: parsed.text });
        summary.text += parsed.text;
      }
      if (parsed.action) {
        config.writer({ type: 'action', content: parsed.action });
        summary.actions.push(parsed.action);
      }
    }
  }
  
  return { messages: [new AIMessage(summary.text)], turnSummary: summary };
}

Graph Construction and Execution Flow

Building the StateGraph

The orchestration graph is wired together in [lib/orchestration/director-graph.ts](https://github.com/THU-MAIC/OpenMAIC/blob/main/lib/orchestration/director-graph.ts#L84-L95):

// lib/orchestration/director-graph.ts#L84-95
export function createOrchestrationGraph(): CompiledStateGraph<typeof OrchestratorStateAnnotation.State> {
  const workflow = new StateGraph(OrchestratorStateAnnotation)
    .addNode('director', directorNode)
    .addNode('agent_generate', agentGenerateNode)
    .addEdge(START, 'director')
    .addConditionalEdges('director', directorRouter, {
      agent_generate: 'agent_generate',
      __end__: END,
    })
    .addEdge('agent_generate', END);

  return workflow.compile();
}

The directorRouter function examines the director's output to determine the next edge—either to agent_generate for the selected agent or to END if the turn concludes.

Per-Request State Initialization

Each incoming chat request builds fresh state via buildInitialState():

// lib/orchestration/director-graph.ts#L102-118
export function buildInitialState(
  request: StatelessChatRequest,
  languageModel: LanguageModel,
  thinkingConfig?: ThinkingConfig
): typeof OrchestratorStateAnnotation.State {
  return {
    messages: request.messages.map(convertOpenAIMessageToBaseMessage),
    availableAgentIds: request.agents.map(a => a.id),
    triggerAgentId: request.triggerAgentId,
    agents: Object.fromEntries(request.agents.map(a => [a.id, a])),
    languageModel,
    thinkingConfig,
    turnCount: 0,
    whiteboardState: request.initialWhiteboardState ?? {},
  };
}

Complete Usage Example

Here's how to integrate the full orchestration pipeline in a Next.js API route:

// app/api/chat/route.ts
import { createOrchestrationGraph, buildInitialState } from '@/lib/orchestration/director-graph';
import { AISdkLangGraphAdapter } from '@/lib/orchestration/ai-sdk-adapter';
import { myLanguageModel } from '@/lib/models';

// Compile once at module initialization
const graph = createOrchestrationGraph();

export async function POST(req: Request) {
  const request = await req.json() as StatelessChatRequest;
  
  const initialState = buildInitialState(
    request,
    myLanguageModel,
    request.thinkingConfig
  );
  
  const adapter = new AISdkLangGraphAdapter(myLanguageModel);
  
  const stream = new ReadableStream({
    async start(controller) {
      const result = await graph.invoke(initialState, {
        configurable: { adapter },
        writer: (chunk) => {
          controller.enqueue(`data: ${JSON.stringify(chunk)}\n\n`);
        },
      });
      
      controller.enqueue(`data: ${JSON.stringify({ type: 'done', result })}\n\n`);
      controller.close();
    },
  });
  
  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
    },
  });
}

Key Design Decisions in OpenMAIC's LangGraph Implementation

Stateless server architecture — The graph executes a single director-agent cycle per request. The client controls conversation length by sending successive requests, eliminating server-side session management.

No explicit turn limits — Unlike systems with maxTurns configuration, OpenMAIC delegates loop control to the client. This simplifies reasoning about state and enables flexible UI patterns.

Adapter pattern for model abstraction — The AISdkLangGraphAdapter ([lib/orchestration/ai-sdk-adapter.ts](https://github.com/THU-MAIC/OpenMAIC/blob/main/lib/orchestration/ai-sdk-adapter.ts)) decouples LangGraph from specific LLM providers, allowing any AI SDK-compatible model.

Structured streaming output — Both director decisions and agent responses use structured parsing ([lib/orchestration/director-prompt.ts](https://github.com/THU-MAIC/OpenMAIC/blob/main/lib/orchestration/director-prompt.ts)), enabling type-safe handling of whiteboard actions alongside text generation.

Summary

• OpenMAIC uses LangGraph's StateGraph with a minimal two-node topology: director selects speakers, agent_generate streams responses.
• OrchestratorState centralizes all mutable state through LangChain's annotation system in types.ts.
• Three director decision paths handle single-agent, trigger-agent, and multi-agent scenarios with appropriate LLM usage.
• Streaming architecture emits real-time events via config.writer for text deltas and whiteboard actions.
• Stateless design puts conversation loop control in the client, with each request executing one graph traversal.

Frequently Asked Questions

How does OpenMAIC decide which agent speaks next?

The director node implements a cascading decision strategy. For single-agent configurations, it uses pure code. For multi-agent scenarios, it may fast-path a trigger agent on the first turn. Otherwise, it sends a structured prompt to the LLM via AISdkLangGraphAdapter._generate() and parses the response with parseDirectorDecision() to extract the next agent ID, user cue, or termination signal.

Why doesn't OpenMAIC use a looping graph topology?

The repository deliberately avoids server-side loops. Each request executes START → director → agent_generate → END exactly once. This keeps the server stateless and simplifies deployment—conversation history management happens client-side, and the client initiates subsequent turns as new requests. The trade-off is additional latency between turns compared to persistent connections.

What is AISdkLangGraphAdapter and why is it needed?

AISdkLangGraphAdapter bridges LangGraph's expected interfaces with the AI SDK's LanguageModel abstraction. It implements _generate() for director decisions and streamGenerate() for agent responses, normalizing different provider streaming formats into consistent chunks. This adapter lives in [lib/orchestration/ai-sdk-adapter.ts](https://github.com/THU-MAIC/OpenMAIC/blob/main/lib/orchestration/ai-sdk-adapter.ts) and enables swapping underlying models without changing orchestration logic.

How are whiteboard actions handled during agent streaming?

The agent generate node parses structured chunks from the LLM stream using parseStructuredChunk(). When it detects action payloads, it emits action events through config.writer alongside text_delta events. These actions update the whiteboardState field in OrchestratorState, making them available to subsequent turns and for final summary generation in AgentTurnSummary.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}