Where to Find the LangGraph StateGraph Implementation in OpenMAIC

The LangGraph StateGraph implementation in OpenMAIC is located in lib/orchestration/director-graph.ts, which defines the graph structure and node logic, and lib/orchestration/stateless-generate.ts, which handles instantiation and streaming execution.

OpenMAIC builds its multi-agent orchestration system on top of LangChain's LangGraph framework. The LangGraph StateGraph powers the director-agent workflow that coordinates multiple AI agents in a single conversation turn. This article maps the exact files, functions, and patterns that implement this orchestration layer in the THU-MAIC/OpenMAIC repository.

Core StateGraph Implementation: director-graph.ts

The primary LangGraph StateGraph definition lives in lib/orchestration/director-graph.ts. This file contains three essential components: state annotation, node definitions, and graph construction.

State Definition with Annotation.Root

Lines 48-77 declare the OrchestratorState using LangGraph's Annotation.Root:

// lib/orchestration/director-graph.ts
import { Annotation } from '@langchain/langgraph';

const OrchestratorState = Annotation.Root({
  // Immutable inputs
  messages: Annotation<Message[]>,
  availableAgentIds: Annotation<string[]>,
  
  // Mutable fields updated by nodes
  currentAgentId: Annotation<string | undefined>,
  turnCount: Annotation<number>,
  agentResponses: Annotation<AgentResponse[]>,
  shouldEnd: Annotation<boolean>,
});

The state separates read-only inputs (messages, available agents) from mutable fields that nodes modify during execution (current agent selection, turn counting, response accumulation).

The Director Node

The director node (lines 89-123 and 126-196) implements the orchestration logic:

async function directorNode(state: typeof OrchestratorState.State, config: RunnableConfig) {
  const { messages, availableAgentIds } = state;
  
  // Single-agent shortcut: skip LLM call when only one agent exists
  if (availableAgentIds.length === 1) {
    return { currentAgentId: availableAgentIds[0] };
  }
  
  // LLM-driven agent selection with streaming status events
  const selection = await selectAgentWithLLM(messages, availableAgentIds, {
    writer: config.writer, // Streams "director_thinking" events
  });
  
  return {
    currentAgentId: selection.agentId,
    shouldEnd: selection.shouldEnd,
  };
}

The director writes streaming events via config.writer to provide real-time feedback during agent selection.

The Agent Generation Node

The agent_generate node (lines 334-468) executes individual agent responses:

async function agentGenerateNode(state: typeof OrchestratorState.State, config: RunnableConfig) {
  const { currentAgentId, messages } = state;
  
  // Stream lifecycle events: agent_start → text_delta → action → agent_end
  config.writer?.({ type: 'agent_start', agentId: currentAgentId });
  
  const response = await generateAgentResponse(currentAgentId, messages, {
    onToken: (delta) => config.writer?.({ type: 'text_delta', delta }),
    onAction: (action) => config.writer?.({ type: 'action', action }),
  });
  
  config.writer?.({ type: 'agent_end', agentId: currentAgentId });
  
  return {
    agentResponses: [...state.agentResponses, response],
    turnCount: state.turnCount + 1,
  };
}

Graph Construction and Compilation

Lines 84-95 assemble the StateGraph topology:

export function createOrchestrationGraph() {
  const graph = new StateGraph(OrchestratorState)
    .addNode('director', directorNode)
    .addNode('agent_generate', agentGenerateNode)
    .addEdge(START, 'director')
    .addConditionalEdges('director', directorCondition, {
      agent_generate: 'agent_generate',
      [END]: END,
    })
    .addEdge('agent_generate', END);
  
  return graph.compile();
}

The conditional edge directorCondition (lines 24-27) routes execution based on state.shouldEnd:

function directorCondition(state: typeof OrchestratorState.State): 'agent_generate' | typeof END {
  return state.shouldEnd ? END : 'agent_generate';
}

Graph Execution: stateless-generate.ts

The LangGraph StateGraph runtime entry point is lib/orchestration/stateless-generate.ts. This file demonstrates practical graph instantiation and streaming (lines 5-14):

// lib/orchestration/stateless-generate.ts
import { createOrchestrationGraph, buildInitialState } from './director-graph';

export async function* statelessGenerate(
  request: StatelessChatRequest,
  abortSignal: AbortSignal,
  languageModel: LanguageModel,
  thinkingConfig?: ThinkingConfig,
) {
  // 1. Create compiled graph
  const graph = createOrchestrationGraph();
  
  // 2. Build initial state from request
  const initialState = buildInitialState(request, languageModel, thinkingConfig);
  
  // 3. Stream with custom mode for real-time events
  const stream = await graph.stream(initialState, {
    streamMode: 'custom' as any,
    signal: abortSignal,
  });
  
  // 4. Yield events to client
  for await (const chunk of stream) {
    yield chunk as StatelessEvent;
  }
}

The streamMode: 'custom' option enables LangGraph's event streaming for director thinking updates, token deltas, and agent lifecycle events.

Architectural Flow Diagram

┌─────────────┐     ┌─────────────┐     ┌─────────────────┐     ┌─────┐
│    START    │────▶│   director  │────▶│ agent_generate  │────▶│ END │
└─────────────┘     └─────────────┘     └─────────────────┘     └─────┘
                          │                                             
                          │ (shouldEnd=true)                            
                          └────────────────────────────────────────────▶

• Single-round execution: Each request runs one director decision followed by zero or one agent generation
• Streaming throughout: Both nodes emit events via config.writer
• Stateless design: No conversation persistence between requests

LangGraph StateGraph Configuration Details

Aspect	Implementation
State container	Annotation.Root<OrchestratorState> with typed fields
Node functions	Async functions receiving (state, config)
Conditional routing	directorCondition returns edge keys
Compilation	graph.compile() produces executable CompiledStateGraph
Streaming	streamMode: 'custom' yields real-time events
Cancellation	AbortSignal passed through to graph.stream()

Summary

• Primary location: lib/orchestration/director-graph.ts contains the complete LangGraph StateGraph definition including state schema, nodes, edges, and the createOrchestrationGraph() factory
• Execution entry: lib/orchestration/stateless-generate.ts instantiates and streams the compiled graph for single-turn requests
• State management: Uses LangGraph's Annotation.Root pattern with explicit mutable/immutable field separation
• Event streaming: Both nodes leverage config.writer to emit director_thinking, agent_start, text_delta, and agent_end events
• Conditional logic: directorCondition enables early termination when shouldEnd is true

Frequently Asked Questions

What LangGraph version does OpenMAIC use?

OpenMAIC imports from @langchain/langgraph using the modern Annotation.Root API and StateGraph class. This corresponds to LangGraph v0.2.x, which introduced the annotation-based state definition pattern replacing the older Channel API.

Can I extend the graph to support multiple agent turns?

The current topology in director-graph.ts hardcodes a single round: director → agent_generate → END. To support multi-turn conversations, you would modify the conditional edge to route agent_generate back to director instead of END, adding termination logic based on turnCount or conversation goals.

How does OpenMAIC handle graph errors and cancellation?

The statelessGenerate function accepts an AbortSignal passed directly to graph.stream({ signal }). This leverages LangGraph's native cancellation support. For node-level error handling, each async node function includes try/catch blocks that emit error events via config.writer before re-throwing.

Where are the agent definitions configured?

Agent configurations (capabilities, system prompts, tool bindings) are passed through buildInitialState() and stored in the availableAgentIds field of OrchestratorState. The actual agent implementations are imported from separate agent packages and invoked by ID within agentGenerateNode.

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