让你的 Agent 像生物体一样拥有免疫系统——故障不可避免，但崩溃可以。
一、引言：自愈是 Agent 的”免疫系统”
人类的身体每天都在与病毒、细菌、异常细胞战斗。我们之所以能健康地活着，不是因为没有病原体入侵，而是因为我们拥有一套精密的免疫系统——它能识别威胁、隔离损害、修复创伤，甚至在下次遇到同类威胁时做出更快的响应。
对于 AI Agent 而言，自我修复（Self-Healing）就是它的免疫系统。
在前三篇文章中，我们分别了解了Loop Engineering的核心概念、动手设计了第一个Loop、并深入学习了反馈循环设计。这些技术让 Agent 能够高效地完成任务。但一个真正可靠的 Agent，不仅要能在理想条件下工作，更要能在各种异常情况下生存下来。
现实世界充满了不确定性：
- API 服务可能在凌晨 3 点突然宕机
- 网络连接可能在传输大文件时中断
- 模型可能返回格式错误的 JSON
- 内存可能在处理大量数据时逐渐耗尽
- 一个看似简单的任务可能触发无限循环
如果没有自我修复能力，这些异常中的任何一个都可能导致 Agent 崩溃、任务失败、甚至造成数据损失。而一个拥有良好自愈机制的 Agent，能够在检测到问题后自动采取恢复措施，继续完成任务，或者至少优雅地降级并保存现场。
本文将深入探讨如何为 Agent Loop 构建完整的自我修复机制。我们将从故障模式的分类开始，逐步建立错误检测、自愈策略、断路器保护和循环防护的完整体系。
二、故障模式分类
要构建有效的自愈机制，首先需要理解 Agent 可能面临的各种故障模式。根据故障的性质和影响范围，我们可以将其分为四大类。
2.1 外部依赖失败
这是最常见也是最容易理解的故障类型。Agent 通常依赖多个外部服务：
API 调用失败
`python
# 典型的 API 失败场景
- 网络超时：请求发送后长时间没有响应
- 服务端错误：5xx 状态码（500, 502, 503）
- 限流：429 Too Many Requests
- 认证失败：401 Unauthorized / 403 Forbidden
- 数据格式错误：返回的 JSON 不符合预期 schema
`
数据库连接问题
`python
- 连接池耗尽
- 查询超时
- 死锁
- 主从同步延迟
`
第三方服务中断
`python
- 搜索引擎 API 不可用
- 云存储服务故障
- 消息队列服务宕机
`
2.2 逻辑错误
逻辑错误是 Agent 内部的”软件缺陷”，通常更难检测：
状态不一致
`python
# Agent 认为自己处于状态 A，但实际已进入状态 B
# 例如：任务已被标记为完成，但实际上还有子任务未执行
class AgentState:
def __init__(self):
self.completed_tasks = set()
self.pending_tasks = set()
def mark_complete(self, task_id):
self.completed_tasks.add(task_id)
# 漏掉了从 pending_tasks 中移除！
# self.pending_tasks.discard(task_id) # 这行被遗漏了
`
边界条件处理不当
`python
# 处理空列表时崩溃
def process_items(items):
result = items[0] # 如果 items 为空，这里会抛出 IndexError
for item in items[1:]:
result = combine(result, item)
return result
`
类型假设错误
`python
# 假设 API 总是返回特定格式
def parse_response(response):
data = response.json()
return data[“results”][“items”] # 如果 “results” 不存在呢？
`
2.3 资源耗尽
资源问题是渐进式的，通常在运行一段时间后才显现：
内存泄漏
`python
# 每次迭代都在累积数据，但从不清理
class DataProcessor:
def __init__(self):
self.history = [] # 这个列表会无限增长
def process(self, data):
result = transform(data)
self.history.append((data, result)) # 不断添加，从不清理
return result
`
磁盘空间不足
`python
- 日志文件不断增长
- 缓存文件未及时清理
- 临时文件堆积
`
连接池耗尽
`python
# 获取连接但忘记释放
async def fetch_data():
conn = await pool.acquire()
data = await conn.execute(query)
return data # 忘记 await pool.release(conn)
`
CPU 过载
`python
- 正则表达式回溯导致灾难性性能问题
- 算法复杂度超出预期
- 竞态条件导致忙等待
`
2.4 死循环与无限递归
这是最危险的故障类型之一，因为它会持续消耗资源直到系统崩溃：
逻辑死循环
`python
# 条件永远无法满足
while not task_complete:
result = check_status()
# 忘记更新 task_complete 的状态
# 或者 check_status() 总是返回相同的结果
`
相互调用死循环
`python
# Agent A 等待 Agent B 的结果
# Agent B 等待 Agent A 的结果
async def agent_a():
result_b = await agent_b()
return process(result_b)
async def agent_b():
result_a = await agent_a() # 永远等不到
return transform(result_a)
`
重试风暴
`python
# 失败后立即重试，不加任何延迟或限制
async def unreliable_operation():
while True:
try:
return await external_api()
except Exception:
pass # 永远重试下去，而且没有退避
`
2.5 故障的影响层级
理解故障的影响范围同样重要：
| 层级 | 影响范围 | 恢复难度 | 示例 |
|---|
|——|———-|———-|——|
| 临时性 | 单次操作 | 低 | API 超时，重试即可 |
|---|---|---|---|
| 局部性 | 当前子任务 | 中 | 模型输出格式错误，需要重新生成 |
| 全局性 | 整个 Agent | 高 | 内存耗尽，需要重启 |
| 级联性 | 多个 Agent | 很高 | 共享资源故障导致多个 Agent 同时失败 |
三、错误检测机制
发现问题比解决问题更难。很多故障在初期是隐蔽的，只有建立了有效的检测机制，才能在问题恶化前采取行动。
3.1 基于阈值的检测
最简单也最直接的检测方式：当某个指标超过预设阈值时触发告警。
`python
class ThresholdDetector:
def __init__(self):
self.thresholds = {
“response_time_ms”: 5000, # 响应时间超过5秒
“error_rate”: 0.1, # 错误率超过10%
“memory_usage_mb”: 1024, # 内存使用超过1GB
“queue_length”: 1000, # 队列长度超过1000
“retry_count”: 3, # 重试次数超过3次
“consecutive_failures”: 5, # 连续失败超过5次
}
self.metrics = defaultdict(list)
def record(self, metric_name, value):
self.metrics[metric_name].append({
“value”: value,
“timestamp”: time.time()
})
self._cleanup_old_metrics(metric_name)
def check_violations(self):
violations = []
for metric_name, threshold in self.thresholds.items():
recent_values = self._get_recent_values(metric_name)
if not recent_values:
continue
current_value = recent_values[-1]
if metric_name in [“error_rate”]:
avg_value = sum(recent_values) / len(recent_values)
if avg_value > threshold:
violations.append({
“metric”: metric_name,
“value”: avg_value,
“threshold”: threshold,
“type”: “average”
})
else:
if current_value > threshold:
violations.append({
“metric”: metric_name,
“value”: current_value,
“threshold”: threshold,
“type”: “instant”
})
return violations
`
3.2 异常检测
更智能的检测方式是识别”不正常”的模式，而不仅仅是超过固定阈值：
`python
class AnomalyDetector:
def __init__(self, window_size=100, sigma_threshold=3):
self.window_size = window_size
self.sigma_threshold = sigma_threshold
self.history = deque(maxlen=window_size)
def add_observation(self, value):
self.history.append(value)
def is_anomaly(self, value):
if len(self.history) < 10: # 需要足够的历史数据
return False
mean = np.mean(self.history)
std = np.std(self.history)
if std == 0: # 所有历史值都相同
return value != mean
z_score = abs(value – mean) / std
return z_score > self.sigma_threshold
def detect_trend_anomaly(self):
“””检测趋势异常：值是否在持续恶化”””
if len(self.history) < 20:
return False
recent = list(self.history)[-10:]
older = list(self.history)[-20:-10]
recent_mean = np.mean(recent)
older_mean = np.mean(older)
# 如果最近的均值比之前高了50%以上
return recent_mean > older_mean * 1.5
`
3.3 健康检查
定期主动检查系统各组件的状态：
`python
class HealthChecker:
def __init__(self):
self.checks = {}
self.results = {}
def register_check(self, name, check_fn, critical=False):
self.checks[name] = {
“fn”: check_fn,
“critical”: critical,
“last_run”: None,
“interval”: 60 # 默认每60秒检查一次
}
async def run_checks(self):
results = {}
all_healthy = True
for name, check_config in self.checks.items():
now = time.time()
if check_config[“last_run”] and
now – check_config[“last_run”] < check_config["interval"]:
results[name] = self.results.get(name, {“status”: “skipped”})
continue
try:
start_time = time.time()
is_healthy = await check_config[“fn”]()
duration = time.time() – start_time
results[name] = {
“status”: “healthy” if is_healthy else “unhealthy”,
“duration_ms”: duration * 1000,
“critical”: check_config[“critical”],
“timestamp”: now
}
if not is_healthy and check_config[“critical”]:
all_healthy = False
except Exception as e:
results[name] = {
“status”: “error”,
“error”: str(e),
“critical”: check_config[“critical”],
“timestamp”: now
}
if check_config[“critical”]:
all_healthy = False
check_config[“last_run”] = now
self.results = results
return all_healthy, results
# 注册各种健康检查
health_checker = HealthChecker()
# 检查数据库连接
async def check_database():
try:
await db.execute(“SELECT 1”)
return True
except Exception:
return False
# 检查 API 可用性
async def check_external_api():
try:
response = await api_client.get(“/health”, timeout=5)
return response.status == 200
except Exception:
return False
# 检查磁盘空间
def check_disk_space():
import shutil
total, used, free = shutil.disk_usage(“/”)
return free / total > 0.1 # 至少10%空闲空间
# 检查内存使用
def check_memory():
import psutil
memory = psutil.virtual_memory()
return memory.percent < 90 # 内存使用不超过90%
health_checker.register_check(“database”, check_database, critical=True)
health_checker.register_check(“api”, check_external_api, critical=True)
health_checker.register_check(“disk”, check_disk_space, critical=False)
health_checker.register_check(“memory”, check_memory, critical=True)
`
3.4 心跳检测
对于长时间运行的任务，心跳机制可以检测任务是否”卡住”：
`python
class HeartbeatMonitor:
def __init__(self, timeout_seconds=300):
self.timeout = timeout_seconds
self.last_heartbeat = {}
self.callbacks = {}
def register_task(self, task_id, on_timeout=None):
self.last_heartbeat[task_id] = time.time()
if on_timeout:
self.callbacks[task_id] = on_timeout
def heartbeat(self, task_id):
if task_id in self.last_heartbeat:
self.last_heartbeat[task_id] = time.time()
def check_stale_tasks(self):
now = time.time()
stale_tasks = []
for task_id, last_beat in self.last_heartbeat.items():
if now – last_beat > self.timeout:
stale_tasks.append(task_id)
if task_id in self.callbacks:
try:
self.callbackstask_id
except Exception as e:
logger.error(f”Heartbeat callback failed for {task_id}: {e}”)
return stale_tasks
# 在 Agent 中使用心跳
class AgentWithHeartbeat:
def __init__(self):
self.heartbeat_monitor = HeartbeatMonitor(timeout_seconds=120)
async def run_long_task(self, task_id):
self.heartbeat_monitor.register_task(task_id, self._handle_stale)
try:
while not self.is_task_complete(task_id):
# 执行一小步
await self.execute_step(task_id)
# 发送心跳
self.heartbeat_monitor.heartbeat(task_id)
await asyncio.sleep(0.1)
except TaskStaleError:
logger.warning(f”Task {task_id} became stale, initiating recovery”)
await self.recover_task(task_id)
def _handle_stale(self, task_id):
raise TaskStaleError(f”Task {task_id} hasn’t sent heartbeat”)
`
3.5 断言与不变量检查
在关键节点设置断言，确保系统状态符合预期：
`python
class InvariantChecker:
def __init__(self):
self.invariants = []
def add_invariant(self, name, check_fn, error_msg):
self.invariants.append({
“name”: name,
“check”: check_fn,
“error_msg”: error_msg
})
def verify_all(self, context):
violations = []
for inv in self.invariants:
try:
if not inv“check”:
violations.append({
“name”: inv[“name”],
“error”: inv[“error_msg”]
})
except Exception as e:
violations.append({
“name”: inv[“name”],
“error”: f”Check failed with exception: {e}”
})
if violations:
raise InvariantViolationError(violations)
return True
# 使用示例
invariant_checker = InvariantChecker()
# 添加不变量
invariant_checker.add_invariant(
“task_count_consistency”,
lambda ctx: len(ctx[“completed”]) + len(ctx[“pending”]) == ctx[“total”],
“Task count inconsistency: completed + pending != total”
)
invariant_checker.add_invariant(
“no_duplicate_processing”,
lambda ctx: len(ctx[“in_progress”]) == len(set(ctx[“in_progress”])),
“Duplicate tasks detected in progress queue”
)
invariant_checker.add_invariant(
“resource_within_limits”,
lambda ctx: ctx[“memory_used”] < ctx["memory_limit"],
“Memory usage exceeds limit”
)
`
四、自愈策略
检测到问题后，需要选择合适的恢复策略。不同的故障类型需要不同的应对方式。
4.1 重试策略
重试是最基本的恢复策略，但简单地重试往往不够，需要考虑：
`python
class RetryStrategy:
def __init__(
self,
max_retries=3,
base_delay=1.0,
max_delay=60.0,
exponential_base=2,
jitter=True
):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.exponential_base = exponential_base
self.jitter = jitter
def get_delay(self, attempt):
# 指数退避
delay = self.base_delay * (self.exponential_base ** attempt)
delay = min(delay, self.max_delay)
# 添加抖动，避免多个客户端同时重试
if self.jitter:
delay = delay * (0.5 + random.random())
return delay
def should_retry(self, exception, attempt):
# 某些异常不应该重试
non_retryable = [
AuthenticationError,
PermissionDeniedError,
ValidationError,
NotFoundError
]
for exc_type in non_retryable:
if isinstance(exception, exc_type):
return False
return attempt < self.max_retries
async def with_retry(fn, retry_strategy, *args, **kwargs):
last_exception = None
for attempt in range(retry_strategy.max_retries + 1):
try:
result = await fn(*args, **kwargs)
return result
except Exception as e:
last_exception = e
if not retry_strategy.should_retry(e, attempt):
raise
delay = retry_strategy.get_delay(attempt)
logger.warning(
f”Attempt {attempt + 1} failed: {e}. “
f”Retrying in {delay:.2f}s…”
)
await asyncio.sleep(delay)
raise last_exception
`
4.2 降级策略
当主要方案不可用时，切换到备选方案：
`python
class FallbackChain:
def __init__(self):
self.strategies = []
def add_strategy(self, name, fn, priority=0, condition=None):
self.strategies.append({
“name”: name,
“fn”: fn,
“priority”: priority,
“condition”: condition,
“failures”: 0,
“last_failure”: None
})
self.strategies.sort(key=lambda s: s[“priority”], reverse=True)
async def execute(self, *args, **kwargs):
errors = []
for strategy in self.strategies:
# 检查是否满足执行条件
if strategy[“condition”] and not strategy[“condition”]():
continue
try:
result = await strategy“fn”
return {
“result”: result,
“strategy”: strategy[“name”]
}
except Exception as e:
strategy[“failures”] += 1
strategy[“last_failure”] = time.time()
errors.append({
“strategy”: strategy[“name”],
“error”: str(e)
})
logger.warning(
f”Strategy ‘{strategy[‘name’]}’ failed: {e}”
)
raise AllStrategiesFailedError(errors)
# 使用示例：多模型降级
model_chain = FallbackChain()
# 优先使用最强的模型
model_chain.add_strategy(
“gpt4”,
lambda q: openai_client.complete(q, model=”gpt-4″),
priority=100
)
# 次选使用较快的模型
model_chain.add_strategy(
“gpt35”,
lambda q: openai_client.complete(q, model=”gpt-3.5-turbo”),
priority=50
)
# 最后使用本地模型
model_chain.add_strategy(
“local”,
lambda q: local_model.complete(q),
priority=10
)
`
4.3 回滚策略
当操作产生副作用且失败时，需要能够回滚：
`python
class TransactionManager:
def __init__(self):
self.operations = []
self.rollback_handlers = []
def add_operation(self, operation, rollback_handler):
self.operations.append({
“execute”: operation,
“rollback”: rollback_handler,
“executed”: False,
“result”: None
})
async def execute_all(self):
executed_ops = []
try:
for op in self.operations:
result = await op[“execute”]()
op[“result”] = result
op[“executed”] = True
executed_ops.append(op)
return [op[“result”] for op in self.operations]
except Exception as e:
logger.error(f”Transaction failed at operation: {e}”)
# 按逆序回滚已执行的操作
for op in reversed(executed_ops):
try:
await op“rollback”
logger.info(f”Rolled back operation successfully”)
except Exception as rollback_error:
logger.critical(
f”Rollback failed: {rollback_error}. “
f”System may be in inconsistent state!”
)
raise TransactionFailedError(e)
# 使用示例
async def deploy_new_version():
tx = TransactionManager()
# 步骤1: 备份当前版本
tx.add_operation(
operation=lambda: create_backup(),
rollback=lambda backup: delete_backup(backup)
)
# 步骤2: 停止当前服务
tx.add_operation(
operation=lambda: stop_service(),
rollback=lambda _: start_service()
)
# 步骤3: 部署新版本
tx.add_operation(
operation=lambda: deploy_files(),
rollback=lambda _: restore_backup()
)
# 步骤4: 启动新服务
tx.add_operation(
operation=lambda: start_service(),
rollback=lambda _: stop_service()
)
return await tx.execute_all()
`
4.4 重启策略
有时最简单的恢复方式就是重启：
`python
class RestartManager:
def __init__(self, max_restarts=3, restart_window=300):
self.max_restarts = max_restarts
self.restart_window = restart_window # 秒
self.restart_history = []
def can_restart(self):
now = time.time()
# 清理过期记录
self.restart_history = [
t for t in self.restart_history
if now – t < self.restart_window
]
return len(self.restart_history) < self.max_restarts
def record_restart(self):
self.restart_history.append(time.time())
async def restart_with_backoff(self, component_factory):
if not self.can_restart():
raise MaxRestartsExceededError(
f”Exceeded {self.max_restarts} restarts “
f”in {self.restart_window}s”
)
restart_count = len(self.restart_history)
delay = min(2 ** restart_count, 60) # 指数退避，最大60秒
logger.info(f”Restarting in {delay}s (attempt {restart_count + 1})”)
await asyncio.sleep(delay)
self.record_restart()
component = component_factory()
await component.start()
return component
# 在 Supervisor 中使用
class ProcessSupervisor:
def __init__(self):
self.restart_manager = RestartManager()
self.process = None
async def run(self, process_factory):
while True:
try:
self.process = process_factory()
await self.process.run()
# 正常退出
if self.process.exit_code == 0:
break
except Exception as e:
logger.error(f”Process crashed: {e}”)
try:
self.process = await self.restart_manager.restart_with_backoff(
process_factory
)
except MaxRestartsExceededError:
logger.critical(“Too many restarts, giving up”)
raise
`
五、断路器模式
断路器（Circuit Breaker）是处理外部依赖故障的核心模式。它模仿电路中的断路器：当检测到故障时，自动”断开”电路，防止故障蔓延。
5.1 基本断路器实现
`python
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, Any
import time
import asyncio
class CircuitState(Enum):
CLOSED = “closed” # 正常状态，允许请求通过
OPEN = “open” # 断开状态，拒绝所有请求
HALF_OPEN = “half_open” # 半开状态，允许少量请求测试
@dataclass
class CircuitBreakerStats:
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
consecutive_failures: int = 0
last_failure_time: float = 0
last_success_time: float = 0
class CircuitBreaker:
def __init__(
self,
name: str,
failure_threshold: int = 5,
recovery_timeout: float = 60.0,
half_open_max_calls: int = 3,
fallback: Callable = None
):
self.name = name
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_calls = half_open_max_calls
self.fallback = fallback
self.state = CircuitState.CLOSED
self.stats = CircuitBreakerStats()
self.half_open_calls = 0
self._lock = asyncio.Lock()
async def call(self, func: Callable, *args, **kwargs) -> Any:
async with self._lock:
self._check_state_transition()
if self.state == CircuitState.OPEN:
if self.fallback:
logger.warning(f”Circuit {self.name} is OPEN, using fallback”)
return await self.fallback(*args, **kwargs)
raise CircuitOpenError(f”Circuit {self.name} is OPEN”)
if self.state == CircuitState.HALF_OPEN:
if self.half_open_calls >= self.half_open_max_calls:
raise CircuitOpenError(
f”Circuit {self.name} half-open limit reached”
)
self.half_open_calls += 1
self.stats.total_requests += 1
try:
result = await func(*args, **kwargs)
await self._on_success()
return result
except Exception as e:
await self._on_failure(e)
raise
def _check_state_transition(self):
now = time.time()
if self.state == CircuitState.OPEN:
if now – self.stats.last_failure_time >= self.recovery_timeout:
logger.info(f”Circuit {self.name} transitioning to HALF_OPEN”)
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
async def _on_success(self):
async with self._lock:
self.stats.successful_requests += 1
self.stats.consecutive_failures = 0
self.stats.last_success_time = time.time()
if self.state == CircuitState.HALF_OPEN:
logger.info(f”Circuit {self.name} recovering, closing circuit”)
self.state = CircuitState.CLOSED
async def _on_failure(self, error):
async with self._lock:
self.stats.failed_requests += 1
self.stats.consecutive_failures += 1
self.stats.last_failure_time = time.time()
if self.stats.consecutive_failures >= self.failure_threshold:
logger.warning(
f”Circuit {self.name} tripping OPEN “
f”(failures: {self.stats.consecutive_failures})”
)
self.state = CircuitState.OPEN
elif self.state == CircuitState.HALF_OPEN:
logger.warning(f”Circuit {self.name} half-open test failed”)
self.state = CircuitState.OPEN
def get_status(self):
return {
“name”: self.name,
“state”: self.state.value,
“stats”: {
“total”: self.stats.total_requests,
“success”: self.stats.successful_requests,
“failed”: self.stats.failed_requests,
“consecutive_failures”: self.stats.consecutive_failures
}
}
`
5.2 滑动窗口断路器
基本断路器只看连续失败次数，滑动窗口更精确：
`python
class SlidingWindowCircuitBreaker:
def __init__(
self,
name: str,
window_size: int = 60, # 窗口大小（秒）
failure_rate_threshold: float = 0.5, # 失败率阈值
min_calls: int = 10, # 窗口内最少调用次数
recovery_timeout: float = 30.0
):
self.name = name
self.window_size = window_size
self.failure_rate_threshold = failure_rate_threshold
self.min_calls = min_calls
self.recovery_timeout = recovery_timeout
self.state = CircuitState.CLOSED
self.call_history = deque() # (timestamp, success)
self.last_state_change = time.time()
def _cleanup_window(self):
cutoff = time.time() – self.window_size
while self.call_history and self.call_history[0][0] < cutoff:
self.call_history.popleft()
def _calculate_failure_rate(self):
if len(self.call_history) < self.min_calls:
return 0.0
failures = sum(1 for _, success in self.call_history if not success)
return failures / len(self.call_history)
async def call(self, func, *args, **kwargs):
self._check_recovery()
if self.state == CircuitState.OPEN:
raise CircuitOpenError(f”Circuit {self.name} is OPEN”)
try:
result = await func(*args, **kwargs)
self._record_call(True)
return result
except Exception as e:
self._record_call(False)
raise
def _record_call(self, success):
self.call_history.append((time.time(), success))
self._cleanup_window()
failure_rate = self._calculate_failure_rate()
if self.state == CircuitState.CLOSED:
if failure_rate >= self.failure_rate_threshold:
logger.warning(
f”Circuit {self.name} tripping OPEN “
f”(failure rate: {failure_rate:.2%})”
)
self.state = CircuitState.OPEN
self.last_state_change = time.time()
elif self.state == CircuitState.HALF_OPEN:
if not success:
self.state = CircuitState.OPEN
self.last_state_change = time.time()
elif failure_rate < self.failure_rate_threshold / 2:
self.state = CircuitState.CLOSED
self.last_state_change = time.time()
def _check_recovery(self):
if self.state == CircuitState.OPEN:
if time.time() – self.last_state_change >= self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
self.last_state_change = time.time()
`
5.3 断路器的集成使用
`python
class ResilientAPIClient:
def __init__(self):
self.circuit_breakers = {}
def get_circuit_breaker(self, service_name):
if service_name not in self.circuit_breakers:
self.circuit_breakers[service_name] = CircuitBreaker(
name=service_name,
failure_threshold=5,
recovery_timeout=60,
fallback=self._get_fallback(service_name)
)
return self.circuit_breakers[service_name]
def _get_fallback(self, service_name):
fallbacks = {
“search”: self._search_fallback,
“translate”: self._translate_fallback,
}
return fallbacks.get(service_name)
async def _search_fallback(self, query, **kwargs):
# 使用缓存结果或本地搜索
cached = await cache.get(f”search:{query}”)
if cached:
return cached
return {“results”: [], “source”: “fallback”}
async def call_service(self, service_name, method, *args, **kwargs):
cb = self.get_circuit_breaker(service_name)
try:
return await cb.call(method, *args, **kwargs)
except CircuitOpenError:
logger.warning(f”Service {service_name} unavailable”)
raise
def get_all_status(self):
return {
name: cb.get_status()
for name, cb in self.circuit_breakers.items()
}
`
六、循环保护
Agent Loop 最大的风险之一是失控——无限循环、无限递归、资源无限消耗。循环保护机制是最后一道防线。
6.1 最大迭代限制
`python
class IterationLimiter:
def __init__(self, max_iterations=100):
self.max_iterations = max_iterations
self.current_iteration = 0
def increment(self):
self.current_iteration += 1
if self.current_iteration >= self.max_iterations:
raise MaxIterationsExceeded(
f”Exceeded maximum iterations: {self.max_iterations}”
)
def reset(self):
self.current_iteration = 0
@property
def remaining(self):
return self.max_iterations – self.current_iteration
class ProtectedAgentLoop:
def __init__(self, max_iterations=100):
self.iteration_limiter = IterationLimiter(max_iterations)
async def run(self, task):
while not task.is_complete():
self.iteration_limiter.increment()
try:
result = await self.execute_step(task)
task.update(result)
except MaxIterationsExceeded:
logger.error(“Max iterations reached, stopping loop”)
task.mark_failed(“iteration_limit”)
break
`
6.2 超时保护
`python
class TimeoutProtector:
def __init__(self):
self.timeouts = {}
def set_timeout(self, operation_id, timeout_seconds):
self.timeouts[operation_id] = {
“timeout”: timeout_seconds,
“start_time”: time.time()
}
def check_timeout(self, operation_id):
if operation_id not in self.timeouts:
return False
config = self.timeouts[operation_id]
elapsed = time.time() – config[“start_time”]
return elapsed >= config[“timeout”]
def get_elapsed(self, operation_id):
if operation_id not in self.timeouts:
return 0
return time.time() – self.timeouts[operation_id][“start_time”]
async def with_timeout(coro, timeout_seconds, operation_name=”unknown”):
try:
return await asyncio.wait_for(coro, timeout=timeout_seconds)
except asyncio.TimeoutError:
logger.error(f”Operation {operation_name} timed out after {timeout_seconds}s”)
raise OperationTimeoutError(operation_name, timeout_seconds)
# 分层超时
class LayeredTimeout:
def __init__(self):
self.layers = {
“task”: 3600, # 整个任务：1小时
“step”: 300, # 单个步骤：5分钟
“operation”: 30, # 单个操作：30秒
“api_call”: 10, # API调用：10秒
}
async def execute_with_timeouts(self, task):
return await with_timeout(
self._execute_steps(task),
self.layers[“task”],
“task”
)
async def _execute_steps(self, task):
while not task.is_complete():
step_result = await with_timeout(
self._execute_operations(task),
self.layers[“step”],
“step”
)
task.update(step_result)
async def _execute_operations(self, task):
operations = task.get_next_operations()
results = []
for op in operations:
result = await with_timeout(
op.execute(),
self.layers[“operation”],
f”operation_{op.name}”
)
results.append(result)
return results
`
6.3 成本上限保护
对于使用 LLM API 的 Agent，成本控制至关重要：
`python
class CostGuard:
def __init__(self, budget_limit=10.0, currency=”USD”):
self.budget_limit = budget_limit
self.currency = currency
self.total_cost = 0.0
self.cost_history = []
self.alerts = []
def record_cost(self, amount, description=””):
self.total_cost += amount
self.cost_history.append({
“amount”: amount,
“total”: self.total_cost,
“description”: description,
“timestamp”: time.time()
})
# 检查是否接近预算上限
usage_ratio = self.total_cost / self.budget_limit
if usage_ratio >= 1.0:
raise BudgetExceededError(
f”Budget exceeded: {self.total_cost:.4f} / {self.budget_limit:.4f} {self.currency}”
)
elif usage_ratio >= 0.9:
self.alerts.append(f”WARNING: 90% budget used ({self.total_cost:.4f})”)
elif usage_ratio >= 0.75:
self.alerts.append(f”NOTICE: 75% budget used ({self.total_cost:.4f})”)
@property
def remaining_budget(self):
return self.budget_limit – self.total_cost
def can_afford(self, estimated_cost):
return self.total_cost + estimated_cost <= self.budget_limit
class CostAwareAgent:
def __init__(self, budget=10.0):
self.cost_guard = CostGuard(budget)
self.token_costs = {
“gpt-4”: {“input”: 0.03, “output”: 0.06}, # per 1K tokens
“gpt-3.5-turbo”: {“input”: 0.001, “output”: 0.002},
}
async def call_llm(self, model, messages, **kwargs):
# 估算成本
estimated_tokens = sum(len(m[“content”]) / 4 for m in messages)
estimated_cost = (estimated_tokens / 1000) * self.token_costs[model][“input”]
if not self.cost_guard.can_afford(estimated_cost):
raise BudgetExceededError(
f”Estimated cost {estimated_cost:.4f} exceeds remaining budget”
)
response = await llm_client.complete(model=model, messages=messages)
# 记录实际成本
actual_cost = self._calculate_actual_cost(model, response.usage)
self.cost_guard.record_cost(actual_cost, f”LLM call: {model}”)
return response
def _calculate_actual_cost(self, model, usage):
costs = self.token_costs[model]
input_cost = (usage.prompt_tokens / 1000) * costs[“input”]
output_cost = (usage.completion_tokens / 1000) * costs[“output”]
return input_cost + output_cost
`
6.4 综合循环保护
`python
class LoopProtector:
“””综合循环保护器，整合所有保护机制”””
def __init__(self, config=None):
config = config or {}
self.max_iterations = config.get(“max_iterations”, 100)
self.max_time_seconds = config.get(“max_time_seconds”, 3600)
self.max_cost = config.get(“max_cost”, 10.0)
self.max_errors = config.get(“max_errors”, 10)
self.max_consecutive_errors = config.get(“max_consecutive_errors”, 3)
self.iteration = 0
self.start_time = time.time()
self.cost = 0.0
self.errors = 0
self.consecutive_errors = 0
def check(self):
“””检查所有保护条件，任何一项超限都抛出异常”””
self.iteration += 1
# 迭代次数检查
if self.iteration > self.max_iterations:
raise LoopProtectionError(
f”Max iterations ({self.max_iterations}) exceeded”
)
# 时间检查
elapsed = time.time() – self.start_time
if elapsed > self.max_time_seconds:
raise LoopProtectionError(
f”Max time ({self.max_time_seconds}s) exceeded”
)
# 成本检查
if self.cost > self.max_cost:
raise LoopProtectionError(
f”Max cost ({self.max_cost}) exceeded”
)
# 错误次数检查
if self.errors > self.max_errors:
raise LoopProtectionError(
f”Max errors ({self.max_errors}) exceeded”
)
# 连续错误检查
if self.consecutive_errors > self.max_consecutive_errors:
raise LoopProtectionError(
f”Max consecutive errors ({self.max_consecutive_errors}) exceeded”
)
def record_success(self):
self.consecutive_errors = 0
def record_error(self):
self.errors += 1
self.consecutive_errors += 1
def add_cost(self, amount):
self.cost += amount
def get_status(self):
return {
“iteration”: self.iteration,
“elapsed_seconds”: time.time() – self.start_time,
“cost”: self.cost,
“errors”: self.errors,
“consecutive_errors”: self.consecutive_errors
}
`
七、实战：构建自愈 Agent Loop
现在，让我们把所有组件整合起来，构建一个完整的自愈 Agent Loop。
7.1 整体架构
`python
class SelfHealingAgent:
“””具有自我修复能力的 Agent”””
def __init__(self, config):
self.config = config
# 基础组件
self.task_queue = asyncio.Queue()
self.result_store = {}
# 检测组件
self.health_checker = HealthChecker()
self.anomaly_detector = AnomalyDetector()
self.heartbeat_monitor = HeartbeatMonitor()
# 保护组件
self.loop_protector = LoopProtector(config.get(“loop_protection”, {}))
self.cost_guard = CostGuard(config.get(“budget”, 10.0))
# 恢复组件
self.circuit_breakers = {}
self.retry_strategy = RetryStrategy(
max_retries=config.get(“max_retries”, 3)
)
self.restart_manager = RestartManager()
# 状态管理
self.state = “idle”
self.checkpoints = []
self._setup_health_checks()
def _setup_health_checks(self):
self.health_checker.register_check(
“memory”, self._check_memory, critical=True
)
self.health_checker.register_check(
“api_connectivity”, self._check_apis, critical=True
)
self.health_checker.register_check(
“task_progress”, self._check_progress, critical=False
)
async def run(self, task):
“””主运行循环”””
self.state = “running”
self.heartbeat_monitor.register_task(
task.id,
on_timeout=self._handle_task_timeout
)
try:
while not task.is_complete():
# 循环保护检查
self.loop_protector.check()
# 发送心跳
self.heartbeat_monitor.heartbeat(task.id)
# 健康检查
healthy, health_status = await self.health_checker.run_checks()
if not healthy:
await self._handle_unhealthy_state(health_status)
continue
# 执行步骤
try:
result = await self._execute_step(task)
task.update(result)
self.loop_protector.record_success()
except Exception as e:
self.loop_protector.record_error()
await self._handle_step_error(task, e)
# 保存检查点
if self.loop_protector.iteration % 10 == 0:
self._save_checkpoint(task)
await asyncio.sleep(0.1)
self.state = “completed”
return task.get_result()
except LoopProtectionError as e:
logger.error(f”Loop protection triggered: {e}”)
self.state = “protected_shutdown”
return await self._graceful_shutdown(task)
except Exception as e:
logger.error(f”Unexpected error: {e}”)
self.state = “error”
return await self._emergency_recovery(task, e)
async def _execute_step(self, task):
“””执行单个步骤，带完整错误处理”””
step = task.get_next_step()
# 获取或创建断路器
cb = self._get_circuit_breaker(step.service)
# 通过断路器执行
try:
result = await cb.call(
self._execute_with_retry,
step
)
return result
except CircuitOpenError:
# 使用降级策略
return await self._fallback_execute(step)
async def _execute_with_retry(self, step):
“””带重试的执行”””
return await with_retry(
step.execute,
self.retry_strategy
)
def _get_circuit_breaker(self, service_name):
if service_name not in self.circuit_breakers:
self.circuit_breakers[service_name] = CircuitBreaker(
name=service_name,
failure_threshold=5,
recovery_timeout=60
)
return self.circuit_breakers[service_name]
async def _fallback_execute(self, step):
“””降级执行策略”””
logger.warning(f”Using fallback for step: {step.name}”)
# 尝试备选方案
if hasattr(step, ‘fallback’):
return await step.fallback()
# 使用缓存结果
cached = await self._get_cached_result(step)
if cached:
return cached
# 跳过此步骤（如果允许）
if step.skippable:
logger.info(f”Skipping non-critical step: {step.name}”)
return None
raise FallbackFailedError(f”No fallback available for {step.name}”)
async def _handle_step_error(self, task, error):
“””处理步骤执行错误”””
logger.error(f”Step failed: {error}”)
# 分析错误类型
if isinstance(error, (ConnectionError, TimeoutError)):
# 网络问题，等待后重试
await asyncio.sleep(5)
elif isinstance(error, ValidationError):
# 数据问题，可能需要重新生成
task.invalidate_last_result()
elif isinstance(error, AuthenticationError):
# 认证问题，尝试刷新凭证
await self._refresh_credentials()
else:
# 未知错误，记录并继续
logger.error(f”Unknown error type: {type(error)}”)
async def _handle_unhealthy_state(self, health_status):
“””处理不健康状态”””
critical_failures = [
name for name, status in health_status.items()
if status.get(“status”) != “healthy” and status.get(“critical”)
]
if critical_failures:
logger.critical(f”Critical health check failures: {critical_failures}”)
# 尝试恢复
for failure in critical_failures:
await self._attempt_recovery(failure)
async def _attempt_recovery(self, component):
“””尝试恢复指定组件”””
logger.info(f”Attempting recovery for: {component}”)
recovery_actions = {
“memory”: self._recover_memory,
“api_connectivity”: self._recover_api,
“database”: self._recover_database,
}
action = recovery_actions.get(component)
if action:
try:
await action()
logger.info(f”Recovery successful for: {component}”)
except Exception as e:
logger.error(f”Recovery failed for {component}: {e}”)
async def _recover_memory(self):
“””内存恢复：清理缓存，触发GC”””
import gc
# 清理内部缓存
self.result_store.clear()
# 触发垃圾回收
gc.collect()
# 验证恢复
import psutil
if psutil.virtual_memory().percent > 90:
raise RecoveryFailedError(“Memory still critical after cleanup”)
async def _recover_api(self):
“””API恢复：重新建立连接”””
# 重置断路器
for cb in self.circuit_breakers.values():
cb.state = CircuitState.CLOSED
cb.stats = CircuitBreakerStats()
def _save_checkpoint(self, task):
“””保存检查点”””
checkpoint = {
“iteration”: self.loop_protector.iteration,
“task_state”: task.serialize(),
“agent_state”: {
“cost”: self.cost_guard.total_cost,
“errors”: self.loop_protector.errors
},
“timestamp”: time.time()
}
self.checkpoints.append(checkpoint)
# 只保留最近10个检查点
if len(self.checkpoints) > 10:
self.checkpoints = self.checkpoints[-10:]
async def _graceful_shutdown(self, task):
“””优雅关闭”””
logger.info(“Initiating graceful shutdown”)
# 保存当前状态
self._save_checkpoint(task)
# 等待进行中的操作完成
await asyncio.sleep(5)
# 返回部分结果
return {
“status”: “partial”,
“completed_steps”: task.completed_steps,
“checkpoint”: self.checkpoints[-1] if self.checkpoints else None,
“reason”: “loop_protection_triggered”
}
async def _emergency_recovery(self, task, error):
“””紧急恢复”””
logger.critical(f”Emergency recovery triggered: {error}”)
# 保存所有可能的状态
emergency_state = {
“task”: task.serialize(),
“error”: str(error),
“traceback”: traceback.format_exc(),
“timestamp”: time.time()
}
# 持久化紧急状态
await self._persist_emergency_state(emergency_state)
return {
“status”: “emergency_stopped”,
“error”: str(error),
“partial_results”: task.get_partial_results()
}
async def _persist_emergency_state(self, state):
“””持久化紧急状态，确保不会丢失”””
try:
with open(f”emergency_state_{int(time.time())}.json”, “w”) as f:
json.dump(state, f, indent=2, default=str)
except Exception as e:
logger.critical(f”Failed to persist emergency state: {e}”)
`
7.2 使用示例
`python
# 配置
config = {
“loop_protection”: {
“max_iterations”: 100,
“max_time_seconds”: 1800, # 30分钟
“max_cost”: 5.0,
“max_errors”: 10,
“max_consecutive_errors”: 3
},
“budget”: 5.0,
“max_retries”: 3
}
# 创建 Agent
agent = SelfHealingAgent(config)
# 定义任务
task = ResearchTask(
id=”research_001″,
query=”Latest developments in quantum computing”,
steps=[
SearchStep(query=”quantum computing 2024″, service=”search”),
AnalyzeStep(service=”llm”),
SynthesizeStep(service=”llm”),
FormatStep(service=”local”)
]
)
# 运行
async def main():
try:
result = await agent.run(task)
print(f”Task completed: {result}”)
except Exception as e:
print(f”Task failed: {e}”)
asyncio.run(main())
`
八、常见陷阱
在实现自我修复机制时，有一些常见的陷阱需要避免。
8.1 陷阱一：过度重试
问题：盲目重试可能导致”重试风暴”，让已经过载的系统更加不堪重负。
`python
# ❌ 错误做法
async def bad_retry():
while True:
try:
return await overloaded_service()
except Exception:
pass # 无限重试，没有延迟
# ✅ 正确做法
async def good_retry():
strategy = RetryStrategy(max_retries=3, base_delay=1.0)
return await with_retry(overloaded_service, strategy)
`
教训：重试必须有上限，并且要使用指数退避。对于限流错误，应该尊重 Retry-After 头部。
8.2 陷阱二：断路器状态不一致
问题：多个并发请求同时操作断路器状态，导致竞态条件。
`python
# ❌ 有问题的实现
class BadCircuitBreaker:
def __init__(self):
self.state = “closed”
self.failures = 0
async def call(self, func):
if self.state == “open”:
raise CircuitOpenError()
try:
result = await func()
self.failures = 0 # 可能与其他协程冲突
return result
except Exception:
self.failures += 1 # 可能出现竞态条件
if self.failures >= 5:
self.state = “open”
raise
# ✅ 正确做法：使用锁
class GoodCircuitBreaker:
def __init__(self):
self._lock = asyncio.Lock()
# … 其他代码使用 async with self._lock
`
8.3 陷阱三：忽略错误分类
问题：对所有错误都采用相同的处理策略。
`python
# ❌ 过于简单的处理
async def simple_handler():
try:
return await operation()
except Exception:
return await retry() # 对所有错误都重试
# ✅ 根据错误类型采取不同策略
async def smart_handler():
try:
return await operation()
except RateLimitError as e:
# 等待指定时间后重试
await asyncio.sleep(e.retry_after)
return await operation()
except ValidationError as e:
# 数据问题，不重试，直接失败
raise
except ConnectionError as e:
# 网络问题，重试
return await with_retry(operation, retry_strategy)
except AuthenticationError as e:
# 认证问题，刷新凭证后重试
await refresh_credentials()
return await operation()
`
8.4 陷阱四：检查点保存失败
问题：保存检查点本身可能失败，导致无法恢复。
`python
# ❌ 单点失败
def save_checkpoint(state):
with open(“checkpoint.json”, “w”) as f:
json.dump(state, f) # 如果写入失败，检查点丢失
# ✅ 多重保存
async def robust_save_checkpoint(state):
timestamp = int(time.time())
# 保存到多个位置
save_targets = [
f”checkpoint_{timestamp}.json”,
“checkpoint_latest.json”,
# 可以加上远程存储
]
saved = False
for target in save_targets:
try:
with open(target, “w”) as f:
json.dump(state, f, indent=2)
saved = True
except Exception as e:
logger.warning(f”Failed to save to {target}: {e}”)
if not saved:
logger.critical(“Failed to save checkpoint to any location!”)
# 最后的手段：写入stderr
sys.stderr.write(f”EMERGENCY CHECKPOINT: {json.dumps(state)}n”)
`
8.5 陷阱五：恢复过程中的级联失败
问题：恢复操作本身失败，导致更严重的问题。
`python
# ❌ 危险的恢复操作
async def dangerous_recovery():
# 先停止服务
await stop_service()
# 然后恢复数据——如果这一步失败呢？
await restore_data() # 服务已停止，数据可能不一致
# 最后重启
await start_service()
# ✅ 安全的恢复操作
async def safe_recovery():
# 创建恢复点
recovery_id = create_recovery_point()
try:
# 执行恢复
await stop_service()
await restore_data()
await start_service()
# 验证恢复成功
if not await verify_service_health():
raise RecoveryVerificationError()
except Exception as e:
logger.error(f”Recovery failed: {e}”)
# 回滚到恢复点
await rollback_to_recovery_point(recovery_id)
raise
`
8.6 陷阱六：日志和监控不足
问题：没有足够的日志，无法诊断问题根因。
`python
# ❌ 信息不足
async def poor_logging():
try:
result = await operation()
except Exception as e:
logger.error(f”Error: {e}”) # 没有上下文
# ✅ 详细的日志
async def good_logging():
context = {
“operation”: “fetch_data”,
“attempt”: attempt,
“timestamp”: time.time(),
“correlation_id”: get_correlation_id()
}
try:
result = await operation()
logger.info(“Operation succeeded”, extra=context)
return result
except Exception as e:
logger.error(
f”Operation failed: {e}”,
extra={
**context,
“error_type”: type(e).__name__,
“traceback”: traceback.format_exc()
}
)
raise
`
8.7 陷阱七：忽略优雅降级
问题：系统要么完美运行，要么完全崩溃，没有中间状态。
`python
# ❌ 全有或全无
async def all_or_nothing():
result_a = await critical_operation_a()
result_b = await critical_operation_b()
result_c = await optional_operation_c()
return combine(result_a, result_b, result_c)
# ✅ 支持优雅降级
async def graceful_degradation():
result = {}
# 关键操作
try:
result[“a”] = await critical_operation_a()
except Exception as e:
logger.critical(f”Critical operation A failed: {e}”)
raise # 关键操作失败必须抛出
try:
result[“b”] = await critical_operation_b()
except Exception as e:
logger.critical(f”Critical operation B failed: {e}”)
raise
# 可选操作
try:
result[“c”] = await optional_operation_c()
except Exception as e:
logger.warning(f”Optional operation C failed, continuing: {e}”)
result[“c”] = None # 降级处理
return result
`
九、总结
构建自我修复的 Agent Loop 是一项系统工程，需要从多个层面进行防护：
核心原则
- 防御性编程：假设任何外部调用都可能失败
- 快速失败：问题应该尽早暴露，而不是隐藏到无法挽回
- 优雅降级：部分功能丧失不应该导致系统完全瘫痪
- 可观测性：没有监控和日志，故障就是黑箱
关键组件
| 组件 | 作用 | 关键指标 |
|---|
|——|——|———-|
| 健康检查 | 主动发现问题 | 检查频率、响应时间 |
|---|---|---|
| 断路器 | 防止故障蔓延 | 失败阈值、恢复时间 |
| 重试机制 | 应对临时故障 | 重试次数、退避策略 |
| 循环保护 | 防止失控 | 迭代次数、时间、成本 |
| 检查点 | 支持恢复 | 保存频率、保存位置 |
实施建议
- 从小处开始：先实现基本的重试和超时，再逐步添加更复杂的机制
- 测试故障场景：使用混沌工程方法，主动注入故障验证恢复机制
- 监控一切：每个断路器的状态、每次重试、每个检查点都应该有记录
- 定期演练：恢复机制如果不定期测试，关键时刻可能会失败
最后的思考
自我修复不是一次性的工作，而是持续改进的过程。随着你对系统的理解加深，随着新的故障模式的发现，你的自愈机制也需要不断进化。
最好的自愈系统是这样的：它在正常运行时几乎不被察觉，但在异常发生时能平稳地处理问题，让用户甚至不知道曾经出过故障。
这就是 Loop Engineering 的终极目标——让 Agent 在任何情况下都能可靠地完成任务，即使世界并不完美。
系列导航：第一篇：什么是Loop | 第二篇：设计你的第一个Loop | 第三篇：反馈循环设计 | 第四篇：自我修复Loop | 第五篇：生产环境Loop | 第六篇：完整实战
下一篇预告：第五篇：生产环境Loop —— 从玩具到生产，如何让你的Loop在真实环境中稳定运行。
发表回复