feat: Add reduceOutputStream option to StringOutputParser (#368)

packages/langchain_core/lib/src/chat_models/fake.dart

@@ -29,6 +29,24 @@ class FakeChatModel extends SimpleChatModel {
     return Future<String>.value(responses[_i++ % responses.length]);
   }
 
+  @override
+  Stream<ChatResult> stream(
+    final PromptValue input, {
+    final ChatModelOptions? options,
+  }) {
+    final res = responses[_i++ % responses.length].split('');
+    return Stream.fromIterable(res).map(
+      (final char) => ChatResult(
+        id: 'fake-chat-model',
+        output: AIChatMessage(content: char),
+        finishReason: FinishReason.stop,
+        metadata: const {},
+        usage: const LanguageModelUsage(),
+        streaming: true,
+      ),
+    );
+  }
+
   @override
   Future<List<int>> tokenize(
     final PromptValue promptValue, {

packages/langchain_core/lib/src/output_parsers/string.dart

@@ -2,6 +2,7 @@ import '../../llms.dart';
 import '../chat_models/types.dart';
 import '../documents/document.dart';
 import '../language_models/types.dart';
+import '../runnables/runnable.dart';
 import 'base.dart';
 import 'types.dart';
 
@@ -34,21 +35,55 @@ import 'types.dart';
 class StringOutputParser<ParserInput extends Object?>
     extends BaseOutputParser<ParserInput, OutputParserOptions, String> {
   /// {@macro string_output_parser}
-  const StringOutputParser()
-      : super(defaultOptions: const OutputParserOptions());
+  const StringOutputParser({
+    this.reduceOutputStream = false,
+  }) : super(defaultOptions: const OutputParserOptions());
+
+  /// When invoking this parser with [Runnable.stream], every item from the
+  /// input stream will be parsed and emitted by default.
+  ///
+  /// If [reduceOutputStream] is set to `true`, the parser will reduce the
+  /// output stream into a single String and emit it as a single item. This is
+  /// useful when the next [Runnable] in a chain expects a single String as
+  /// input.
+  ///
+  /// Visual example:
+  /// - reduceOutputStream = false
+  /// 'A', 'B', 'C' -> 'A', 'B', 'C'
+  /// - reduceOutputStream = true
+  /// 'A', 'B', 'C' -> 'ABC'
+  final bool reduceOutputStream;
 
   @override
   Future<String> invoke(
     final ParserInput input, {
     final OutputParserOptions? options,
   }) {
+    return Future.value(_parse(input));
+  }
+
+  @override
+  Stream<String> streamFromInputStream(
+    final Stream<ParserInput> inputStream, {
+    final OutputParserOptions? options,
+  }) async* {
+    if (reduceOutputStream) {
+      yield await inputStream.map(_parse).reduce((final a, final b) => '$a$b');
+    } else {
+      await for (final input in inputStream) {
+        yield _parse(input);
+      }
+    }
+  }
+
+  String _parse(final ParserInput input) {
     final output = switch (input) {
       null => '',
       final LanguageModelResult res => res.outputAsString,
       final ChatMessage res => res.contentAsString,
       final Document res => res.pageContent,
       _ => input.toString(),
     };
-    return Future.value(output);
+    return output;
   }
 }

packages/langchain_core/test/output_parsers/string_test.dart

@@ -3,6 +3,7 @@ import 'package:langchain_core/chat_models.dart';
 import 'package:langchain_core/language_models.dart';
 import 'package:langchain_core/llms.dart';
 import 'package:langchain_core/output_parsers.dart';
+import 'package:langchain_core/prompts.dart';
 import 'package:test/test.dart';
 
 void main() {
@@ -30,5 +31,20 @@ void main() {
       final res = await const StringOutputParser().invoke(result);
       expect(res, 'Hello world!');
     });
+
+    test('Test reduceOutputStream', () async {
+      final chat = FakeChatModel(responses: ['ABC']);
+
+      final chain1 =
+          chat.pipe(const StringOutputParser(reduceOutputStream: false));
+      final chain2 =
+          chat.pipe(const StringOutputParser(reduceOutputStream: true));
+
+      final res1 = await chain1.stream(PromptValue.string('test')).toList();
+      final res2 = await chain2.stream(PromptValue.string('test')).toList();
+
+      expect(res1, ['A', 'B', 'C']);
+      expect(res2, ['ABC']);
+    });
   });
 }