【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(prompt="a person walking").videos[0]

报错：

AttributeError： 'Cosmos2_5_PredictBasePipeline' object has no attribute 'check_text_safety'

或者只在特定分支触发（如当 prompt 命中某个长度阈值、或开启了enable_safety=True时）：

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 接进来。于是基类调用了一个只有「别的分支」才有的方法。

三、根因

根因两点：

1. 基类调用了未定义的方法：__call__里的self.check_text_safety(prompt)依赖一个 mixin/子类方法，但PredictBase没有接入它。
2. 缺少「安全方法存在性」的契约校验：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(frozen=True) 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(f"pipeline 缺少安全方法: {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 True

CI 把test_safety_method_exists与test_call_does_not_attribute_error作为 Cosmos pipeline 的必过项，要求「任何调用check_text_safety的 pipeline 必须确认该方法存在（自带默认或接 mixin）」。

八、排查清单

Cosmos pipelineAttributeError: check_text_safety按顺序查：

1. hasattr(pipe, "check_text_safety")？没有就是基类调了未接的方法，__call__里self.check_text_safety(prompt)必崩。
2. 该类是否继承了带check_text_safety的 mixin？没继承就补默认实现或接入 mixin。
3. __call__调用check_text_safety前是否有getattr兜底？没有则任何子类漏接都会AttributeError。
4. 是否只在开启 safety 时触发？若enable_safety才调，检查该分支的方法是否真的被接进来。
5. 默认 passthrough 是否合理？无安全模型时返回(False, "", None)放过，有真实模型时覆盖。
6. 截断 prompt 时是否返回空输出？return_empty_on_flag=True时拦截应给空结果而非崩溃。

九、小结

「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。