How OpenMAIC Handles Structured Output Parsing for Streaming LLM Responses

OpenMAIC processes streaming LLM responses through a four-stage pipeline that accumulates raw SSE chunks into an incremental buffer, parses complete JSON objects using a brace-depth state machine, validates them against a Zod DSL schema, and injects the results into a LangGraph orchestration graph.

OpenMAIC is an open-source classroom content generation system that leverages large language models with streaming enabled to deliver real-time feedback. When working with continuous token streams from providers like OpenAI, the platform must transform fragmented text chunks into type-safe, structured data without blocking the user experience. This article examines how structured output parsing is implemented across the OpenMAIC codebase to handle partial JSON objects that span multiple network chunks.

The Four-Stage Streaming Architecture

The implementation spans four distinct phases, moving from raw network bytes to validated domain objects ready for agent orchestration.

Stage 1: Establishing the SSE Stream in app/api/chat/route.ts

The server initiates a streaming request to the LLM provider using the provider's SDK with stream: true enabled. This creates a Server-Sent Events (SSE) connection that proxies chunked responses to the client as they arrive from the model. The app/api/chat/route.ts file handles this proxying logic, ensuring that partial tokens flow immediately to the frontend rather than waiting for the complete response.

Stage 2: Incremental Buffer Accumulation in hooks/useChat.ts

On the client side, hooks/useChat.ts consumes the SSE stream using the native EventSource API. The hook maintains an in-memory string buffer that appends each incoming data chunk. After every update, it forwards the accumulated buffer to the parser, allowing the system to handle JSON structures split across arbitrary chunk boundaries.

// hooks/useChat.ts – reads SSE and feeds the parser
const source = new EventSource(url);
source.onmessage = (e) => {
  buffer += e.data;                     // 1️⃣ accumulate
  const parsed = parseStructured(buffer); // 2️⃣ try to extract JSON objects
  if (parsed) {
    setMessages((old) => [...old, parsed]); // 3️⃣ hand over to LangGraph
    buffer = "";                         // reset after a full object
  }
};

Stage 3: Brace-Depth State Machine Parsing

The core structured output parsing logic lives in lib/orchestration/structuredOutputParser.ts. Instead of attempting to parse the entire buffer on every chunk, the parser implements a state machine that tracks brace and bracket depth. It iterates through the buffer character by character, incrementing a counter on { or [ and decrementing on } or ]. When the depth returns to zero, the parser extracts the substring bounded by matching braces and attempts JSON.parse() inside a try-catch block.

Stage 4: LangGraph Integration via lib/orchestration/graph.ts

Successfully parsed objects flow into lib/orchestration/graph.ts, where they are injected into the LangGraph state machine. Each parsed message—whether a tool call or assistant response—becomes a node in the execution graph, enabling the workbench to resume, steer, or replay sessions deterministically.

Implementing the Structured Output Parser State Machine

The parseStructured function in lib/orchestration/structuredOutputParser.ts demonstrates the sliding-window approach. It scans the accumulated text without resetting the buffer until a valid, schema-compliant object is found.

// lib/orchestration/structuredOutputParser.ts
export function parseStructured(text: string): ParsedMessage | null {
  let depth = 0, start = -1;
  for (let i = 0; i < text.length; i++) {
    if (text[i] === '{' || text[i] === '[') {
      if (depth === 0) start = i;
      depth++;
    } else if (text[i] === '}' || text[i] === ']') {
      depth--;
      if (depth === 0 && start !== -1) {
        const candidate = text.slice(start, i + 1);
        try {
          const obj = JSON.parse(candidate);
          // Validate against DSL schema
          if (dslSchema.safeParse(obj).success) return obj as ParsedMessage;
        } catch {}
        start = -1; // reset on failure
      }
    }
  }
  return null;
}

This implementation is robust to nested objects and arrays because it tracks both bracket types. The parser only returns when depth reaches zero, ensuring that incomplete structures spanning multiple SSE chunks remain in the buffer for the next iteration.

DSL Schema Validation and Type Safety

All parsed candidates are validated against the canonical domain schema defined in packages/@openmaic/dsl using Zod. The dslSchema.safeParse() call ensures that only objects matching the expected shapes—slides, quizzes, or PBL tasks—are emitted into the orchestration layer. Invalid fragments are logged and discarded, preventing malformed data from corrupting the agent's state graph.

Summary

• OpenMAIC uses a four-stage pipeline to transform streaming LLM responses into structured data
• The brace-depth state machine in lib/orchestration/structuredOutputParser.ts handles JSON objects split across multiple chunks without external dependencies
• Incremental buffering in hooks/useChat.ts accumulates SSE data using the native EventSource API
• Zod schema validation against packages/@openmaic/dsl guarantees type safety before LangGraph integration in lib/orchestration/graph.ts
• The architecture enables real-time streaming feedback while maintaining deterministic replay capabilities through LangGraph state management

Frequently Asked Questions

How does OpenMAIC handle incomplete JSON objects that span multiple chunks?

The parser maintains a sliding window buffer that accumulates all incoming text from the SSE stream. It only attempts extraction when the brace-depth counter returns to zero, indicating a complete JSON structure. This allows the system to reconstruct objects that arrive fragmented across dozens of network events without data loss.

What happens if the LLM produces malformed JSON during streaming?

Malformed candidates fail either the JSON.parse() call inside the try/catch block or the subsequent dslSchema.safeParse() validation in lib/orchestration/structuredOutputParser.ts. These fragments are logged for debugging and silently discarded, ensuring only valid, schema-compliant objects enter the LangGraph state machine.

Why use a brace-counting state machine instead of a streaming JSON parser?

The brace-depth approach is lightweight, requires no external dependencies beyond native JSON.parse(), and is specifically optimized for the pattern of complete JSON objects concatenated in a stream. It efficiently handles nested structures while avoiding the overhead of incremental parsers that must track internal state for partial primitives.

Can the structured output parser handle streaming arrays of objects?

Yes, the state machine in structuredOutputParser.ts tracks both {/} braces and [/] brackets. When processing an array of classroom content objects, it correctly identifies when individual objects within the array complete or when the outer array closes, depending on whether the LLM streams an array wrapper or individual newline-delimited objects.

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