decode method

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

Implementation

Stream<ChatResponseUpdate> decode(Stream<String> tokens) async* {
  // Hold back the longest-marker-minus-one trailing chars so the
  // `<|tool_call_start|>` marker is never split across two pieces and
  // emitted as content.
  final holdback = _callOpen.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 callAt = buf.indexOf(_callOpen);
    if (callAt >= 0) {
      final prose = buf.substring(0, callAt);
      if (prose.isNotEmpty) yield _text(prose);
      tail.write(buf.substring(callAt));
      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) {
    // A truncated or malformed tool call (e.g. the run hit maxTokens
    // mid-call) must not error the whole turn; surface the raw tail as text
    // instead so the user sees what the model produced.
    Lfm2Turn turn;
    try {
      turn = template.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);
  }
}