如何实现类LangChain LCEL管道的步骤类型兼容性检查？

完全可以实现这种类型检查，核心是通过泛型约束__or__方法的类型签名，让静态类型检查器（如Pyright、Mypy）能识别前后步骤的输入输出类型不匹配。

核心实现思路 #

定义泛型Runnable类时，明确__or__方法的参数必须是接收当前步骤输出类型的Runnable，返回组合后的新Runnable。这样类型检查器会在编译阶段直接报错，提示类型不兼容。

代码示例 #

from typing import Generic, TypeVar

# 定义泛型变量，In为输入类型，Out为输出类型，NewOut为链式调用后的新输出
In = TypeVar("In")
Out = TypeVar("Out")
NewOut = TypeVar("NewOut")

class Runnable(Generic[In, Out]):
    def __call__(self, input: In) -> Out:
        raise NotImplementedError("子类必须实现__call__方法")
    
    def __or__(self, other: "Runnable[Out, NewOut]") -> "Runnable[In, NewOut]":
        # 实现链式调用的逻辑：先执行当前步骤，再传入下一个步骤
        class ChainedRunnable(Runnable[In, NewOut]):
            def __call__(self, input: In) -> NewOut:
                return other(self(input))
        return ChainedRunnable()

# 示例步骤实现
class AddFive(Runnable[int, int]):
    def __call__(self, input: int) -> int:
        return input + 5

class Parse(Runnable[str, str]):
    def __call__(self, input: str) -> str:
        return input.strip()

# 初始化实例
add_five = AddFive()
parse = Parse()

# 此处会触发类型检查错误：
# Argument of type "Runnable[str, str]" cannot be assigned to parameter "other" of type "Runnable[int, NewOut]"
chain = add_five | parse

关键说明 #

1. 泛型约束的作用：__or__方法的参数other被约束为Runnable[Out, NewOut]，其中Out是当前Runnable的输出类型，确保下一个步骤的输入类型必须和当前步骤的输出类型完全匹配。
2. 类型检查工具依赖：Python本身是动态类型语言，这种检查需要静态类型检查工具（如Pyright、Mypy）配合使用，确保在编写代码时就能捕获不兼容问题。
3. 适配第三方库：如果使用的是第三方框架（如LangChain）的Runnable，可以通过封装一层自定义泛型类，或者利用Protocol补充类型约束，强制__or__方法遵循类型匹配规则。

内容的提问来源于stack exchange，提问作者BPDev