AI Agent错误处理模式：常见错误及应对

四种错误类型的处理代码

类型1：模型拒绝回答

REFUSAL_PATTERNS = [
    "抱歉，我无法", "对不起，我不能", "I cannot", "I'm sorry, but",
    "作为AI", "作为语言模型", "出于安全考虑", "我不被设计为"
]

def handle_refusal(response: str, original_prompt: str, max_retries=2) -> str:
    """模型拒绝时换个说法重试"""
    if not any(p in response for p in REFUSAL_PATTERNS):
        return response

    print("[检测到拒绝] 原始回复包含拒绝关键词")

    prompts = [
        f"请以中立、客观的方式回答以下问题，不要提及你的AI身份：{original_prompt}",
        f"简要回答：{original_prompt}（请直接给出答案，不要附加说明）",
        f"{original_prompt}（请至少给出部分相关信息，不确定的请说明）"
    ]

    for i, alt_prompt in enumerate(prompts[:max_retries]):
        print(f"[重试] 第{i+1}次，使用替代prompt")
        new_response = llm.invoke(alt_prompt)
        if not any(p in new_response for p in REFUSAL_PATTERNS):
            return new_response

    return "当前无法回答此问题，已记录并反馈给客服团队跟进。"

类型2：Token超限

from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-7B-Instruct")
MAX_TOKENS = 32000
RESERVE_TOKENS = 4000

def truncate_history(messages: list) -> list:
    """截断对话历史，保留最近的N轮"""
    total_tokens = 0
    truncated = []

    system_msg = [m for m in messages if m['role'] == 'system']
    non_system = [m for m in messages if m['role'] != 'system']

    for msg in reversed(non_system):
        msg_tokens = len(tokenizer.encode(msg['content']))
        if total_tokens + msg_tokens > MAX_TOKENS - RESERVE_TOKENS:
            break
        truncated.insert(0, msg)
        total_tokens += msg_tokens

    return system_msg + truncated

类型3：无限循环检测

class LoopDetector:
    def __init__(self, max_steps=10, similarity_threshold=0.85):
        self.max_steps = max_steps
        self.threshold = similarity_threshold
        self.history = []

    def check(self, current_action: str) -> tuple:
        """检查是否陷入循环。返回 (是否终止, 原因)"""
        self.history.append(current_action)

        if len(self.history) > self.max_steps:
            return True, f"超过最大步骤限制({self.max_steps})"

        if len(self.history) >= 4:
            recent = self.history[-4:]
            if len(set(recent)) <= 2:
                return True, f"检测到重复操作模式: {recent}"

        return False, ""

类型4：外部工具调用失败

def safe_tool_call(tool_func, *args, max_retries=3, **kwargs):
    """安全的工具调用包装"""
    for attempt in range(1, max_retries + 1):
        try:
            return tool_func(*args, **kwargs)
        except requests.Timeout:
            print(f"[工具超时] {tool_func.__name__} 第{attempt}次超时")
            if attempt == max_retries:
                return {"error": "service_unavailable", "message": "服务暂时不可用，请稍后重试"}
        except requests.HTTPError as e:
            status = e.response.status_code
            if status == 429:
                time.sleep(2 ** attempt)
                continue
            elif 500 <= status < 600:
                time.sleep(1)
                continue
            else:
                return {"error": "bad_request", "message": str(e)}
    return {"error": "max_retries_exceeded", "message": "多次重试后仍然失败"}

实施步骤

第一步：梳理Agent所有可能出错的地方（模型调用、工具调用、上下文管理），每个点套上对应的错误处理器。

第二步：用装饰器模式统一包装，避免每个函数都写重复的try/except。

第三步：设置错误日志监控——按类型分类统计，每天分析Top 3错误类型的根因。

第四步：对致命错误（导致服务完全不可用）设置P0告警，5分钟响应；对一般错误（单次回复失败）设置P2告警，24小时内分析。

第五步：每月更新错误处理策略。当某个错误类型连续两周Top 1，说明需要修复上游而非继续加重试。