You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python子类方法允许逆变参数及类型提示兼容问题修复方案

问题:Pipeline-Action动态链式调用的类型提示兼容问题

我用以下代码实现了多动作的动态链式调用:

from abc import ABC, abstractmethod
from typing import Union, Any


class Pipeline(ABC):
    def __init__(self, *actions: "Action"):
        self.actions = actions

    def __add__(self, other: "Pipeline") -> "Pipeline":
        new_pipeline = Pipeline()
        new_pipeline.actions = self.actions + other.actions
        return new_pipeline

    def __getitem__(self, index: Union[int, slice]) -> "Pipeline":
        if isinstance(index, int):
            return self.actions[index]
        else:
            new_pipeline = Pipeline()
            new_pipeline.actions = self.actions[index]
            return new_pipeline

    def __len__(self) -> int:
        return len(self.actions)

    def run(self, pipeline_input: Any = None) -> Any:
        value = pipeline_input
        for action in self.actions:
            value = action.run(value)
        return value


class Action(Pipeline):
    def __init__(self):
        super().__init__(self)  # 注意是传入self而非空参数

    @abstractmethod
    def run(self, pipeline_input: Any = None) -> Any:
        return


class Add(Action):
    def __init__(self, value: int):
        super().__init__()
        self.value = value

    def run(self, pipeline_input: int) -> int:
        return pipeline_input + self.value


class Multiply(Action):
    def __init__(self, value: int):
        super().__init__()
        self.value = value

    def run(self, pipeline_input: int) -> int:
        return pipeline_input * self.value

运行方式如下:

action1 = Add(3)
action2 = Multiply(4)

action1.run(2) # 结果 5
action2.run(5) # 结果 20

pipeline = action1 + action2

pipeline.run(5) # 结果 40
pipeline[0].run(3) # 结果 6
pipeline[0:2].run(5) # 结果 40

但当前结构存在类型提示问题:Action子类的run方法要求特定输入类型,导致Pylance抛出如下警告:

Method "run" overrides class "Action" in an incompatible manner
Parameter 2 mismatch: base parameter has default argument value, override parameter does not Pylancereport IncompatibleMethodOverride

需求:保留原有流水线-动作结构的核心逻辑,解决该类型提示问题,同时允许子类方法使用逆变参数(动作链式调用前的输入输出类型检查在代码外完成,动作通过名称动态链式调用)。


解决方案

1. 使用泛型定义Action类型,支持逆变输入

通过typing模块的Generic和逆变TypeVar,可以让Action子类明确指定自身的输入输出类型,同时保持与父类的方法签名兼容。

2. 调整方法签名兼容性

父类Action的run方法带有默认参数,子类需要与父类签名对齐。结合外部已做类型检查的前提,我们可以做如下修改:

修改后的完整代码:

from abc import ABC, abstractmethod
from typing import Union, Any, Generic, TypeVar

# 定义逆变的输入类型变量,支持子类使用更具体的输入类型
InputType = TypeVar('InputType', contravariant=True)
OutputType = TypeVar('OutputType')


class Pipeline(ABC):
    def __init__(self, *actions: "Action[Any, Any]"):
        self.actions = actions

    def __add__(self, other: "Pipeline") -> "Pipeline":
        new_pipeline = Pipeline()
        new_pipeline.actions = self.actions + other.actions
        return new_pipeline

    def __getitem__(self, index: Union[int, slice]) -> "Pipeline":
        if isinstance(index, int):
            return self.actions[index]
        else:
            new_pipeline = Pipeline()
            new_pipeline.actions = self.actions[index]
            return new_pipeline

    def __len__(self) -> int:
        return len(self.actions)

    def run(self, pipeline_input: Any = None) -> Any:
        value = pipeline_input
        for action in self.actions:
            value = action.run(value)
        return value


class Action(Pipeline, Generic[InputType, OutputType]):
    def __init__(self):
        super().__init__(self)

    @abstractmethod
    def run(self, pipeline_input: InputType) -> OutputType:
        """移除父类的默认参数,子类明确指定输入类型"""
        raise NotImplementedError


class Add(Action[int, int]):
    def __init__(self, value: int):
        super().__init__()
        self.value = value

    def run(self, pipeline_input: int) -> int:
        return pipeline_input + self.value


class Multiply(Action[int, int]):
    def __init__(self, value: int):
        super().__init__()
        self.value = value

    def run(self, pipeline_input: int) -> int:
        return pipeline_input * self.value

修改说明:

  • 泛型与逆变:定义InputType为逆变类型变量,允许子类使用比父类更具体的输入类型,满足逆变参数的需求。
  • 方法签名对齐:移除Action父类run方法的= None默认参数,让子类的run方法签名与父类保持一致,直接解决Pylance的不兼容警告。
  • 类型明确化:子类通过Action[int, int]明确指定输入输出类型,既保留了准确的类型提示,又符合业务逻辑中特定类型输入的要求。
  • 核心逻辑保留:Pipeline的加法、索引、run等核心功能完全保留,原有调用方式不受任何影响。

额外兼容场景说明

如果需要兼容None作为初始输入的场景,可以将子类的run方法参数类型调整为Union[InputType, None],同时在方法内部做简单判断(结合外部类型检查的前提,这一步可简化):

class Add(Action[Union[int, None], int]):
    def __init__(self, value: int):
        super().__init__()
        self.value = value

    def run(self, pipeline_input: Union[int, None]) -> int:
        # 外部已做类型检查,此处仅为兼容类型提示
        if pipeline_input is None:
            pipeline_input = 0
        return pipeline_input + self.value

内容的提问来源于stack exchange,提问作者이준서

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.04 12:38:10