如何让Python的pipeline函数类型标注匹配最后一个函数的输出？

我想要编写一个pipeline函数，它接收任意数量的函数，将前一个函数的输出依次传递给下一个，直到最后一个。pipeline的输出是一个Callable，该Callable的输入与传入pipeline的第一个函数匹配，输出与传入的最后一个函数匹配。

目前的实现和类型标注如下：

from typing import TypeVar
from typing_extensions import ParamSpec, Concatenate

P = ParamSpec("P")
Output = TypeVar("Output")

def pipeline(func: Callable[Concatenate[P], T], *funcs: Callable[..., Output]) -> Callable[Concatenate[P], Output]:
    ...

但这会导致*funcs所有Output类型被合并为Union。举个例子：

from typing import Tuple

def double(x: int) -> Tuple[int, int]:
    return x, x*2

def add(x: int, y: int) -> int:
    return x + y

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

new_func = pipeline(add, double, add, to_string) 

使用上述类型标注时，new_func的类型签名被错误推断为：

new_func: (x: int, y: int) -> (Tuple[int, int] | int | str) 

显然正确的签名应该是：

new_func: (x: int, y: int) -> str

请问是否有方法通过类型标注实现这一效果？

要实现流水线函数的精确类型推断，需要让类型系统跟踪每一步函数的输入输出类型衔接。以下是两种可行方案：

方案一：递归泛型（兼容Python 3.10+） #

借助typing_extensions的TypeVarTuple和Unpack，可以定义递归的类型约束，确保每个后续函数的输入匹配前一个函数的输出：

from typing import Callable, TypeVar, Tuple
from typing_extensions import ParamSpec, TypeVarTuple, Unpack

P = ParamSpec("P")
R = TypeVar("R")
Fs = TypeVarTuple("Fs")

def pipeline(
    first: Callable[P, R],
    *funcs: Unpack[Tuple[Callable[[R], R], ...]]
) -> Callable[P, R]:
    def composed(*args: P.args, **kwargs: P.kwargs) -> R:
        result = first(*args, **kwargs)
        for func in funcs:
            result = func(result)
        return result
    return composed

这个方案要求所有中间函数都是单参数的（接收前一个函数的输出作为唯一输入），完美契合流水线的执行逻辑。用示例测试时，new_func的类型会被正确推断为(x: int, y: int) -> str。

方案二：Python 3.12+ 泛型展开（更直观） #

Python 3.12引入的泛型展开语法*，结合递归类型可以更清晰地跟踪每一步的类型转换：

from typing import Callable, TypeVar, ParamSpec
from typing_extensions import TypeVarTuple, Unpack

P = ParamSpec("P")
T = TypeVar("T")
Ts = TypeVarTuple("Ts")

def pipeline(
    f1: Callable[P, T],
    *funcs: Unpack[tuple[Callable[[T], Ts[0]], Callable[[Ts[0]], Ts[1]], ..., Callable[[Ts[-2]], Ts[-1]]]]
) -> Callable[P, Ts[-1]]:
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> Ts[-1]:
        res = f1(*args, **kwargs)
        for func in funcs:
            res = func(res)
        return res
    return wrapper

这个方案会让类型检查器严格验证每个函数的输入输出衔接关系，最终返回类型准确指向最后一个函数的输出类型。

注意事项 #

• 确保类型检查器（如mypy、pyright）支持相关特性：mypy需开启--enable-recursive-aliases（Python 3.11+）或使用最新版本；pyright对这些特性的支持更友好。
• 流水线中后续函数的输入必须与前一个函数的输出类型匹配，否则类型检查器会报错，这也能提前发现逻辑错误。

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