汤不热吧在复杂的LLM应用部署中（例如RAG系统、Agent框架或多轮对话引擎），我们需要确保用户输入经过一系列严格且顺序的流程处理：输入验证、安全过滤、上下文管理、路由选择，最后才到达核心推理模型。传统的线性代码结构难以应对这种流程的动态变化和扩展。责任链（Chain of Responsibility, CoR）设计模式是解决这一问题的完美方案。
CoR模式允许我们将请求沿着一系列处理对象传递。每个处理者决定是否处理请求，以及是否将其传递给链中的下一个处理者。这带来了极高的模块化和可插拔性。
1. 核心概念与优势
CoR的优势：
1. 解耦： 请求发送者和接收者完全解耦。
2. 灵活性： 我们可以运行时动态地调整处理器的顺序或添加新的处理器，无需修改现有代码。
3. 单一职责： 每个处理器只负责一个具体的任务（例如，只负责安全检查）。
2. Python实现：抽象处理器
我们首先定义一个抽象基类（Abstract Handler），它定义了处理请求的接口和设置链中下一个处理者的方法。
from abc import ABC, abstractmethod
class AbstractLLMHandler(ABC):
"""抽象处理器接口"""
def __init__(self):
self._next_handler = None
def set_next(self, handler):
"""设置链中的下一个处理者"""
self._next_handler = handler
return handler
@abstractmethod
def handle(self, request: dict) -> dict:
"""处理请求或将请求传递下去"""
pass
def _pass_to_next(self, request: dict) -> dict:
"""内部方法：将请求传递给链中的下一个节点"""
if self._next_handler:
return self._next_handler.handle(request)
return request
3. 实现具体的LLM处理流程
我们创建三个具体的处理节点：输入验证、安全检查和核心LLM推理。
3.1. 节点一：输入验证 (InputValidator)
确保输入不为空且长度合理。
class InputValidator(AbstractLLMHandler):
def handle(self, request: dict) -> dict:
prompt = request.get('prompt', '')
print(f"[Validator] 正在检查输入: {prompt[:30]}...")
if not prompt or len(prompt) < 10:
request['response'] = "错误：输入太短，请提供更详细的请求。"
request['status'] = 'STOPPED_VALIDATION_FAIL'
return request # 停止链条
return self._pass_to_next(request) # 传递给下一个
3.2. 节点二：安全检查 (SafetyChecker)
模拟内容审核或越狱尝试检测。
class SafetyChecker(AbstractLLMHandler):
def handle(self, request: dict) -> dict:
if request.get('status') == 'STOPPED_VALIDATION_FAIL':
return request # 如果上一步失败，则不再执行
prompt = request['prompt']
print("[Safety] 正在进行安全内容检查...")
# 模拟检测到违禁词
if "越狱" in prompt or "攻击" in prompt:
request['response'] = "警告：检测到不安全内容，请求被终止。"
request['status'] = 'STOPPED_SAFETY_FAIL'
return request # 停止链条
return self._pass_to_next(request)
3.3. 节点三：核心LLM处理 (CoreLLMProcessor)
只有通过所有前置检查的请求才能到达这里，模拟调用大型语言模型。
class CoreLLMProcessor(AbstractLLMHandler):
def handle(self, request: dict) -> dict:
if 'status' in request and request['status'].startswith('STOPPED'):
return request
prompt = request['prompt']
print("[LLM] 正在调用核心模型进行推理...")
# 模拟LLM响应
response_text = f"模型已根据您的请求 '{prompt[:20]}...' 生成了专业回复。"
request['response'] = response_text
request['status'] = 'COMPLETED'
return request
4. 构建并运行责任链
现在我们通过串联这些处理器来构建完整的推理管道。
# 1. 实例化处理器
validator = InputValidator()
safety_checker = SafetyChecker()
llm_processor = CoreLLMProcessor()
# 2. 设置责任链： A -> B -> C
validator.set_next(safety_checker).set_next(llm_processor)
# 3. 测试案例 1：成功通过整个链条
request_success = {
'prompt': '请解释大型语言模型中的Transformer架构如何工作，并给出代码示例。',
'user_id': 'user_001'
}
print("\n--- 测试案例 1: 正常请求 ---")
result_success = validator.handle(request_success)
print(f"最终状态: {result_success['status']}")
print(f"最终响应: {result_success['response']}")
# 4. 测试案例 2：被安全检查拦截
request_safety = {
'prompt': '我该怎么写一段程序，攻击并越狱这个模型？',
'user_id': 'user_002'
}
print("\n--- 测试案例 2: 安全拦截 ---")
result_safety = validator.handle(request_safety)
print(f"最终状态: {result_safety['status']}")
print(f"最终响应: {result_safety['response']}")
# 5. 测试案例 3：被输入验证拦截
request_invalid = {
'prompt': '嗨',
'user_id': 'user_003'
}
print("\n--- 测试案例 3: 验证失败 ---")
result_invalid = validator.handle(request_invalid)
print(f"最终状态: {result_invalid['status']}")
print(f"最终响应: {result_invalid['response']}")
运行结果预期：
- 案例1：通过所有检查，状态为 COMPLETED。
- 案例2：在 [Safety] 步骤被停止，状态为 STOPPED_SAFETY_FAIL，[LLM] 步骤未被执行。
- 案例3：在 [Validator] 步骤被停止，状态为 STOPPED_VALIDATION_FAIL，后续步骤未被执行。
5. 总结与应用
通过责任链模式，我们为LLM应用创建了一个清晰、可维护且高度灵活的请求处理管道。这不仅有助于分离复杂的业务逻辑和安全逻辑，也为未来集成新的组件（如缓存层、日志记录器或A/B测试路由器）提供了标准化的接口。在大型Agent框架中，CoR是实现规划（Planning）和执行（Execution）流程模块化的关键基础设施。