feat(lcel): Add support for Runnable.mapInput() (#229)

A `RunnableMapInput` allows you to map the input to a different value.

You can create a `RunnableMapInput` using the `Runnable.mapInput` static method.

When you call `invoke` on a `RunnableMapInput`, it will take the input it receives and returns the output returned by the given `inputMapper` function.

Example:

```dart
final agent = Agent.fromRunnable(
  Runnable.mapInput(
    (final AgentPlanInput planInput) => <String, dynamic>{
      'input': planInput.inputs['input'],
      'agent_scratchpad': buildScratchpad(planInput.intermediateSteps),
    },
  ).pipe(prompt).pipe(model).pipe(outputParser),
  tools: [tool],
);
```

docs/expression_language/interface.md

@@ -25,7 +25,8 @@ The type of the input and output varies by component:
 | `RunnableFunction`          | Runnable input type    | Runnable output type   |
 | `RunnablePassthrough`       | Runnable input type    | Runnable input type    |
 | `RunnableItemFromMap`       | `Map<String, dynamic>` | Runnable output type   |
-| `RunnableMapFromInput`       | Runnable input type    | `Map<String, dynamic>` |
+| `RunnableMapFromInput`      | Runnable input type    | `Map<String, dynamic>` |
+| `RunnableMapInput`          | Runnable input type    | Runnable output type   |
 
 You can combine `Runnable` objects into sequences in three ways:
 
@@ -351,3 +352,25 @@ final res = await chain.invoke('bears');
 print(res);
 // Why don't bears wear shoes? Because they have bear feet!
 ```
+
+### RunnableMapInput
+
+A `RunnableMapInput` allows you to map the input to a different value.
+
+You can create a `RunnableMapInput` using the `Runnable.mapInput` static method.
+
+When you call `invoke` on a `RunnableMapInput`, it will take the input it receives and returns the output returned by the given `inputMapper` function.
+
+Example:
+
+```dart
+final agent = Agent.fromRunnable(
+  Runnable.mapInput(
+    (final AgentPlanInput planInput) => <String, dynamic>{
+      'input': planInput.inputs['input'],
+      'agent_scratchpad': buildScratchpad(planInput.intermediateSteps),
+    },
+  ).pipe(prompt).pipe(model).pipe(outputParser),
+  tools: [tool],
+);
+```

examples/docs_examples/bin/expression_language/interface.dart

@@ -18,6 +18,7 @@ void main(final List<String> arguments) async {
   await _runnableTypesRunnablePassthrough();
   await _runnableTypesRunnableItemFromMap();
   await _runnableTypesRunnableMapFromInput();
+  await _runnableTypesRunnableMapInput();
 }
 
 Future<void> _runnableInterfaceInvoke() async {
@@ -250,3 +251,55 @@ Future<void> _runnableTypesRunnableMapFromInput() async {
   print(res);
   // Why don't bears wear shoes? Because they have bear feet!
 }
+
+Future<void> _runnableTypesRunnableMapInput() async {
+  final openaiApiKey = Platform.environment['OPENAI_API_KEY'];
+
+  final prompt = ChatPromptTemplate.fromPromptMessages([
+    SystemChatMessagePromptTemplate.fromTemplate(
+      'You are a helpful assistant',
+    ),
+    HumanChatMessagePromptTemplate.fromTemplate('{input}'),
+    const MessagesPlaceholder(variableName: 'agent_scratchpad'),
+  ]);
+
+  final tool = CalculatorTool();
+
+  final model = ChatOpenAI(
+    apiKey: openaiApiKey,
+    temperature: 0,
+  ).bind(ChatOpenAIOptions(functions: [tool.toChatFunction()]));
+
+  const outputParser = OpenAIFunctionsAgentOutputParser();
+
+  List<ChatMessage> buildScratchpad(final List<AgentStep> intermediateSteps) {
+    return intermediateSteps
+        .map((final s) {
+          return s.action.messageLog +
+              [
+                ChatMessage.function(
+                  name: s.action.tool,
+                  content: s.observation,
+                ),
+              ];
+        })
+        .expand((final m) => m)
+        .toList(growable: false);
+  }
+
+  final agent = Agent.fromRunnable(
+    Runnable.mapInput(
+      (final AgentPlanInput planInput) => <String, dynamic>{
+        'input': planInput.inputs['input'],
+        'agent_scratchpad': buildScratchpad(planInput.intermediateSteps),
+      },
+    ).pipe(prompt).pipe(model).pipe(outputParser),
+    tools: [tool],
+  );
+  final executor = AgentExecutor(agent: agent);
+
+  final res = await executor.invoke({
+    'input': 'What is 40 raised to the 0.43 power?',
+  });
+  print(res['output']);
+}

packages/langchain/lib/src/core/runnable/base.dart

@@ -6,6 +6,7 @@ import '../base.dart';
 import 'binding.dart';
 import 'function.dart';
 import 'input_getter.dart';
+import 'input_map.dart';
 import 'map.dart';
 import 'passthrough.dart';
 import 'sequence.dart';
@@ -93,6 +94,15 @@ abstract class Runnable<RunInput extends Object?,
     return RunnableMapFromInput<RunInput>(key);
   }
 
+  /// Creates a [RunnableMapInput] which allows you to map the input to a
+  /// different value.
+  static Runnable<RunInput, BaseLangChainOptions, RunOutput>
+      mapInput<RunInput extends Object, RunOutput extends Object>(
+    final RunOutput Function(RunInput input) inputMapper,
+  ) {
+    return RunnableMapInput<RunInput, RunOutput>(inputMapper);
+  }
+
   /// Invokes the [Runnable] on the given [input].
   ///
   /// - [input] - the input to invoke the [Runnable] on.

packages/langchain/lib/src/core/runnable/input_map.dart

@@ -0,0 +1,47 @@
+import '../base.dart';
+import 'base.dart';
+
+/// {@template runnable_map_input}
+/// A [RunnableMapInput] allows you to map the input to a different value.
+///
+/// You can create a [RunnableMapInput] using the [Runnable.mapInput] static
+/// method.
+///
+/// When you call [invoke] on a [RunnableMapInput], it will take the
+/// input it receives and returns the output returned by the given
+/// [inputMapper] function.
+///
+/// Example:
+///
+/// ```dart
+/// final agent = Agent.fromRunnable(
+///   Runnable.mapInput(
+///     (final AgentPlanInput planInput) => <String, dynamic>{
+///       'input': planInput.inputs['input'],
+///       'agent_scratchpad': buildScratchpad(planInput.intermediateSteps),
+///     },
+///   ).pipe(prompt).pipe(model).pipe(outputParser),
+///   tools: [tool],
+/// );
+/// ```
+/// {@endtemplate}
+class RunnableMapInput<RunInput extends Object, RunOutput extends Object>
+    extends Runnable<RunInput, BaseLangChainOptions, RunOutput> {
+  /// {@macro runnable_map_from_input_items}
+  const RunnableMapInput(this.inputMapper);
+
+  /// A function that maps [RunInput] to [RunOutput].
+  final RunOutput Function(RunInput input) inputMapper;
+
+  /// Invokes the [RunnableMapInput] on the given [input].
+  ///
+  /// - [input] - the input to invoke the [RunnableMapInput] on.
+  /// - [options] - not used.
+  @override
+  Future<RunOutput> invoke(
+    final RunInput input, {
+    final BaseLangChainOptions? options,
+  }) async {
+    return inputMapper(input);
+  }
+}

packages/langchain/lib/src/core/runnable/runnable.dart

@@ -2,6 +2,7 @@ export 'base.dart';
 export 'extensions.dart';
 export 'function.dart';
 export 'input_getter.dart';
+export 'input_map.dart';
 export 'map.dart';
 export 'passthrough.dart';
 export 'sequence.dart';

packages/langchain/test/core/runnable/input_map_test.dart

@@ -0,0 +1,32 @@
+import 'package:langchain/langchain.dart';
+import 'package:test/test.dart';
+
+void main() {
+  group('RunnableMapInput tests', () {
+    test('RunnableMapInput from Runnable.getItemFromMap', () async {
+      final chain = Runnable.mapInput<ChainValues, ChainValues>(
+        (final input) => {
+          'input': '${input['foo']}${input['bar']}',
+        },
+      );
+
+      final res = await chain.invoke({'foo': 'foo1', 'bar': 'bar1'});
+      expect(res, {'input': 'foo1bar1'});
+    });
+
+    test('Streaming RunnableMapInput', () async {
+      final chain = Runnable.mapInput<ChainValues, ChainValues>(
+        (final input) => {
+          'input': '${input['foo']}${input['bar']}',
+        },
+      );
+      final stream = chain.stream({'foo': 'foo1', 'bar': 'bar1'});
+
+      final streamList = await stream.toList();
+      expect(streamList.length, 1);
+
+      final item = streamList.first;
+      expect(item, {'input': 'foo1bar1'});
+    });
+  });
+}