Stricter Type Handling in LCEL Provided by langchain-core

[CastaChick]

Checked

• I searched existing ideas and did not find a similar one
• I added a very descriptive title
• I've clearly described the feature request and motivation for it

Feature request

Stricter Type Handling in LCEL Provided by langchain-core

The pipe operator-based chaining of Runnables defined in LCEL is highly convenient; however, its type leniency poses challenges for production use. I recommend enforcing stricter type definitions for the arguments in the __or__, __ror__, and pipe functions within libs/core/langchain_core/runnables/base.py to enhance type consistency.

While this change could be breaking for some existing chains that currently work without type consistency, it is a necessary step for ensuring the safety and reliability of generative AI products in production environments.

Motivation

The current implementation permits chains that fail to function correctly with inputs due to type inconsistencies. For instance, the following chain does not raise a type error despite the inconsistency:

def to_str(x: int) -> str:
    return str(x)

def add_one(x: int) -> int:
    return x + 1

chain = RunnableLambda(to_str) | RunnableLambda(add_one)

In contrast, langchain-js on TypeScript identifies such issues at compile time, as demonstrated below:

import { RunnableLambda } from "@langchain/core/runnables";

const to_str = RunnableLambda.from((n: number) => {
  return n.toString()
})

const add_one = RunnableLambda.from((n: number) => {
  return n + 1
})

const chain = to_str.pipe(add_one)  // Raise a TypeError Here

The absence of strict type enforcement in LCEL complicates its use in production environments, where ensuring type safety is essential for stability and reliability. Modern Python development relies heavily on type checking tools such as mypy and plugins of IDEs to identify type inconsistencies in real-time, thereby enhancing code quality and maintainability.

Proposal (If applicable)

libs/core/langchain_core/runnables/base.py here

eg: __or__

def __or__(
      self,
      other: Union[
          Runnable[Output, NewOutput],
          Callable[[Output], NewOutput],
          Callable[[Iterator[Output]], Iterator[NewOutput]],
          # The modifications to the next line are likely to be somewhat complex, 
          # but the precise approach is not yet clear to me :(
          Mapping[str, Union[Runnable[Any, Other], Callable[[Any], Other], Any]],
      ],
  ) -> RunnableSerializable[Input, NewOutput]:
      """Compose this Runnable with another object to create a RunnableSequence."""
      return RunnableSequence(self, coerce_to_runnable(other))

ref: The type definition of langchain-js is below

pipe<NewRunOutput>(coerceable: RunnableLike<RunOutput, NewRunOutput>): Runnable<RunInput, Exclude<NewRunOutput, Error>>;