如何让Python静态识别reduce实现的管道的返回结果类型？

基于reduce函数实现管道类时，重构代码支持类型追踪发现reduce丢失类型信息，导致静态类型检查工具（如Pylance）识别的类型与实际运行结果不符。示例代码如下：

from functools import reduce
from typing import Callable, TypeVar, TypeAlias

def int_to_str(i: int) -> str:
    return str(i)

def str_to_float(s: str) -> float:
    return float(s)

pipes = [int_to_str, str_to_float]

_S = TypeVar("S")
_T = TypeVar("T")
Pipe : TypeAlias = Callable[[_S], _T]

def reducer(prev: _S, pipe: Pipe[_S, _T]) -> _T:
    return pipe(prev)

initial_arg = 100
# Pylance显示what_type的类型为int，与实际运行结果的float不符
what_type = reduce(reducer, pipes, initial_arg)
# what_type=100.0（浮点值）
print(f"{what_type=}")
# type(what_type)=<class 'float'>（浮点类型）
print(f"{type(what_type)=}")

环境：Python 3.10

方法1：显式标注管道序列的精确类型 #

类型检查器无法自动推断列表中函数的链式类型关系，需将管道定义为元组（元组元素类型固定，可被类型检查器逐个追踪）并显式标注类型：

from typing import Tuple, Callable

# 显式标注管道的精确链式类型：int→str→float
pipes: Tuple[Callable[[int], str], Callable[[str], float]] = (int_to_str, str_to_float)

改用元组是因为列表默认被视为元素类型为Callable[[int], str] | Callable[[str], float]的可变序列，无法追踪链式调用的类型流转。

方法2：自定义带类型追踪的管道函数替代reduce #

手动实现支持泛型链式类型推断的管道工具，避免依赖reduce的默认类型注解：

from typing import TypeVar, Callable

_S = TypeVar("_S")
_T = TypeVar("_T")

def pipe(value: _S) -> "PipeBuilder[_S]":
    class PipeBuilder:
        def __init__(self, current_value: _S):
            self.current_value = current_value
        
        def then(self, func: Callable[[_S], _T]) -> "PipeBuilder[_T]":
            return PipeBuilder(func(self.current_value))
        
        def result(self) -> _S:
            return self.current_value
    return PipeBuilder(value)

# 使用示例
what_type = pipe(100).then(int_to_str).then(str_to_float).result()

该实现通过泛型TypeVar逐步追踪每一步的类型转换，Pylance可准确识别what_type的类型为float。

方法3：为reduce添加重载类型注解 #

通过重载注解覆盖reduce的默认类型提示，让类型检查器能根据管道序列的长度和函数类型推断最终返回值：

from functools import reduce
from typing import Callable, TypeVar, TypeAlias, overload

_S = TypeVar("_S")
_T = TypeVar("_T")
_U = TypeVar("_U")
Pipe : TypeAlias = Callable[[_S], _T]

# 为两步、三步链式调用添加重载注解
@overload
def reduce_typed(func: Callable[[_S, Callable[[_S], _T]], _T], iterable: tuple[Callable[[_S], _T]], initializer: _S) -> _T: ...
@overload
def reduce_typed(func: Callable[[_T, Callable[[_T], _U]], _U], iterable: tuple[Callable[[_S], _T], Callable[[_T], _U]], initializer: _S) -> _U: ...

def reduce_typed(func, iterable, initializer):
    return reduce(func, iterable, initializer)

# 使用带重载的reduce_typed
pipes = (int_to_str, str_to_float)
what_type = reduce_typed(reducer, pipes, initial_arg)

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