# core/circuit_breaker.py
from enum import Enum
from datetime import datetime, timedelta
import asyncio
class CircuitState(Enum):
CLOSED = "closed" # 正常：请求通过
OPEN = "open" # 故障：快速失败
HALF_OPEN = "half_open" # 恢复中：探测性请求
class CircuitBreakerConfig:
"""断路器参数配置"""
def __init__(
self,
failure_threshold: int = 5,
success_threshold: int = 2,
timeout_seconds: int = 60,
expected_exception: type = Exception
):
self.failure_threshold = failure_threshold
self.success_threshold = success_threshold
self.timeout_seconds = timeout_seconds
self.expected_exception = expected_exception
class CircuitBreaker:
"""断路器实现(CLOSED ↔ HALF_OPEN ↔ OPEN)"""
def __init__(self, name: str, config: CircuitBreakerConfig):
self.name = name
self.config = config
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time = None
async def call(self, func, *args, **kwargs):
"""通过断路器调用函数"""
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self.state = CircuitState.HALF_OPEN
self.success_count = 0
print(f"Circuit breaker {self.name} entering HALF_OPEN")
else:
raise Exception(f"Circuit breaker {self.name} is OPEN")
try:
result = await func(*args, **kwargs)
self._on_success()
return result
except self.config.expected_exception as e:
self._on_failure()
raise
def _on_success(self):
"""处理成功调用"""
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
self.state = CircuitState.CLOSED
self.success_count = 0
print(f"Circuit breaker {self.name} closed (recovered)")
def _on_failure(self):
"""处理失败调用"""
self.failure_count += 1
self.last_failure_time = datetime.now()
if self.state == CircuitState.CLOSED:
if self.failure_count >= self.config.failure_threshold:
self.state = CircuitState.OPEN
print(f"Circuit breaker {self.name} opened")
elif self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
print(f"Circuit breaker {self.name} reopened")
def _should_attempt_reset(self) -> bool:
"""检查是否应该尝试从 OPEN 转为 HALF_OPEN"""
if self.last_failure_time is None:
return True
elapsed = (datetime.now() - self.last_failure_time).total_seconds()
return elapsed >= self.config.timeout_seconds
def get_state(self) -> dict:
"""获取断路器状态信息"""
return {
'name': self.name,
'state': self.state.value,
'failure_count': self.failure_count,
'success_count': self.success_count,
'last_failure_time': self.last_failure_time
}# core/bulkhead.py
"""
隔舱模式：为不同的操作分配独立的资源池
防止一个故障影响其他操作
"""
import asyncio
from asyncio import Semaphore
from typing import Dict
import logging
logger = logging.getLogger(__name__)
class BulkheadConfig:
"""隔舱配置"""
def __init__(self, max_concurrent: int = 10, timeout: int = 30):
self.max_concurrent = max_concurrent
self.timeout = timeout
class Bulkhead:
"""隔舱：限制并发数"""
def __init__(self, name: str, config: BulkheadConfig):
self.name = name
self.config = config
self.semaphore = Semaphore(config.max_concurrent)
self.active_tasks = 0
async def execute(self, func, *args, **kwargs):
"""在隔舱内执行任务"""
async with self.semaphore:
self.active_tasks += 1
logger.info(
f"Bulkhead {self.name}: "
f"{self.active_tasks}/{self.config.max_concurrent} active"
)
try:
return await asyncio.wait_for(
func(*args, **kwargs),
timeout=self.config.timeout
)
finally:
self.active_tasks -= 1
class BulkheadManager:
"""隔舱管理器"""
def __init__(self):
self.bulkheads: Dict[str, Bulkhead] = {}
def create_bulkhead(self, name: str, config: BulkheadConfig) -> Bulkhead:
"""创建隔舱"""
bulkhead = Bulkhead(name, config)
self.bulkheads[name] = bulkhead
return bulkhead
async def execute(self, bulkhead_name: str, func, *args, **kwargs):
"""通过指定隔舱执行任务"""
if bulkhead_name not in self.bulkheads:
raise ValueError(f"Bulkhead {bulkhead_name} not found")
return await self.bulkheads[bulkhead_name].execute(func, *args, **kwargs)
def get_stats(self) -> dict:
"""获取所有隔舱的统计"""
return {
name: {
'max_concurrent': bulkhead.config.max_concurrent,
'active_tasks': bulkhead.active_tasks
}
for name, bulkhead in self.bulkheads.items()
}# core/timeout.py
"""
超时模式：设定最长等待时间,避免无限阻塞
"""
import asyncio
from contextlib import asynccontextmanager
class TimeoutManager:
"""超时管理"""
@staticmethod
@asynccontextmanager
async def timeout(seconds: int, operation_name: str = "operation"):
"""
超时上下文管理器
使用示例：
async with TimeoutManager.timeout(30, "database_query") as tm:
result = await db.query(...)
"""
try:
async with asyncio.timeout(seconds):
yield
except asyncio.TimeoutError:
raise TimeoutError(
f"Operation {operation_name} exceeded {seconds}s timeout"
)
@staticmethod
async def with_timeout(func, timeout_seconds: int, *args, **kwargs):
"""为函数调用添加超时"""
try:
return await asyncio.wait_for(
func(*args, **kwargs),
timeout=timeout_seconds
)
except asyncio.TimeoutError:
raise TimeoutError(
f"Function {func.__name__} exceeded {timeout_seconds}s timeout"
)
class AdaptiveTimeout:
"""自适应超时(根据历史数据调整)"""
def __init__(self, initial_timeout: int = 30):
self.initial_timeout = initial_timeout
self.execution_times = []
self.max_samples = 100
def record(self, duration: float):
"""记录执行时间"""
self.execution_times.append(duration)
if len(self.execution_times) > self.max_samples:
self.execution_times.pop(0)
def get_timeout(self) -> int:
"""获取建议的超时时间"""
if not self.execution_times:
return self.initial_timeout
# 使用 P95 作为超时时间[1]
sorted_times = sorted(self.execution_times)
p95_index = int(len(sorted_times) * 0.95)
p95_time = sorted_times[p95_index]
# 添加 20% 的缓冲
return max(self.initial_timeout, int(p95_time * 1.2))# core/retry.py
import asyncio
import random
from typing import Callable
import logging
logger = logging.getLogger(__name__)
class RetryConfig:
"""重试配置参数"""
def __init__(
self,
max_attempts: int = 3,
initial_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2.0,
jitter: bool = True,
retryable_exceptions: tuple = (Exception,)
):
self.max_attempts = max_attempts
self.initial_delay = initial_delay
self.max_delay = max_delay
self.exponential_base = exponential_base
self.jitter = jitter
self.retryable_exceptions = retryable_exceptions
class RetryPolicy:
"""重试执行策略(指数退避 + 抖动)"""
def __init__(self, config: RetryConfig):
self.config = config
self.attempt_count = 0
async def execute(self, func: Callable, *args, **kwargs):
"""执行函数,失败时自动重试"""
last_exception = None
for attempt in range(1, self.config.max_attempts + 1):
try:
result = await func(*args, **kwargs)
logger.info(f"Succeeded on attempt {attempt}")
return result
except self.config.retryable_exceptions as e:
last_exception = e
self.attempt_count = attempt
if attempt == self.config.max_attempts:
logger.error(f"Failed after {attempt} attempts: {e}")
raise
# 计算延迟：指数退避 + 抖动
delay = self._calculate_delay(attempt)
logger.warning(f"Attempt {attempt} failed: {e}. Retrying in {delay:.1f}s...")
await asyncio.sleep(delay)
raise last_exception
def _calculate_delay(self, attempt: int) -> float:
"""计算延迟时间：指数退避 + 10% 随机抖动"""
delay = self.config.initial_delay * (self.config.exponential_base ** (attempt - 1))
delay = min(delay, self.config.max_delay)
if self.config.jitter:
jitter = random.uniform(0, delay * 0.1)
delay += jitter
return delay
class RetryDecorator:
"""重试装饰器"""
@staticmethod
def retry(config: RetryConfig = None):
"""为函数添加重试能力的装饰器"""
if config is None:
config = RetryConfig()
def decorator(func):
async def wrapper(*args, **kwargs):
policy = RetryPolicy(config)
return await policy.execute(func, *args, **kwargs)
return wrapper
return decorator# core/idempotency.py
import hashlib
import json
from typing import Dict, Any, Optional
from datetime import datetime, timedelta
import asyncio
class IdempotencyManager:
"""幂等性管理：重复操作返回缓存结果"""
def __init__(self, ttl_seconds: int = 3600):
self.cache: Dict[str, Any] = {}
self.metadata: Dict[str, Dict] = {}
self.ttl = ttl_seconds
@staticmethod
def generate_key(operation: str, params: dict) -> str:
"""生成幂等性 Key(操作 + 参数的 SHA256 哈希)"""
key_data = {'operation': operation, 'params': params}
key_string = json.dumps(key_data, sort_keys=True)
return hashlib.sha256(key_string.encode()).hexdigest()
async def execute_idempotent(
self,
idempotency_key: str,
operation: str,
func,
*args,
**kwargs
) -> Any:
"""执行幂等操作(缓存相同操作的结果)"""
# 检查缓存
if idempotency_key in self.cache:
metadata = self.metadata[idempotency_key]
if metadata['status'] == 'completed':
return self.cache[idempotency_key]
elif metadata['status'] == 'in_progress':
# 等待进行中的操作完成
await asyncio.sleep(0.1)
return await self.execute_idempotent(
idempotency_key, operation, func, *args, **kwargs
)
# 标记为进行中
self.metadata[idempotency_key] = {
'status': 'in_progress',
'operation': operation,
'started_at': datetime.now()
}
try:
result = await func(*args, **kwargs)
# 缓存成功结果
self.cache[idempotency_key] = result
self.metadata[idempotency_key] = {
'status': 'completed',
'operation': operation,
'completed_at': datetime.now()
}
return result
except Exception as e:
# 记录失败状态
self.metadata[idempotency_key] = {
'status': 'failed',
'operation': operation,
'error': str(e),
'failed_at': datetime.now()
}
raise
def cleanup_expired(self):
"""清理超过 TTL 的缓存条目"""
now = datetime.now()
expired_keys = []
for key, metadata in self.metadata.items():
timestamp = metadata.get('completed_at') or metadata.get('started_at')
if timestamp and (now - timestamp).total_seconds() > self.ttl:
expired_keys.append(key)
for key in expired_keys:
del self.cache[key]
del self.metadata[key]# core/checkpoint.py
from datetime import datetime
from typing import Dict, Any, Optional
import uuid
class Checkpoint:
"""检查点：记录特定步骤的状态"""
def __init__(self, step_id: str, state: dict, timestamp: datetime):
self.step_id = step_id
self.state = state
self.timestamp = timestamp
class CheckpointManager:
"""检查点管理器：保存和恢复工作流状态"""
def __init__(self):
self.checkpoints: Dict[str, list] = {}
self.current_checkpoint: Dict[str, int] = {}
def create_checkpoint(self, workflow_id: str, step_id: str, state: dict) -> Checkpoint:
"""创建新检查点"""
checkpoint = Checkpoint(step_id=step_id, state=state, timestamp=datetime.now())
if workflow_id not in self.checkpoints:
self.checkpoints[workflow_id] = []
self.checkpoints[workflow_id].append(checkpoint)
return checkpoint
def get_last_checkpoint(self, workflow_id: str) -> Optional[Checkpoint]:
"""获取最后一个检查点"""
if workflow_id not in self.checkpoints:
return None
checkpoints = self.checkpoints[workflow_id]
return checkpoints[-1] if checkpoints else None
def restore_from_checkpoint(self, workflow_id: str) -> Optional[dict]:
"""从最后一个检查点恢复状态"""
last_checkpoint = self.get_last_checkpoint(workflow_id)
if last_checkpoint:
return {
'step_id': last_checkpoint.step_id,
'state': last_checkpoint.state,
'resumed_at': datetime.now()
}
return None
def rollback_to_checkpoint(self, workflow_id: str, steps: int = 1) -> Optional[dict]:
"""回滚到前面的检查点"""
if workflow_id not in self.checkpoints:
return None
checkpoints = self.checkpoints[workflow_id]
target_index = len(checkpoints) - steps - 1
if target_index < 0:
return None
target_checkpoint = checkpoints[target_index]
return {
'step_id': target_checkpoint.step_id,
'state': target_checkpoint.state,
'rolled_back_at': datetime.now()
}# 可恢复的工作流
class RecoverableWorkflow:
"""长流程工作流(支持故障恢复)"""
def __init__(self, workflow_id: str):
self.workflow_id = workflow_id
self.checkpoint_manager = CheckpointManager()
self.steps = []
async def add_step(self, step_id: str, func, *args, **kwargs):
"""添加工作流步骤"""
self.steps.append({
'id': step_id,
'func': func,
'args': args,
'kwargs': kwargs
})
async def execute(self, resume: bool = False) -> dict:
"""执行工作流,支持恢复"""
start_step = 0
if resume:
last_checkpoint = self.checkpoint_manager.get_last_checkpoint(self.workflow_id)
if last_checkpoint:
# 找到恢复点后的第一个步骤
for i, step in enumerate(self.steps):
if step['id'] == last_checkpoint.step_id:
start_step = i + 1
break
results = []
for i in range(start_step, len(self.steps)):
step = self.steps[i]
try:
result = await step['func'](*step['args'], **step['kwargs'])
results.append(result)
# 保存检查点
self.checkpoint_manager.create_checkpoint(
self.workflow_id,
step['id'],
{'results': results}
)
except Exception as e:
return {
'status': 'failed',
'error': str(e),
'failed_step': step['id'],
'can_resume': True,
'last_checkpoint': step['id']
}
return {'status': 'completed', 'results': results}# core/error_as_observation.py
from dataclasses import dataclass
from enum import Enum
from datetime import datetime
class ErrorObservationType(Enum):
"""错误类型分类"""
TOOL_ERROR = "tool_error"
TIMEOUT = "timeout"
RATE_LIMIT = "rate_limit"
VALIDATION_ERROR = "validation_error"
NETWORK_ERROR = "network_error"
@dataclass
class ErrorObservation:
"""错误观察对象"""
type: ErrorObservationType
source: str
message: str
error_code: str
timestamp: datetime
severity: int # 1-5
recoverable: bool
suggested_action: str
retry_after: int = 0
class ErrorAsObservationHandler:
"""错误处理器：将异常转化为观察"""
def __init__(self):
self.observations = []
async def handle_tool_error(
self,
tool_name: str,
error: Exception,
context: dict
) -> ErrorObservation:
"""将工具错误转化为观察(而不是抛异常)"""
observation = self._classify_error(tool_name, error, context)
self.observations.append(observation)
return observation
def _classify_error(
self,
tool_name: str,
error: Exception,
context: dict
) -> ErrorObservation:
"""根据异常类型分类错误"""
error_type = type(error).__name__
if 'timeout' in error_type.lower():
return ErrorObservation(
type=ErrorObservationType.TIMEOUT,
source=tool_name,
message=str(error),
error_code='TIMEOUT',
timestamp=datetime.now(),
severity=3,
recoverable=True,
suggested_action='retry',
retry_after=5
)
elif 'rate' in error_type.lower():
return ErrorObservation(
type=ErrorObservationType.RATE_LIMIT,
source=tool_name,
message=f"Rate limit exceeded for {tool_name}",
error_code='RATE_LIMIT',
timestamp=datetime.now(),
severity=2,
recoverable=True,
suggested_action='backoff_and_retry',
retry_after=60
)
else:
return ErrorObservation(
type=ErrorObservationType.TOOL_ERROR,
source=tool_name,
message=str(error),
error_code=error_type,
timestamp=datetime.now(),
severity=4,
recoverable=False,
suggested_action='report_to_user'
)
async def inject_observations(self, agent_context: dict) -> dict:
"""将观察注入到 Agent 上下文"""
observations_list = [
{
'type': obs.type.value,
'source': obs.source,
'message': obs.message,
'recoverable': obs.recoverable,
'suggested_action': obs.suggested_action,
'retry_after': obs.retry_after
}
for obs in self.observations
]
agent_context['observations'] = agent_context.get('observations', []) + observations_list
return agent_context
def get_summary(self) -> dict:
"""获取错误观察摘要"""
by_type = {}
for obs in self.observations:
obs_type = obs.type.value
by_type[obs_type] = by_type.get(obs_type, 0) + 1
return {
'total_observations': len(self.observations),
'by_type': by_type,
'unrecoverable_count': sum(1 for o in self.observations if not o.recoverable)
}# examples/fault_tolerance_system.py
"""
完整的容错系统示例
"""
import asyncio
class ResilientAgent:
"""具有容错能力的 Agent"""
def __init__(self):
self.circuit_breaker = CircuitBreaker(
"api_calls",
CircuitBreakerConfig(failure_threshold=5, timeout_seconds=30)
)
self.bulkhead_manager = BulkheadManager()
self.retry_config = RetryConfig(
max_attempts=3,
initial_delay=1.0,
exponential_base=2.0
)
self.idempotency_manager = IdempotencyManager()
self.error_handler = ErrorAsObservationHandler()
# 创建隔舱
self.bulkhead_manager.create_bulkhead(
"external_api",
BulkheadConfig(max_concurrent=10, timeout=30)
)
async def execute_action(
self,
action_id: str,
tool_name: str,
params: dict
) -> dict:
"""执行带容错的操作"""
# 步骤 1:生成幂等性 Key
idempotency_key = IdempotencyManager.generate_key(tool_name, params)
# 步骤 2:通过幂等性管理器执行
async def execute_with_faults():
# 步骤 3:通过隔舱限制并发
async def isolated_call():
# 步骤 4:通过重试策略执行
policy = RetryPolicy(self.retry_config)
return await policy.execute(
self._call_tool,
tool_name, params
)
return await self.bulkhead_manager.execute(
"external_api",
isolated_call
)
try:
result = await self.idempotency_manager.execute_idempotent(
idempotency_key,
tool_name,
execute_with_faults
)
return {
'status': 'success',
'result': result,
'action_id': action_id
}
except Exception as e:
# 转化为观察而非异常
error_obs = await self.error_handler.handle_tool_error(
tool_name, e, {'action_id': action_id}
)
return {
'status': 'error',
'observation': {
'type': error_obs.type.value,
'message': error_obs.message,
'recoverable': error_obs.recoverable,
'suggested_action': error_obs.suggested_action
},
'action_id': action_id
}
async def _call_tool(self, tool_name: str, params: dict):
"""实际调用工具(模拟)"""
# 模拟随机失败
import random
if random.random() < 0.3:
raise TimeoutError(f"Tool {tool_name} timeout")
return f"Result from {tool_name}"
# 测试
async def main():
agent = ResilientAgent()
# 执行多个操作
tasks = []
for i in range(5):
task = agent.execute_action(
action_id=f"action_{i}",
tool_name="external_api",
params={"query": f"request_{i}"}
)
tasks.append(task)
results = await asyncio.gather(*tasks)
print("Results:")
for result in results:
print(f" {result['action_id']}: {result['status']}")
print("\nError observations:")
summary = agent.error_handler.get_summary()
print(f" Total: {summary['total_observations']}")
print(f" By type: {summary['by_type']}")
if __name__ == "__main__":
asyncio.run(main())