【Bug已解决】AttributeError Cosmos2_5_PredictBasePipeline object has no attribute check_text_safety 解决方案一、现象长什么样Cosmos2_5_PredictBasePipeline是 NVIDIA Cosmos 系列的视频预测 pipeline。用户调用它生成视频时pipeline 内部做文本安全检查NSFW 过滤那一步直接崩from diffusers import Cosmos2_5_PredictBasePipeline pipe Cosmos2_5_PredictBasePipeline.from_pretrained(nvidia/Cosmos2_5-Predict) video pipe(prompta person walking).videos[0]报错AttributeError Cosmos2_5_PredictBasePipeline object has no attribute check_text_safety或者只在特定分支触发如当 prompt 命中某个长度阈值、或开启了enable_safetyTrue时File .../pipelines/cosmos/cosmos_predict.py, line 88, in __call__ flagged self.check_text_safety(prompt) AttributeError Cosmos2_5_PredictBasePipeline object has no attribute check_text_safety现象总结Cosmos2_5_PredictBasePipeline 的__call__里调用了self.check_text_safety(prompt)但这个方法根本没有定义在PredictBase这个基类的 pipeline 上它可能只存在于某个子类或另一个 mixin 里于是基类的实例一走到安全检查就AttributeError。二、背景很多 diffusers pipeline 会把安全过滤文本/图像 NSFW 检查做成可插拔的方法。常见模式基类 pipeline 在__call__里调self.check_text_safety(prompt)真正的安全逻辑由某个 mixin如TextualNeuronSafetyMixin或子类提供如果某个 pipeline 类只继承了基类、却没继承带check_text_safety的 mixin调用就会AttributeError。Cosmos2_5_PredictBasePipeline显然属于这种情况它的__call__假设check_text_safety存在可能从别的 Cosmos pipeline 复制了调用代码但PredictBase这一支没有把提供该方法的 mixin 接进来。于是基类调用了一个只有「别的分支」才有的方法。三、根因根因两点基类调用了未定义的方法__call__里的self.check_text_safety(prompt)依赖一个 mixin/子类方法但PredictBase没有接入它。缺少「安全方法存在性」的契约校验pipeline 构建/__call__入口没有检查hasattr(self, check_text_safety)于是错误推迟到运行时才暴露且信息晦涩。本质「调用安全检查」与「提供安全检查实现」两件事在类继承上脱节PredictBase调了它没接的方法。四、最小可运行复现用标准库复现「基类调了 mixin 才有、自己没接的方法」class SafetyMixin: def check_text_safety(self, prompt): return False # 未标记危险 class PredictBase: def __call__(self, prompt): # 假设 check_text_safety 一定存在从别的类复制来的调用 return self.check_text_safety(prompt) # PredictBase 没这个方法 # 错误PredictBase 没继承 SafetyMixin class CosmosPredict(PredictBase): pass p CosmosPredict() try: p(a person walking) except AttributeError as e: print(AttributeError, e) # CosmosPredict has no attribute check_text_safety复现「正确」让CosmosPredict(PredictBase, SafetyMixin)调用即通过。五、解决方案第一层最小直接修复最小修复给PredictBase提供check_text_safety的默认实现或接入带该方法的 mixin并在调用前用getattr兜底import warnings class Cosmos2_5_PredictBasePipeline(DiffusionPipeline): # 默认实现不接外部安全模型时提供 passthrough不拦截 def check_text_safety(self, prompt, **kwargs): # 默认不做深度安全检查仅返回「未标记危险」 空警告 return False, , None torch.no_grad() def __call__(self, prompt, **kwargs): # 调用前用 getattr 兜底避免 AttributeError checker getattr(self, check_text_safety, None) if checker is not None: flagged, *_ checker(prompt) if flagged: warnings.warn(prompt 触发文本安全检查已拦截) return self._empty_output() # 正常生成 ...这样即使某个子类没接安全 mixin基类也有默认check_text_safety不会再AttributeError若子类接了真正的实现默认会被覆盖。六、解决方案第二层结构性改进把「安全检查方法的存在契约」收敛成一个 dataclass 单一真源并在 pipeline 构建/调用时校验from dataclasses import dataclass, field from typing import List, Optional dataclass(frozenTrue) class CosmosTextSafetyPolicy: Cosmos pipeline 文本安全契约的单一真源。 # 必须存在的安全方法名 required_safety_methods: tuple (check_text_safety,) # 是否允许默认 passthrough 实现True无安全模型时放过 allow_default_passthrough: bool True # 默认 passthrough 的返回约定 (flagged, warning, hidden) default_return: tuple (False, , None) # 触发拦截时是否返回空输出 return_empty_on_flag: bool True def ensure_method(self, pipeline_instance) - List[str]: problems [] for m in self.required_safety_methods: if not hasattr(pipeline_instance, m): if self.allow_default_passthrough: # 动态补一个默认实现 pipeline_instance.check_text_safety lambda prompt, **k: self.default_return else: problems.append(fpipeline 缺少安全方法: {m}) return problems def validate_call_site(self, pipeline_instance) - bool: return all(hasattr(pipeline_instance, m) for m in self.required_safety_methods)pipeline 在__init__结尾调policy.ensure_method(self)保证方法存在__call__入口调policy.validate_call_site(self)再做安全检查。七、解决方案第三层断言 / CI 守护用 pytest 把「安全方法存在 默认 passthrough 拦截行为」固化成回归import pytest from diffusers import Cosmos2_5_PredictBasePipeline from mylib.cosmos_safety import CosmosTextSafetyPolicy POLICY CosmosTextSafetyPolicy() def test_safety_method_exists(): pipe Cosmos2_5_PredictBasePipeline.from_pretrained(nvidia/Cosmos2_5-Predict) problems POLICY.ensure_method(pipe) assert problems [], 安全检查问题:\n \n.join(problems) assert hasattr(pipe, check_text_safety) def test_call_does_not_attribute_error(): pipe Cosmos2_5_PredictBasePipeline.from_pretrained(nvidia/Cosmos2_5-Predict) # 不应再 AttributeError out pipe(a person walking) assert out is not None def test_default_passthrough_not_flagged(): pipe Cosmos2_5_PredictBasePipeline.from_pretrained(nvidia/Cosmos2_5-Predict) flagged, warn, hidden pipe.check_text_safety(hello) assert flagged is False def test_real_safety_mixin_overrides_default(): class RealSafety(Cosmos2_5_PredictBasePipeline): def check_text_safety(self, prompt, **kw): return (True, blocked, None) # 真实实现覆盖默认 p RealSafety.from_pretrained(nvidia/Cosmos2_5-Predict) assert p.check_text_safety(x)[0] is TrueCI 把test_safety_method_exists与test_call_does_not_attribute_error作为 Cosmos pipeline 的必过项要求「任何调用check_text_safety的 pipeline 必须确认该方法存在自带默认或接 mixin」。八、排查清单Cosmos pipelineAttributeError: check_text_safety按顺序查hasattr(pipe, check_text_safety)没有就是基类调了未接的方法__call__里self.check_text_safety(prompt)必崩。该类是否继承了带check_text_safety的 mixin没继承就补默认实现或接入 mixin。__call__调用check_text_safety前是否有getattr兜底没有则任何子类漏接都会AttributeError。是否只在开启 safety 时触发若enable_safety才调检查该分支的方法是否真的被接进来。默认 passthrough 是否合理无安全模型时返回(False, , None)放过有真实模型时覆盖。截断 prompt 时是否返回空输出return_empty_on_flagTrue时拦截应给空结果而非崩溃。九、小结「AttributeError: Cosmos2_5_PredictBasePipeline object has no attribute check_text_safety」本质是基类 pipeline 的__call__调用了安全检查方法check_text_safety但PredictBase这一支没有接入提供该方法的 mixin导致基类调了它没接的方法而AttributeError。第一层给基类补默认check_text_safety并调用前getattr兜底第二层把安全方法的存在契约收敛到CosmosTextSafetyPolicy单一真源构建时ensure_method、调用时validate_call_site第三层用 pytest 守住「安全方法存在、调用不崩、默认 passthrough、真实实现可覆盖」。通用教训**任何基类里对「可能由 mixin/子类提供的方法」的调用都必须有默认实现或调用前存在性校验否则复制调用代码到没接 mixin 的分支就会运行时AttributeError。
