How to Stream Results from the OpenMAIC Orchestration Graph

OpenMAIC streams multi-agent results by attaching a streaming handler to LangGraph's StateGraph in stateless-generate.ts, which forwards partial AgentTurnSummary objects to clients via WebSocket/SSE as each agent produces tokens.

OpenMAIC is a multi-agent AI platform built on LangGraph that enables complex orchestration workflows. Streaming results from its orchestration graph keeps users engaged with real-time "thinking" indicators and progressive content rendering. This guide explains the exact mechanism—based on the THU-MAIC/OpenMAIC source code—so you can implement or debug streaming in your own deployment.

Core Architecture: Stateless Generator with Streaming Hooks

The streaming pipeline centers on lib/orchestration/stateless-generate.ts. This file constructs a fresh LangGraph StateGraph for each request and wires streaming into the graph's execution lifecycle.

Building the Graph and Attaching the Handler

When a StatelessChatRequest arrives, the generator:

1. Creates a new StateGraph with the specified agent IDs
2. Registers a streaming handler on the graph's onRun hook
3. Executes the orchestration loop

// lib/orchestration/stateless-generate.ts (conceptual flow)
const graph = new StateGraph<OrchestrationState>({
  channels: stateChannels,
})
  .addNode("director", directorNode)
  .addNode("agent", agentNode)
  .addEdge("director", "agent");

// Streaming handler attached during graph.run()
graph.onRun((event) => {
  if (event.type === "message_update") {
    const summary: AgentTurnSummary = {
      ...event.payload,
      streaming: true,  // Indicates partial result
    };
    streamToClient(summary);
  }
});

The handler receives message_update events containing partial text payloads. Each event becomes an AgentTurnSummary with streaming: true while the agent is still generating, and streaming: false when the turn completes.

Director Graph: Annotating Streams for UI Consumption

The lib/orchestration/director-graph.ts file defines the director agent that coordinates multi-agent workflows. It extends base streaming with UI-specific annotations:

• Each node is decorated with data-streaming="true" attributes
• These markers allow the frontend to identify which graph nodes are actively producing output

// lib/orchestration/director-graph.ts
const directorNode = async (state: OrchestrationState) => {
  const result = await directorAgent.invoke(state.messages);
  
  return {
    messages: [result],
    // Annotated for streaming detection
    __metadata: { "data-streaming": "true" },
  };
};

Client-Side Streaming Consumption

The workbench components in lib/workbench/ handle streamed results with progressive rendering capabilities.

TextBlock: Progressive Content Rendering

TextBlock.tsx consumes partial AgentTurnSummary objects and renders them incrementally:

// lib/workbench/TextBlock.tsx (simplified)
interface TextBlockProps {
  content: string;
  streaming?: boolean;
}

export function TextBlock({ content, streaming }: TextBlockProps) {
  return (
    <div 
      data-streamdown={streaming ? "true" : undefined}
      className={streaming ? "streaming" : "settled"}
    >
      {/* Renders math, code blocks, tables progressively */}
      <MarkdownRenderer content={content} />
    </div>
  );
}

The data-streamdown attribute triggers specialized parsers for LaTeX formulas, code blocks, and Markdown tables as soon as sufficient characters arrive—no waiting for the full message.

Thinking Bar: Visual Feedback During Streaming

The thinking bar component (thinking-bar-state.test.ts validates this behavior) shows:

• "思考中…" (thinking) — when any streaming: true message is active
• "已思考" (done) — when streaming: false signals completion

Complete Streaming Flow

Stage	Component	Key Action
1. Request	hooks/use-scene-generator.ts	Calls statelessGenerate() with agent list
2. Graph Build	stateless-generate.ts	Constructs StateGraph, attaches onRun handler
3. Execution	LangGraph runtime	Emits message_update events per token
4. Transform	Graph handler	Wraps events in AgentTurnSummary with streaming flag
5. Transport	WebSocket/SSE channel	Pushes to client (verified in owner-session-client.test.ts)
6. Render	TextBlock.tsx	Progressive display with data-streamdown
7. Complete	Thinking bar	Switches from "thinking" to "done" state

Type Definitions

Shared types in lib/orchestration/types.ts define the streaming contract:

// lib/orchestration/types.ts
export interface AgentTurnSummary {
  agentId: string;
  content: string;
  role: "assistant" | "thinking";
  streaming: boolean;  // true = partial, false = final
  timestamp: number;
}

The streaming boolean is the critical discriminator. Your client logic should:

• Append content to existing UI elements when streaming: true
• Finalize and create new elements when streaming: false

Triggering Orchestration from Components

The use-scene-generator.ts hook demonstrates how user-facing features initiate streamed orchestration:

// lib/hooks/use-scene-generator.ts
export function useSceneGenerator() {
  const generateMedia = async (outlines: Outline[]) => {
    const stream = await statelessGenerate({
      agentIds: ["scene-visualizer", "media-generator"],
      prompt: buildPrompt(outlines),
      enableStreaming: true,
    });
    
    for await (const summary of stream) {
      updateWorkbench(summary);  // Real-time UI updates
    }
  };
  
  return { generateMedia };
}

Testing Streaming Behavior

The OpenMAIC test suite validates streaming through three key files:

• tests/workbench/text-block.test.ts — Confirms data-streamdown attributes handle partial LaTeX, code, and tables correctly
• tests/workbench/thinking-bar-state.test.ts — Verifies state transitions between thinking/done indicators
• tests/owner-session-client.test.ts — Ensures stream lifecycle: retired streams close cleanly, new streams open without data loss, and cursor-free reconnection works

Summary

• Streaming originates in lib/orchestration/stateless-generate.ts — The generator builds a fresh LangGraph StateGraph per request and attaches an onRun handler that forwards partial results
• streaming: true/false flags distinguish partial from final messages in AgentTurnSummary objects defined in types.ts
• The director graph in director-graph.ts adds UI annotations (data-streaming="true") for node-level stream tracking
• Workbench components consume streams progressively — TextBlock.tsx uses data-streamdown for incremental rendering of complex content
• WebSocket/SSE transports verified in owner-session-client.test.ts handle reliable server-to-client delivery

Frequently Asked Questions

How do I enable streaming for a new orchestration workflow?

Pass enableStreaming: true in your StatelessChatRequest and ensure your client handles AgentTurnSummary objects with the streaming flag. The generator in stateless-generate.ts automatically attaches the streaming handler when this option is set.

What determines when a message switches from streaming: true to streaming: false?

The LangGraph runtime emits message_update events continuously as tokens arrive. The streaming handler sets streaming: true for each event. When the agent node completes execution (determined by the LangGraph state machine), the final event receives streaming: false to signal completion.

Can I stream from multiple agents simultaneously?

Yes. The director graph orchestrates multiple agents in parallel, and each emits independent AgentTurnSummary streams. The agentId field in each summary lets the client route updates to the correct UI element. The thinking bar aggregates all active streams and clears only when every agent reports streaming: false.

How does OpenMAIC prevent data loss during stream reconnection?

The owner-session-client.test.ts test suite verifies cursor-free stream reconciliation. The server maintains no per-client state (stateless design), but the client tracks received message IDs. On reconnection, the client requests a fresh stream starting from the last confirmed ID, and the server replays any missed AgentTurnSummary objects before resuming live updates.

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