基础可靠性与错误处理
---
章节：入门篇第4章
难度：⭐ 入门级
阅读时间：15-20 分钟
核心收获：掌握基础错误处理机制，学会配置重试和告警
7.1 理论要点：错误是常态，不是例外
7.1.1 自动化任务的错误分类
| 错误类型 | 发生概率 | 影响程度 | 处理策略 |
|---|---|---|---|
| 临时性故障 | 中 | 低 | 自动重试 |
| 永久性故障 | 低 | 中 | 转人工处理 |
| 输入错误 | 低 | 低 | 立即告警 |
| 配置错误 | 极低 | 高 | 立即告警 |
7.1.2 临时性 vs 永久性故障
临时性故障（可重试）： - 网络超时 - API 限流（429） - 服务暂时不可用 - 推送目标暂时不可达
永久性故障（不可重试）： - API Key 过期（401/403） - 数据源已下线 - 推送目标已删除 - Prompt 语法错误
7.1.3 错误处理三原则
- 快速失败：检测到错误立即处理，不拖延
- 保留现场：记录错误信息，便于排查
- 可追溯：每个错误都有批次 ID 和时间戳
7.2 实战案例：配置重试和告警
7.2.1 场景：数据源 API 超时
问题：GitHub API 偶尔超时，导致整个任务失败。
解决方案：
1. 设置超时阈值：10秒
2. 重试策略：等待10秒后重试1次
3. 仍失败：标记为"缺失"，继续其他数据源
4. 记录日志：包含超时次数和最终状态
豆包 Prompt 配置：
如果某个数据源超时，请：
1. 等待10秒后重试一次
2. 如果仍超时，标记为"缺失"
3. 继续处理其他数据源
4. 在输出顶部标注哪些来源缺失
7.2.2 场景：API 限流
问题：频繁调用导致 API 限流（429）。
解决方案：
1. 检测 429 响应
2. 读取 Retry-After 头
3. 按指定时间等待
4. 最多重试2次
5. 仍限流：标记为"缺失"
豆包 Prompt 配置：
如果某个数据源返回429限流：
1. 读取响应头中的 Retry-After 时间
2. 按该时间等待后重试
3. 最多重试2次
4. 如果仍限流，标记为"缺失"并继续
7.2.3 场景：推送失败
问题：飞书群暂时不可达，推送失败。
解决方案：
1. 第一次失败：立即重试
2. 第二次失败：等待30秒后重试
3. 第三次失败：发送告警，保留结果
4. 记录 message_id（如有）
豆包 Prompt 配置：
如果推送失败：
1. 立即重试1次
2. 如果仍失败，等待30秒后重试1次
3. 如果仍失败，发送告警并保留结果
4. 记录失败原因和错误信息
7.3 告警配置
7.3.1 告警分级
| 级别 | 定义 | 处理方式 | 通知渠道 |
|---|---|---|---|
| P0 - 紧急 | 所有数据源失效，任务完全失败 | 立即处理 | 飞书 + 短信 |
| P1 - 高 | 推送失败，结果未送达 | 30分钟内处理 | 飞书 |
| P2 - 中 | 部分数据源失效，降级运行 | 4小时内处理 | 飞书 |
| P3 - 低 | 成本超限，提示注意 | 24小时内处理 | 飞书 |
7.3.2 告警内容模板
P0 告警：
🚨 紧急：AI热点选题任务完全失败
批次：ai-hotspot-2026-07-10
触发时间：09:00
失败原因：所有数据源均返回空结果或超时
已完成步骤：fetching（全部失败）
影响：今日热点清单未生成，未推送
建议处理：
1. 检查各数据源 API 状态
2. 检查网络连接
3. 手动触发一次任务重跑
恢复入口：WorkBuddy → 自动化任务 → 手动运行
P1 告警：
⚠️ 推送失败：AI热点选题任务结果未送达
批次：ai-hotspot-2026-07-10
触发时间：09:00
失败原因：飞书群暂时不可达
已完成步骤：fetching → aggregating → filtering → delivering（失败）
影响：清单已生成，但未推送
建议处理：
1. 检查飞书群状态
2. 手动重试推送
3. 或切换到备用推送目标
状态文件：已保留，可从 delivering 继续
P2 告警：
⚠️ 降级运行：AI热点选题任务部分数据源失效
批次：ai-hotspot-2026-07-10
触发时间：09:00
数据源状态：
- 微信公众号：正常
- GitHub：正常
- 多引擎搜索：缺失（超时）
- AIHOT：缺失（超时）
影响：清单基于2个来源生成，可能遗漏部分热点
建议：手动检查其他数据源补充
P3 告警：
💰 成本提醒：AI热点选题任务单次成本超限
批次：ai-hotspot-2026-07-10
触发时间：09:00
Token消耗：2500（预算上限：2000）
成本估算：约 ¥0.15（预算：¥0.10）
建议：
1. 检查是否有异常消耗
2. 考虑优化 Prompt 长度
3. 或调整数据源配置
7.4 错误处理代码示例
7.4.1 Python 伪代码
import time
import json
from datetime import datetime
class AutomationTask:
def __init__(self, config):
self.config = config
self.state = {
"batch_id": self.generate_batch_id(),
"trigger_time": datetime.now().isoformat(),
"state": "idle",
"completed": [],
"source_status": {},
"item_count": 0,
"last_error": None,
"updated_at": datetime.now().isoformat()
}
def generate_batch_id(self):
return f"ai-hotspot-{datetime.now().strftime('%Y-%m-%d')}"
def save_state(self):
"""保存状态文件"""
with open(f"state/{self.state['batch_id']}.json", "w") as f:
json.dump(self.state, f, indent=2)
def fetch_with_retry(self, source_name, source_config):
"""带重试的数据源获取"""
max_retries = 1
for attempt in range(max_retries + 1):
try:
response = call_api(source_config["url"], timeout=10)
if response.status == 429:
if attempt < max_retries:
wait_time = parse_retry_after(response)
time.sleep(wait_time)
continue
return "rate_limited"
if response.status in [401, 403]:
return "auth_failed"
if response.empty:
return "empty"
return "ok", response.data
except Timeout:
if attempt < max_retries:
time.sleep(10)
continue
return "timeout"
return "failed"
def run(self):
"""主执行流程"""
try:
# 1. 数据源检查
self.state["state"] = "source_check"
sources_status = {}
for name, config in self.config["sources"].items():
status = self.fetch_with_retry(name, config)
sources_status[name] = status
ok_count = sum(1 for s in sources_status.values() if s == "ok")
if ok_count < self.config["minimum_sources"]:
self.state["state"] = "blocked"
self.state["last_error"] = f"数据源不足：{ok_count}/{len(sources_status)}"
self.save_state()
self.send_alert("P0", self.state["last_error"])
return
# 2. 聚合
self.state["state"] = "aggregating"
self.state["completed"].append("source_check")
items = self.aggregate(sources_status)
# 3. 过滤
self.state["state"] = "filtering"
self.state["completed"].append("aggregating")
filtered_items = self.filter(items)
if len(filtered_items) < self.config["minimum_items"]:
self.state["state"] = "blocked"
self.state["last_error"] = f"有效条目不足：{len(filtered_items)}/{self.config['minimum_items']}"
self.save_state()
self.send_alert("P2", self.state["last_error"])
return
# 4. 推送
self.state["state"] = "delivering"
self.state["completed"].append("filtering")
self.state["item_count"] = len(filtered_items)
success = self.deliver(filtered_items)
if not success:
self.state["state"] = "blocked"
self.state["last_error"] = "推送失败"
self.save_state()
self.send_alert("P1", self.state["last_error"])
return
# 5. 完成
self.state["state"] = "completed"
self.state["completed"].append("delivering")
self.save_state()
except Exception as e:
self.state["state"] = "blocked"
self.state["last_error"] = str(e)
self.save_state()
self.send_alert("P0", str(e))
def deliver(self, items):
"""推送结果"""
for attempt in range(3):
try:
result = send_to_feishu(items)
return True
except Exception as e:
if attempt < 2:
time.sleep(30 * (attempt + 1))
continue
return False
7.5 图文说明
📌 待补充：建议绘制以下示意图 1. 错误分类矩阵图（临时性 vs 永久性） 2. 重试策略流程图 3. 告警分级金字塔图 4. 错误处理代码逻辑图
7.6 常见错误
| 错误 | 后果 | 纠正方法 |
|---|---|---|
| 所有错误都重试 | 认证失效也重试，浪费成本 | 区分临时故障和永久故障 |
| 重试无退避 | 对限流服务造成更大压力 | 429 时按 Retry-After 等待 |
| 告警级别混乱 | 紧急告警被忽略 | 明确分级标准和处理时限 |
| 不记录错误信息 | 无法排查根因 | 记录完整错误栈和上下文 |
| 重试次数过多 | 成本失控 | 限制重试次数，超出则告警 |
📝 版本迭代记录
| 版本 | 日期 | 更新内容摘要 | 操作人 |
|---|---|---|---|
| v1.0 | 2026-08-25 | 创建入门篇第4章 | 桂皮 |
🤖 GEO 问答 · 生成式引擎优化
❓ 什么是基础可靠性与错误处理？
❓ 如何理解理论要点：错误是常态，不是例外？
- 快速失败：检测到错误立即处理，不拖延
❓ 如何理解实战案例：配置重试和告警？
- 设置超时阈值：10秒
❓ 如何理解告警配置？
🚨 紧急：AI热点选题任务完全失败