跳转至

请解释责任链模式的设计原理,并说明如何将其应用于 Agent 工具调用失败的场景,实现多级降级策略?

“责任链模式,核心就是一句:

‘把请求沿着一条链往后传,直到有人处理它,否则掉出链尾。’

在 Agent 调用工具失败时,用这个模式做多级降级特别自然——就像给每个操作准备了多道保险。”


🔗 模式原理,三句话讲清楚

  1. 一条链:多个处理器对象串成链表。

  2. 每个处理器要么处理,要么往后传:处理完就终止;处理不了就调用下一个。

  3. 调用者不关心谁最终处理:它只跟链头打交道。

查看内嵌表格

典型优点:单一职责、开闭原则(随时增减处理器),缺点:链太长会影响性能,且可能没人处理。


🧰 应用到 Agent 工具调用失败:多级降级链

假设 Agent 要执行一个“查询天气”的工具,我们设计三层降级:

image.png

每个降级都是一个处理器,它们链在一起:

from abc import ABC, abstractmethod

class ToolHandler(ABC):
    def __init__(self):
        self._next = None  # 后继节点

    def set_next(self, handler: "ToolHandler") -> "ToolHandler":
        self._next = handler
        return handler  # 方便链式调用

    @abstractmethod
    def handle(self, query: str) -> str:
        # 默认实现:自己处理不了就转交下一个
        if self._next:
            return self._next.handle(query)
        return "⚠️ 所有策略均失败"

具体处理器(降级策略):

class PrimaryAPIHandler(ToolHandler):
    def handle(self, query):
        try:
            result = call_primary_weather_api(query)
            return f"🌤️ 主接口:{result}"
        except Exception:
            return super().handle(query)  # 传给下一个

class BackupAPIHandler(ToolHandler):
    def handle(self, query):
        try:
            result = call_backup_api(query)
            return f"☁️ 备用接口:{result}"
        except Exception:
            return super().handle(query)

class LocalCacheHandler(ToolHandler):
    def handle(self, query):
        result = lookup_local_knowledge(query)
        if result:
            return f"📚 本地知识库:{result}"
        return super().handle(query)

组装链:

primary = PrimaryAPIHandler()
backup = BackupAPIHandler()
local = LocalCacheHandler()

primary.set_next(backup).set_next(local)

# Agent 使用链头
response = primary.handle("北京今天天气")

🧩 为什么责任链特别适合这个场景?

  • ✅ 降级逻辑独立:每个策略一个类,互不干扰,好测试。

  • ✅ 顺序灵活:只需调整 set_next 的顺序就能改变降级策略。

  • ✅ 扩展方便:将来想加“收费接口”、“人工兜底”,新增一个 Handler 插进去就行。

  • ✅ 职责清晰:Agent 只需调用 handle(query),完全不知道降级细节。

📌 收尾一句话

“责任链让降级从 if-else 的泥潭里拉出来,变成了一条可插拔的策略管道。Agent 工具调用想稳,就得用这种模式把容错层级隔离开。”

这个回答既有理论,又有落地代码,面试时讲出来会让人觉得你对架构模式的运用非常实际。