getStreamingResponse method

@override Stream<ChatResponseUpdate> getStreamingResponse({ required Iterable<ChatMessage> messages, ChatOptions? options, CancellationToken? cancellationToken, })

Sends a chat request and returns a stream of response updates.

Implementation

@override
Stream<ChatResponseUpdate> getStreamingResponse({
  required Iterable<ChatMessage> messages,
  ChatOptions? options,
  CancellationToken? cancellationToken,
}) async* {
  final session = await sessionProvider();
  // Resolved after the session so a formatResolver populated during model
  // load (e.g. from the GGUF's embedded chat template) applies to the very
  // first request.
  final activeFormat = formatResolver?.call() ?? format;
  final tools = (options?.tools ?? const <AITool>[])
      .whereType<AIFunctionDeclaration>();
  final thinking =
      (isThinkingEnabled?.call() ?? false) && activeFormat.supportsThinking;
  // Tiling happens before rendering so each crop gets its own media
  // marker, and before chatTurnsFromMessages so the message-level (web)
  // path sees the same crops. Deterministic per image, so re-tiling the
  // history each turn leaves the rendered prefix stable for KV reuse.
  var prepared = messagesWithInstructions(messages, options?.instructions);
  final tiling = imageTiling;
  if (tiling != null) {
    prepared = await tiledMessages(prepared, tiling);
  }
  final prompt = activeFormat.render(
    prepared,
    tools: tools,
    enableThinking: thinking,
  );

  // A model whose format did not emit any media marker for audio the caller
  // attached cannot "hear" it: the audio would be silently dropped and the
  // model asked to transcribe nothing. Fail loudly instead. (Gemma 4 does
  // collect audio, so this fires only for text/vision-only families; the
  // absent-audio-projector case surfaces as a clear runtime error deeper in.)
  if (prompt.media.isEmpty && _hasAudioContent(prepared)) {
    throw UnsupportedError(
      'This local model cannot accept audio input. Load an audio-capable '
      'model (Gemma 4 with an audio projector) to transcribe speech.',
    );
  }

  final maxTokens = options?.maxOutputTokens ?? sampling.maxTokens;
  final temperature = options?.temperature ?? sampling.temperature;
  final topK = options?.topK ?? sampling.topK;
  final topP = options?.topP ?? sampling.topP;
  final seed = options?.seed ?? sampling.seed;

  inspector?.record(
    PromptSnapshot(
      text: prompt.text,
      stopSequences: prompt.stopSequences,
      maxTokens: maxTokens,
      temperature: temperature,
      topK: topK,
      topP: topP,
      seed: seed,
      imageCount: prompt.media.length,
      contextSize: contextSize,
      capturedAt: DateTime.now(),
    ),
  );

  if (cancellationToken?.isCancellationRequested ?? false) return;

  _logger.logDebug(
    'Generating with ${activeFormat.runtimeType}: '
    '${prompt.text.length} prompt chars, ${prompt.media.length} media, '
    '${tools.length} tools, maxTokens: $maxTokens, thinking: $thinking.',
  );

  LlamaGenerationStats? stats;
  var tokens = session.generate(
    prompt.text,
    maxTokens: maxTokens,
    temperature: temperature,
    topK: topK,
    topP: topP,
    seed: seed,
    stopSequences: prompt.stopSequences,
    media: prompt.media.isEmpty ? null : prompt.media,
    turns: prompt.media.isEmpty ? null : chatTurnsFromMessages(prepared),
    onStats: (reported) => stats = reported,
  );
  // Cancellation is forwarded to the engine (which stops decoding within a
  // step and ends the stream) rather than only filtered here: takeWhile
  // alone would leave the engine generating until its next token arrived.
  CancellationTokenRegistration? cancelRegistration;
  if (cancellationToken != null) {
    cancelRegistration = cancellationToken.register(
      (_) => unawaited(session.cancel()),
    );
    tokens = tokens.takeWhile(
      (_) => !cancellationToken.isCancellationRequested,
    );
  }

  try {
    yield* activeFormat.decode(tokens);
  } finally {
    cancelRegistration?.dispose();
  }

  // The runtime reports token accounting on the done event, after the
  // token stream has drained; surface it the way cloud clients do, as a
  // trailing usage-only update.
  final reported = stats;
  if (reported != null) {
    _logger.logDebug(
      'Generation finished: ${reported.promptTokenCount} in '
      '(${reported.cachedTokenCount} cached), '
      '${reported.generatedTokenCount} out, '
      '${reported.finishReason?.name ?? 'unknown'}.',
    );
    final extraCounts = <String, int>{
      'prefillMicroseconds': ?reported.prefillDuration?.inMicroseconds,
      'decodeMicroseconds': ?reported.decodeDuration?.inMicroseconds,
      'draftedTokenCount': ?reported.draftedTokenCount,
      'acceptedTokenCount': ?reported.acceptedTokenCount,
    };
    yield ChatResponseUpdate(
      role: ChatRole.assistant,
      finishReason: _chatFinishReason(reported.finishReason),
      usage: UsageDetails(
        inputTokenCount: reported.promptTokenCount,
        outputTokenCount: reported.generatedTokenCount,
        totalTokenCount:
            reported.promptTokenCount + reported.generatedTokenCount,
        cachedInputTokenCount: reported.cachedTokenCount,
        additionalCounts: extraCounts.isEmpty ? null : extraCounts,
      ),
    );
  }
}