stream method

@override Stream<ChatChunk> stream(ChatRequest request, { AgenticContext? context, })

Generates an answer incrementally.

The stream ends after a chunk with a ChatChunk.finishReason. Cancelling the subscription must close the underlying connection: an abandoned streaming request that keeps generating is billed in full.

Collect the result with stream(request).collect() when you want the streaming transport but not the incremental updates.

Implementation

@override
Stream<ChatChunk> stream(
  ChatRequest request, {
  AgenticContext? context,
}) async* {
  _requests.add(request);
  if (latency > Duration.zero) {
    await (context?.clock ?? const SystemClock()).delay(latency);
  }
  context?.throwIfCancelled();

  final turn = _nextTurn();
  if (turn.error case final error?) throw error;

  if (turn.chunks case final chunks?) {
    for (final chunk in chunks) {
      context?.throwIfCancelled();
      yield chunk;
    }
    return;
  }

  final response = turn.response!;
  final message = response.message;
  if (message.reasoning case final reasoning?) {
    yield ChatChunk.reasoning(reasoning);
  }
  // Split into words so a test can observe more than one chunk without the
  // script having to spell them out.
  for (final word in _wordsOf(message.text)) {
    context?.throwIfCancelled();
    yield ChatChunk.text(word);
  }
  for (var i = 0; i < message.toolCalls.length; i++) {
    final call = message.toolCalls[i];
    yield ChatChunk.tool(
      ToolCallDelta(
        index: i,
        id: call.id,
        name: call.name,
        argumentsDelta: call.argumentsJson,
      ),
    );
  }
  yield ChatChunk(
    finishReason: response.finishReason,
    usage: response.usage,
    modelId: response.modelId,
  );
}