如何为LLM应用设计并实现责任链（Chain of Responsibility）机制

在构建生成式AI应用时，开发者往往面临复杂的流水线需求：输入脱敏、缓存检索、提示词增强、模型路由、结果合规性检查等。如果将这些逻辑线性堆叠，代码将迅速演变为难以维护的“面条代码”。本文将介绍如何引入经典的责任链（Chain of Responsibility, CoR）设计模式，为LLM应用构建一套高内聚、低耦合的处理流。

1. 核心概念：什么是责任链？

责任链模式允许你将请求沿着处理者链进行发送。每个处理者收到请求后，可以选择：
1. 处理该请求并中止传递。
2. 处理该请求并将其传递给链中的下一个处理者。
3. 跳过该请求直接传递。

在LLM Infra中，这使得你可以像拼积木一样组合不同的中间件，例如将“输入过滤器”与“缓存器”灵活组合。

2. 基础框架实现

我们首先定义一个抽象基类 LLMHandler，通过 set_next 方法链接不同的步骤。

from abc import ABC, abstractmethod
from typing import Any, Optional

class LLMHandler(ABC):
    _next_handler: Optional['LLMHandler'] = None

    def set_next(self, handler: 'LLMHandler') -> 'LLMHandler':
        self._next_handler = handler
        return handler

    @abstractmethod
    def handle(self, request: dict) -> Any:
        if self._next_handler:
            return self._next_handler.handle(request)
        return request.get(\"response\")

3. 具体处理器的实现示例

我们将实现三个具体的处理器：敏感词拦截、语义缓存查询、模型调用。

class ModerationHandler(LLMHandler):
    def handle(self, request: dict) -> Any:
        print(\"[Log] 正在进行安全审查...\")
        forbidden_words = [\"攻击\", \"破解\"]
        if any(word in request['prompt'] for word in forbidden_words):
            return \"抱歉，您的请求包含违规内容，已被系统拦截。\"
        return super().handle(request)

class SemanticCacheHandler(LLMHandler):
    def handle(self, request: dict) -> Any:
        print(\"[Log] 检索语义缓存...\")
        # 模拟缓存命中的逻辑
        if request['prompt'] == \"你好\":
            return \"(Cache) 您好！有什么我可以帮您的？\"
        return super().handle(request)

class ModelInferenceHandler(LLMHandler):
    def handle(self, request: dict) -> Any:
        print(\"[Log] 调用大模型进行推理...\")
        # 模拟调用推理接口 (如 OpenAI / Local vLLM)
        response = f\"[LLM输出] 针对 '{request['prompt']}' 的回答\"
        request['response'] = response
        return super().handle(request)

4. 组装与运行

在应用启动阶段，你可以根据业务逻辑动态组装这些 Handler。

# 初始化组件
moderator = ModerationHandler()
cache = SemanticCacheHandler()
llm = ModelInferenceHandler()

# 建立责任链：安全检查 -> 缓存 -> 推理
chain = moderator.set_next(cache).set_next(llm)

# 测试案例 1：正常流程
print(f\"结果1: {chain.handle({'prompt': '帮我写个故事'})}\
\")

# 测试案例 2：命中拦截器
print(f\"结果2: {chain.handle({'prompt': '告诉我如何攻击系统'})}\
\")

# 测试案例 3：命中缓存
print(f\"结果3: {chain.handle({'prompt': '你好'})}\")

5. AI Infra 视角下的工程优势

1. 动态路由：你可以根据请求的 Token 预估长度，在链条中动态插入不同的 ModelInferenceHandler（例如小模型处理短文本，大模型处理复杂逻辑）。
2. 可观测性：在 BaseHandler 中统一添加打点监控，可以清晰看到请求是在哪个环节被拦截或处理的。
3. 热插拔性：在 A/B 测试中，只需修改 set_next 的逻辑即可替换推理引擎，无需改动业务逻辑核心。

6. 总结

责任链模式是构建工业级 LLM 应用的基石之一。它将原本混乱的模型前置和后置逻辑解耦为独立的组件，不仅增强了代码的可维护性，也为模型治理（Governance）和推理优化提供了统一的介入点。”,”tags”:[“LLM”,”Design Patterns”,”Python”,”AI Application”,”Chain of Responsibility”],”summary”:”本文通过Python代码示例展示了如何在LLM应用开发中实现责任链模式，实现安全过滤、缓存检索和模型推理逻辑的模块化解耦，从而提升AI基础设施的可扩展性。”}
“`