decode method

Stream<ChatResponseUpdate> decode(Stream<String> tokens)

Splits tokens into prose and tool-call updates.

Implementation

Stream<ChatResponseUpdate> decode(Stream<String> tokens) async* {
  // Hold back the marker-minus-one trailing chars so a marker straddling two
  // pieces is never emitted as content.
  final holdback = openMarker.length - 1;

  var buf = '';
  final tail = StringBuffer();
  var buffering = false;

  await for (final piece in tokens) {
    if (buffering) {
      tail.write(piece);
      continue;
    }
    buf += piece;

    final at = buf.indexOf(openMarker);
    if (at >= 0) {
      final prose = buf.substring(0, at);
      if (prose.isNotEmpty) yield _text(prose);
      tail.write(buf.substring(at));
      buf = '';
      buffering = true;
      continue;
    }
    if (buf.length > holdback) {
      final emit = buf.substring(0, buf.length - holdback);
      if (emit.isNotEmpty) yield _text(emit);
      buf = buf.substring(buf.length - holdback);
    }
  }

  if (buffering) {
    ParsedTurn turn;
    try {
      turn = parse(tail.toString());
    } on FormatException {
      yield _text(tail.toString());
      return;
    }
    if (turn.text.isNotEmpty) yield _text(turn.text);
    if (turn.calls.isNotEmpty) {
      yield ChatResponseUpdate(
        role: ChatRole.assistant,
        contents: List<AIContent>.of(turn.calls),
      );
    }
  } else if (buf.isNotEmpty) {
    yield _text(buf);
  }
}