generateStructuredStream static method

Stream<StructuredOutputStreamEvent> generateStructuredStream({ required String prompt, required JSONSchema schema, LLMGenerationOptions? options, })

Stream-shaped structured output API. Mirrors Swift RunAnywhere.generateStructuredStream(prompt:schema:options:) — drives the underlying LLM token stream and re-emits one STRUCTURED_OUTPUT_STREAM_EVENT_KIND_TOKEN event per non-empty token, then a terminal COMPLETED event carrying the schema-validated StructuredOutputResult parsed from the accumulated text by commons (extractStructuredOutput). Errors surface as a single terminal ERROR event so consumers always receive a terminal frame.

Implementation

static Stream<StructuredOutputStreamEvent> generateStructuredStream({
  required String prompt,
  required JSONSchema schema,
  LLMGenerationOptions? options,
}) {
  final controller = StreamController<StructuredOutputStreamEvent>();
  StreamSubscription<LLMStreamEvent>? subscription;
  var seq = Int64.ZERO;
  final accumulated = StringBuffer();

  StructuredOutputStreamEvent makeEvent(
    StructuredOutputStreamEventKind kind, {
    String? token,
    StructuredOutputResult? result,
    String? errorMessage,
  }) {
    seq += Int64.ONE;
    final event = StructuredOutputStreamEvent(kind: kind, seq: seq);
    if (token != null) event.token = token;
    if (result != null) event.result = result;
    if (errorMessage != null) event.errorMessage = errorMessage;
    return event;
  }

  Future<void> emitTerminalError(Object error) async {
    if (controller.isClosed) return;
    controller.add(
      makeEvent(
        StructuredOutputStreamEventKind
            .STRUCTURED_OUTPUT_STREAM_EVENT_KIND_ERROR,
        errorMessage: error.toString(),
      ),
    );
    await controller.close();
  }

  Future<void> start() async {
    if (!DartBridge.isInitialized) {
      await emitTerminalError(SDKException.notInitialized());
      return;
    }
    try {
      final effectiveOptions = LLMGenerationOptions();
      if (options != null) effectiveOptions.mergeFromMessage(options);
      effectiveOptions.structuredOutput =
          StructuredOutputOptionsDefaults.defaults(schema: schema);

      final llmStream = RunAnywhereLLM.shared.generateStream(
        prompt,
        effectiveOptions,
      );

      subscription = llmStream.listen(
        (event) {
          if (controller.isClosed) return;
          if (event.token.isNotEmpty) {
            accumulated.write(event.token);
            controller.add(
              makeEvent(
                StructuredOutputStreamEventKind
                    .STRUCTURED_OUTPUT_STREAM_EVENT_KIND_TOKEN,
                token: event.token,
              ),
            );
          }
        },
        onError: (Object error) {
          unawaited(emitTerminalError(error));
        },
        onDone: () async {
          if (controller.isClosed) return;
          try {
            final result = RunAnywhereLLM.shared.extractStructuredOutput(
              text: accumulated.toString(),
              schema: schema,
            );
            controller.add(
              makeEvent(
                StructuredOutputStreamEventKind
                    .STRUCTURED_OUTPUT_STREAM_EVENT_KIND_COMPLETED,
                result: result,
              ),
            );
            await controller.close();
          } catch (e) {
            await emitTerminalError(e);
          }
        },
        cancelOnError: true,
      );
    } catch (e) {
      await emitTerminalError(e);
    }
  }

  controller.onListen = () => unawaited(start());
  controller.onCancel = () async {
    // Consumer cancelled mid-stream (controller still open): tear down the
    // native LLM generation, mirroring Swift's `.cancelled` termination
    // branch (RunAnywhere+StructuredOutput.swift:122-126). When the
    // controller already closed (terminal COMPLETED/ERROR emitted), this is
    // a `.finished` termination — do NOT fire the native cancel, which
    // would race a follow-up generate() on the lifecycle LLM handle.
    final cancelledMidStream = !controller.isClosed;
    await subscription?.cancel();
    subscription = null;
    if (cancelledMidStream) {
      RunAnywhereLLM.shared.cancelGeneration();
    }
  };
  return controller.stream;
}