LangChain入门-设计模式完全解析

深入分析 LangChain 框架中运用的 10 种设计模式，包括策略模式、模板方法、装饰器模式、责任链模式、工厂模式等

LangChain 设计模式分析

---

概述

LangChain 框架巧妙地运用了多种设计模式来实现其核心功能。本文档详细分析这些模式的应用，帮助开发者理解框架的设计思想。

1. 策略模式 (Strategy Pattern)

定义

定义算法族，分别封装，让它们可以互相替换。策略模式让算法的变化独立于使用算法的客户。

在 LangChain 中的应用

1.1 语言模型策略

不同的 LLM 提供商可以通过相同的接口使用：

# 文件位置: libs/core/langchain_core/language_models/base.py

class BaseLanguageModel(RunnableSerializable[LanguageModelInput, LanguageModelOutputVar], ABC):
    """所有语言模型的抽象基类"""

    @abstractmethod
    def generate_prompt(
        self,
        prompts: list[PromptValue],
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> LLMResult:
        """生成文本的抽象方法，由子类实现"""

使用示例:

# 可以轻松切换不同的模型提供商
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# 使用 OpenAI
llm = ChatOpenAI(model="gpt-4")

# 切换到 Anthropic，只需更改实例化代码
llm = ChatAnthropic(model="claude-3-sonnet")

# 其余代码无需修改
chain = prompt | llm | parser

1.2 检索器策略

不同的检索方式可以互换：

# 向量存储检索
from langchain_community.vectorstores import FAISS
retriever = FAISS.from_documents(docs, embeddings).as_retriever()

# 切换到其他向量存储
from langchain_community.vectorstores import Chroma
retriever = Chroma.from_documents(docs, embeddings).as_retriever()

# 或使用 BM25 检索
from langchain_community.retrievers import BM25Retriever
retriever = BM25Retriever.from_documents(docs)

类图

2. 模板方法模式 (Template Method Pattern)

定义

定义算法骨架，将一些步骤延迟到子类实现。模板方法使得子类可以不改变算法结构即可重定义算法的某些步骤。

在 LangChain 中的应用

2.1 Runnable 执行模板

所有 Runnable 组件遵循相同的执行模式：

# 文件位置: libs/core/langchain_core/runnables/base.py

class Runnable(Generic[Input, Output], ABC):
    """定义执行模板的抽象基类"""

    # 同步执行的模板方法
    def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
        """模板方法：定义执行流程"""
        # 1. 合并配置
        config = merge_configs(self.config, config)

        # 2. 获取回调管理器
        callback_manager = get_callback_manager_for_config(config)

        # 3. 调用子类实现的实际逻辑
        return self._invoke_with_config(input, config, callback_manager)

    @abstractmethod
    def _invoke(
        self, input: Input, config: RunnableConfig, **kwargs: Any
    ) -> Output:
        """子类实现具体的执行逻辑"""
        pass

执行流程

3. 装饰器模式 (Decorator Pattern)

定义

动态地给对象添加额外职责。装饰器模式提供了比继承更灵活的扩展方式。

在 LangChain 中的应用

3.1 Runnable 装饰器

with_types() - 类型装饰:

# 文件位置: libs/core/langchain_core/runnables/base.py

def with_types(
    self,
    input_type: Optional[Type[InputType]] = None,
    output_type: Optional[Type[OutputType]] = None,
) -> Runnable[Input, Output]:
    """为 Runnable 添加类型信息"""
    return RunnableBinding(
        bound=self,
        input_type=input_type,
        output_type=output_type,
    )

with_retry() - 重试装饰:

# 文件位置: libs/core/langchain_core/runnables/retry.py

class RunnableRetry(RunnableBindingBase[Input, Output]):
    """添加重试能力的装饰器"""

    def __init__(
        self,
        bound: Runnable[Input, Output],
        *,
        max_attempts: int = 3,
        wait_exponential_jitter: bool = True,
    ):
        self.bound = bound
        self.max_attempts = max_attempts
        self.wait_exponential_jitter = wait_exponential_jitter

    def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
        """带重试的执行"""
        last_exception = None
        for attempt in range(self.max_attempts):
            try:
                return self.bound.invoke(input, config)
            except Exception as e:
                last_exception = e
                if attempt < self.max_attempts - 1:
                    time.sleep(self._calculate_backoff(attempt))
        raise last_exception

with_fallbacks() - 降级装饰:

# 文件位置: libs/core/langchain_core/runnables/fallbacks.py

class RunnableWithFallbacks(RunnableSerializable[Input, Output]):
    """添加降级处理能力的装饰器"""

    def __init__(
        self,
        runnable: Runnable[Input, Output],
        fallbacks: Sequence[Runnable[Input, Output]],
    ):
        self.runnable = runnable
        self.fallbacks = fallbacks

    def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
        """尝试主流程，失败时使用降级方案"""
        try:
            return self.runnable.invoke(input, config)
        except Exception as primary_error:
            for fallback in self.fallbacks:
                try:
                    return fallback.invoke(input, config)
                except Exception:
                    continue
            raise primary_error

装饰器链

4. 责任链模式 (Chain of Responsibility Pattern)

定义

为请求创建一个接收者对象链。每个接收者都包含对下一个接收者的引用。

在 LangChain 中的应用

4.1 RunnableSequence（LCEL 管道）

# 文件位置: libs/core/langchain_core/runnables/base.py

class RunnableSequence(RunnableSerializable[Input, Output]):
    """顺序执行的链"""

    steps: Sequence[Runnable[langchain_core.utils.input.Input, Output]]

    def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
        """依次执行每个步骤"""
        value: Any = input

        # 遍历所有步骤，将输出传递给下一个
        for step in self.steps:
            value = step.invoke(value, config)

        return value

使用 LCEL 的管道操作符:

# | 操作符重载实现责任链
def __or__(self, other: Runnable[Any, Other]) -> RunnableSequence:
    """实现 | 操作符，创建链"""
    return RunnableSequence([self, other])

执行流程

5. 工厂模式 (Factory Pattern)

定义

定义创建对象的接口，让子类决定实例化哪个类。工厂方法使一个类的实例化延迟到其子类。

在 LangChain 中的应用

5.1 动态组件加载

# 文件位置: libs/langchain/langchain/load/serializable.py

class Serializable(BaseModel):
    """支持序列化的基类"""

    @classmethod
    def is_lc_serializable(cls) -> bool:
        """是否可序列化"""
        return True

    @classmethod
    def get_lc_namespace(cls) -> list[str]:
        """获取命名空间，用于动态加载"""
        return [cls.__module__]

5.2 提示词模板工厂

# 文件位置: libs/core/langchain_core/prompts/chat.py

class ChatPromptTemplate(BasePromptTemplate):
    """聊天提示词模板"""

    @classmethod
    def from_messages(cls, messages: Sequence[MessageLikeRepresentation]) -> ChatPromptTemplate:
        """从消息列表创建模板"""
        return cls(messages=list(messages))

    @classmethod
    def from_template(cls, template: str, **kwargs: Any) -> ChatPromptTemplate:
        """从字符串模板创建"""
        return cls.from_messages([
            ("human", template)
        ])

工厂方法

6. 观察者模式 (Observer Pattern)

定义

定义对象间的一种一对多依赖关系，当一个对象状态发生改变时，所有依赖它的对象都得到通知。

在 LangChain 中的应用

6.1 回调系统

# 文件位置: libs/core/langchain_core/callbacks/base.py

class BaseCallbackHandler(ABC):
    """回调处理器基类"""

    @abstractmethod
    def on_llm_start(self, prompts: list[str], **kwargs: Any) -> Any:
        """LLM 开始时调用"""
        pass

    @abstractmethod
    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
        """LLM 结束时调用"""
        pass

    @abstractmethod
    def on_chain_start(self, serialized: dict, inputs: dict, **kwargs: Any) -> Any:
        """链开始时调用"""
        pass

    @abstractmethod
    def on_chain_end(self, outputs: dict, **kwargs: Any) -> Any:
        """链结束时调用"""
        pass

6.2 回调管理器

# 文件位置: libs/core/langchain_core/callbacks/manager.py

class CallbackManager:
    """管理多个回调处理器"""

    def __init__(self, handlers: list[BaseCallbackHandler]):
        self.handlers = handlers

    def on_llm_start(self, prompts: list[str], **kwargs: Any) -> None:
        """通知所有观察者"""
        for handler in self.handlers:
            handler.on_llm_start(prompts, **kwargs)

    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
        """通知所有观察者"""
        for handler in self.handlers:
            handler.on_llm_end(response, **kwargs)

观察者模式流程

7. 适配器模式 (Adapter Pattern)

定义

将一个类的接口转换成客户期望的另一个接口。适配器让原本接口不兼容的类可以合作。

在 LangChain 中的应用

7.1 RunnableLambda 函数适配器

# 文件位置: libs/core/langchain_core/runnables/base.py

class RunnableLambda(Runnable[Input, Output]):
    """将普通函数适配为 Runnable"""

    func: Union[
        Callable[[Input], Output],
        Callable[[Input], Awaitable[Output]],
    ]

    def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
        """调用被适配的函数"""
        return self.func(input)

    def astream(self, input: Input, config: Optional[RunnableConfig] = None) -> AsyncIterator:
        """异步流式适配"""
        # 将同步函数适配为异步流
        result = self.func(input)
        yield result

使用示例:

# 将普通函数适配为 Runnable
def uppercase(text: str) -> str:
    return text.upper()

uppercase_runnable = RunnableLambda(uppercase)

# 现在可以在链中使用
chain = uppercase_runnable | another_runnable

7.2 文档加载器适配

# 适配不同的数据源
from langchain_community.document_loaders import (
    TextLoader,      # 文本文件
    PDFLoader,       # PDF 文件
    WebLoader,       # 网页
    JSONLoader,      # JSON 数据
)

# 都返回相同的 Document 类型
loader = TextLoader("file.txt")
docs = loader.load()  # List[Document]

8. 组合模式 (Composite Pattern)

定义

将对象组合成树形结构以表示"部分-整体"的层次结构。使用户可以一致地使用单个对象和组合对象。

在 LangChain 中的应用

8.1 Runnable 树形组合

# 复杂的链可以无限嵌套
chain = (
    prompt                          # Runnable
    | llm                          # Runnable
    | {
        "summary": summary_chain,   # RunnableParallel (组合)
        "analysis": analysis_chain, # 子 Runnable
    }
    | final_parser                 # Runnable
)

组合结构

9. 迭代器模式 (Iterator Pattern)

定义

提供一种方法顺序访问一个聚合对象中的各个元素，而不暴露其内部表示。

在 LangChain 中的应用

9.1 流式输出

# 文件位置: libs/core/langchain_core/runnables/base.py

def stream(
    self,
    input: Input,
    config: Optional[RunnableConfig] = None,
    **kwargs: Optional[Any],
) -> Iterator[Output]:
    """流式输出结果"""
    # 默认实现：调用 invoke 然后迭代
    yield self.invoke(input, config)

def astream(
    self,
    input: Input,
    config: Optional[RunnableConfig] = None,
    **kwargs: Optional[Any],
) -> AsyncIterator[Output]:
    """异步流式输出"""
    # 默认实现：调用 ainvoke 然后迭代
    result = await self.ainvoke(input, config)
    yield result

使用示例:

# 流式处理大型文档
for chunk in retriever.stream(query):
    process_chunk(chunk)

10. 命令模式 (Command Pattern)

定义

将请求封装为对象，从而可以用不同的请求对客户进行参数化。

在 LangChain 中的应用

10.1 Tool 命令

# 文件位置: libs/core/langchain_core/tools/base.py

class BaseTool(RunnableSerializable[Union[str, dict], Any]):
    """工具命令的封装"""

    name: str
    description: str
    args_schema: Type[BaseModel] = Field(default=...)

    def _run(
        self,
        *args: Any,
        **kwargs: Any,
    ) -> Any:
        """执行命令的具体逻辑"""
        pass

    def invoke(self, input: Union[str, dict], config: Optional[RunnableConfig] = None) -> Any:
        """执行命令"""
        parsed_input = self._parse_input(input)
        return self._run(**parsed_input)

命令执行流程

总结

设计模式	应用场景	关键类
策略模式	可互换的算法	BaseLanguageModel, BaseRetriever
模板方法	定义执行骨架	Runnable.invoke()
装饰器模式	动态添加功能	with_retry(), with_fallbacks()
责任链	顺序处理	RunnableSequence, `
工厂模式	创建对象	from_template(), from_messages()
观察者	事件通知	BaseCallbackHandler
适配器模式	接口转换	RunnableLambda
组合模式	树形结构	嵌套的链
迭代器模式	流式访问	stream(), astream()
命令模式	请求封装	BaseTool

这些设计模式的组合使用，使 LangChain 成为一个灵活、可扩展、易于使用的 AI 应用开发框架。

---