设计模式(02)-抽象工厂：产品族一致性的工程意义

抽象工厂（Abstract Factory）解决的问题和工厂方法不同。

工厂方法管的是”怎么创建某一类对象”；抽象工厂管的是”如何保证一组相关对象来自同一个家族”。两者是层次性的差别，不是程度性的差别。

---

核心约束：产品族一致性

假设你在构建一个 RAG 系统，需要三个组件：LLM 客户端、Embedding 模型、Tokenizer。

来自不同厂商的组件混用会出问题：

• OpenAI 的 embedding 用 tiktoken 计算 token，Cohere 用自己的 tokenizer，计数方式不兼容
• 不同厂商的 embedding 维度和归一化策略不同，混用会让向量检索返回无意义的相似度分数
• 速率限制、重试策略、错误码各家不同，混合调用链里的异常处理会失控

抽象工厂的价值就在这里：通过类型系统强制这三个组件来自同一个产品族，而不是让调用方在运行时撞上不兼容问题。

---

实现

from abc import ABC, abstractmethod
from typing import Protocol


# 抽象产品接口
class LLMClient(Protocol):
    def complete(self, prompt: str) -> str: ...

class Embedder(Protocol):
    def embed(self, text: str) -> list[float]: ...

class Tokenizer(Protocol):
    def count_tokens(self, text: str) -> int: ...


# 抽象工厂
class LLMProviderFactory(ABC):
    @abstractmethod
    def create_client(self) -> LLMClient: ...

    @abstractmethod
    def create_embedder(self) -> Embedder: ...

    @abstractmethod
    def create_tokenizer(self) -> Tokenizer: ...


# 具体工厂：OpenAI 产品族
class OpenAIFactory(LLMProviderFactory):
    def __init__(self, api_key: str):
        self._api_key = api_key

    def create_client(self) -> LLMClient:
        return OpenAIClient(self._api_key)

    def create_embedder(self) -> Embedder:
        return OpenAIEmbedder(self._api_key)

    def create_tokenizer(self) -> Tokenizer:
        return TiktokenTokenizer()


# 具体工厂：Anthropic 产品族
class AnthropicFactory(LLMProviderFactory):
    def __init__(self, api_key: str):
        self._api_key = api_key

    def create_client(self) -> LLMClient:
        return AnthropicClient(self._api_key)

    def create_embedder(self) -> Embedder:
        return VoyageEmbedder(self._api_key)  # Anthropic 生态推荐 Voyage AI

    def create_tokenizer(self) -> Tokenizer:
        return AnthropicTokenizer()

调用方只接触抽象工厂接口，不感知具体实现：

class RAGPipeline:
    def __init__(self, factory: LLMProviderFactory):
        self.client = factory.create_client()
        self.embedder = factory.create_embedder()
        self.tokenizer = factory.create_tokenizer()

    def query(self, question: str, context: str) -> str:
        token_count = self.tokenizer.count_tokens(context)
        if token_count > 4096:
            context = context[:4096]
        return self.client.complete(f"{context}\n\nQ: {question}")

切换整个产品族只需换一行：

# 开发环境
pipeline = RAGPipeline(OpenAIFactory(api_key=os.getenv("OPENAI_KEY")))

# 生产环境
pipeline = RAGPipeline(AnthropicFactory(api_key=os.getenv("ANTHROPIC_KEY")))

---

与工厂方法的边界

	工厂方法	抽象工厂
目标	创建单一产品，延迟到子类决定	创建相互依赖的产品族
核心问题	如何创建 X	如何保证 X/Y/Z 来自同一族
复杂度	低	高
横向扩展代价	新产品类型：加子类	新产品族：加工厂；新产品维度：改所有工厂

抽象工厂最难受的地方是横向扩展：产品族里需要新增一种产品（比如加 Reranker），就要改抽象工厂接口，所有具体工厂都得跟着改。产品维度稳定时这个代价可以接受；接口还在频繁迭代时，抽象工厂会制造大量改动。

---

用还是不用

适合：系统需要支持多个”后端”或”提供商”，每个后端包含一组相互配合的组件；需要在类型检查阶段而非运行期排除产品族混用。

不适合：只有一个产品维度在变化，工厂方法或注册表模式就够了；产品族边界还没稳定，接口仍在频繁迭代——过早引入抽象工厂会让重构成本倍增。

判断标准只有一个：你约束的是”一组东西必须来自同一体系”，还是”某一类东西怎么创建”。前者是抽象工厂的场景，后者不是。