Chain设计模式：Chain的概念、单链与多链、链式调用的实现、链的异常处理

大家好，我是你们的老朋友。今天我们来聊聊Chain设计模式。说实话，这个模式在我刚接触大模型工作流时，觉得它就是个简单的“流水线”。但用久了才发现，它其实是整个自动化流程的骨架。你想想看，没有Chain，你的大模型调用就是一堆散沙，根本串不起来。

Chain的概念：说白了就是“接力棒”

Chain设计模式，核心思想就是把多个处理步骤串起来。前一个步骤的输出，直接作为后一个步骤的输入。就像接力赛跑，一棒接一棒。

我个人习惯把Chain比作“管道”。数据从一头进去，经过一道道工序，最后从另一头出来。每个工序只做一件事，但组合起来就能完成复杂任务。

核心要点：Chain的本质是“组合”而非“继承”。我们不是把功能写死在一个大函数里，而是把独立的功能模块通过Chain串联起来。

举个例子。你要做一个智能客服。流程可能是：用户输入 → 意图识别 → 实体抽取 → 知识库检索 → 答案生成。这五个步骤，每个都可以是一个独立的Chain节点。

单链与多链：一条路走到黑，还是多条路并行？

单链，就是一条直线。A → B → C → D。简单直接，适合线性流程。我在项目中遇到过很多场景，比如数据清洗、格式转换，用单链就足够了。

多链呢？说白了就是分支和并行。比如：

  • 条件分支：根据中间结果，走不同的子链。比如意图识别为“投诉”走投诉处理链，识别为“咨询”走问答链。
  • 并行链：多个链同时跑，最后合并结果。比如同时调用三个不同的模型做情感分析，然后投票决定最终结果。
类型 适用场景 复杂度 我踩过的坑
单链 线性处理、数据流水线 低 容易忽略中间结果的校验
条件分支链 决策型流程、路由分发 中 分支条件写死，后期改起来很痛苦
并行链 多模型集成、并行计算 高 并发控制没做好，资源打满

我的建议：刚开始做设计时，先用单链把流程跑通。别一上来就搞多链，容易把自己绕进去。等单链稳定了，再考虑哪些地方需要分支或并行。

链式调用的实现：代码里怎么玩？

嗯，这里直接上代码。我习惯用Python来实现Chain，因为它的装饰器和上下文管理器太好用了。

class ChainNode:
    def __init__(self, func):
        self.func = func
        self.next_node = None
    
    def set_next(self, node):
        self.next_node = node
        return node
    
    def execute(self, input_data):
        result = self.func(input_data)
        if self.next_node:
            return self.next_node.execute(result)
        return result

# 定义几个处理函数
def step_a(data):
    print(f"Step A: {data}")
    return data + " -> A"

def step_b(data):
    print(f"Step B: {data}")
    return data + " -> B"

def step_c(data):
    print(f"Step C: {data}")
    return data + " -> C"

# 构建单链
node_a = ChainNode(step_a)
node_b = ChainNode(step_b)
node_c = ChainNode(step_c)

node_a.set_next(node_b).set_next(node_c)

# 执行
result = node_a.execute("Start")
print(f"Final: {result}")

这段代码看着简单，但它是所有复杂Chain的基础。每个节点只关心自己的逻辑，然后通过set_next串起来。我曾经在一个项目中，用这个模式串了20多个节点，从数据采集到最终报告生成，一条链搞定。

对于多链，我通常会加一个Router节点：

class RouterNode(ChainNode):
    def __init__(self, routes):
        # routes: dict, key是条件, value是对应的ChainNode
        self.routes = routes
        super().__init__(self._route)
    
    def _route(self, data):
        condition = data.get("intent")  # 假设根据intent路由
        if condition in self.routes:
            return self.routes[condition].execute(data)
        else:
            raise ValueError(f"Unknown route: {condition}")

注意：路由条件不要硬编码。我建议把路由规则做成可配置的，比如从YAML文件读取。否则每次改分支逻辑都要改代码，太容易出错了。

链的异常处理：别让一根链条断了整条线

这是最容易被忽视的部分。很多人写Chain时，只关注正常流程。但实际生产中，异常才是常态。

我曾经在一个自动化报告系统中，因为某个节点的API超时，导致整条链卡死，所有后续任务都堵住了。那次事故让我养成了一个习惯：每个节点都要有异常处理。

我常用的策略有三种：

  1. 重试机制：对于网络抖动、临时错误，重试2-3次。用指数退避，别一失败就重试，会把服务打崩。
  2. 降级处理：如果某个节点挂了，能不能用默认值代替？比如知识库检索失败，直接返回“抱歉，我暂时无法回答”。
  3. 熔断机制：如果某个节点连续失败超过阈值，直接跳过它，不再调用。等一段时间后再尝试恢复。
class SafeChainNode(ChainNode):
    def __init__(self, func, max_retries=3, fallback=None):
        super().__init__(func)
        self.max_retries = max_retries
        self.fallback = fallback
        self.failure_count = 0
    
    def execute(self, input_data):
        for attempt in range(self.max_retries):
            try:
                result = self.func(input_data)
                self.failure_count = 0  # 成功则重置
                if self.next_node:
                    return self.next_node.execute(result)
                return result
            except Exception as e:
                print(f"Attempt {attempt+1} failed: {e}")
                if attempt == self.max_retries - 1:
                    # 最后一次失败，使用降级
                    if self.fallback:
                        print("Using fallback...")
                        result = self.fallback(input_data)
                        if self.next_node:
                            return self.next_node.execute(result)
                        return result
                    else:
                        raise  # 没有降级，直接抛出

避坑指南：我曾经犯过一个错误——在异常处理中吞掉了错误信息。结果链跑完了，但结果全是错的，排查了三天才发现是某个节点静默失败了。所以，异常一定要记录日志，哪怕你用了降级。

另外，对于并行链的异常处理，要特别注意“部分失败”的情况。比如三个并行任务，两个成功一个失败。你是整体回滚，还是只处理失败的那个？我个人建议：能局部处理就局部处理，别因为一个节点失败就把整个流程回滚，太浪费了。

好了，关于Chain设计模式，今天就聊这么多。记住：Chain的核心是组合，异常处理是灵魂。下次我们聊聊更高级的编排模式。

← 上一章 📖 返回目录 下一章 →