AI Agent 弹性工程与错误处理指南
MSG Chain 智能体系统韧性架构手册
⚠️ No-Go Disclaimer: MSGChain 主网裁决为 No-Go。本文件所有内容反映的是开发阶段的技术设计，不代表主网未独立核验上线状态。生产部署状态请以白皮书为准：https://msgchain.org/whitepaper/
适用链: MSG Chain (msg-chain-1)
地址前缀:msg
版本: v1.0.0
1. 概述
1.1 为什么弹性对自治智能体至关重要
AI Agent 在 MSG Chain 上自主执行链上操作时，面临多重失效风险。不同于传统 Web 应用——用户可以刷新页面重试——自治智能体必须在无人工干预的情况下检测、诊断并从故障中恢复。弹性工程（Resilience Engineering）确保智能体在面对网络分区、节点故障、Gas 不足、LLM 服务中断等异常时，仍能维持可预测的行为。
本指南定义的弹性框架适用于以下场景：
- DeFi 策略执行: 智能体在多个 DEX 之间执行套利策略，需要容忍单个 RPC 节点故障。
- 跨链资产桥接: Agent 发起从 MSG Chain 到其他链的 IBC 转移，需要处理中继超时。
- 自主 NFT 铸造与管理: 在 Gas 拍卖期间，智能体需要自适应地调整 Gas 价格。
- DAO 治理提案: Agent 自动提交提案并投票，需要处理交易回滚和状态不一致。
1.2 主要失效模式
| 失效模式 | 来源 | 典型表现 | 可重试性 |
|---|---|---|---|
| 链错误 | MSG Chain 节点 | 交易回滚、序列号不匹配 | ✅ 部分可重试 |
| 网络错误 | RPC 端点、IBC 中继 | 连接超时、DNS 解析失败 | ✅ 可重试 |
| LLM 错误 | AI 推理服务 | Token 限制、响应格式异常 | ⚠️ 谨慎重试 |
| 合约错误 | 智能合约 | Revert、Gas 耗尽、断言失败 | ❌ 不可重试 |
| 支付错误 | 通证转账 | 余额不足、滑点超出 | ⚠️ 部分可重试 |
1.3 错误分类体系
from enum import Enum, auto
import time
import asyncio
import random
import logging
from typing import (
Optional, Callable, Awaitable, Any, Dict, List,
TypeVar, Generic, Set, Tuple, Union
)
from dataclasses import dataclass, field
from abc import ABC, abstractmethod
from contextlib import asynccontextmanager
from collections import defaultdict
import json
import hashlib
import hmac
import traceback
import os
import sys
import uuid
logger = logging.getLogger("msg_agent.resilience")
class Severity(Enum):
INFO = 0
WARNING = 1
ERROR = 2
CRITICAL = 3
class ErrorCategory(Enum):
TRANSITORY = "transitory" # 瞬时错误，重试后通常可恢复
PERSISTENT = "persistent" # 持久错误，重试无意义
INDETERMINATE = "indeterminate" # 结果不确定（交易可能已提交）
FATAL = "fatal" # 致命错误，需要人工介入
class RetryPolicy(Enum):
NO_RETRY = "no_retry" # 不重试
LIMITED_RETRY = "limited_retry" # 有限次数重试
INFINITE_RETRY = "infinite_retry" # 无限重试（直到成功或取消）
DEDUP_RETRY = "dedup_retry" # 带幂等性检查的重试
1.4 弹性设计原则
- 优雅降级: 当某个依赖服务不可用时，智能体不应整体崩溃，而应降级到功能子集。
- 失败隔离: 一个 Agent 实例的失败不应级联影响其他实例。
- 快速失败: 识别到不可恢复的错误时应立即失败，而非耗尽资源重试。
- 幂等性: 所有链上操作必须具有幂等性，确保重试不会产生副作用。
- 可观测性: 所有错误、重试、熔断事件必须结构化成日志和指标。
2. 错误类型体系
2.1 基础错误类
class AgentError(Exception):
"""所有 Agent 错误的基类"""
def __init__(
self,
code: str,
message: str = "",
severity: Severity = Severity.ERROR,
retryable: bool = False,
category: ErrorCategory = ErrorCategory.PERSISTENT,
cause: Optional[Exception] = None,
):
self.code = code
self.severity = severity
self.retryable = retryable
self.category = category
self.timestamp = time.time()
self.cause = cause
self.trace = traceback.format_exc()
message = message or f"AgentError[{code}]: {severity.name}"
super().__init__(message)
def to_dict(self) -> Dict[str, Any]:
return {
"code": self.code,
"message": str(self),
"severity": self.severity.name,
"retryable": self.retryable,
"category": self.category.value,
"timestamp": self.timestamp,
"cause": str(self.cause) if self.cause else None,
}
def should_retry(self) -> bool:
return self.retryable and self.category != ErrorCategory.FATAL
class RetryableError(AgentError):
"""可重试错误的标记接口"""
def __init__(
self,
code: str,
message: str = "",
severity: Severity = Severity.WARNING,
category: ErrorCategory = ErrorCategory.TRANSITORY,
cause: Optional[Exception] = None,
):
super().__init__(
code=code,
message=message,
severity=severity,
retryable=True,
category=category,
cause=cause,
)
class NonRetryableError(AgentError):
"""不可重试错误的标记接口"""
def __init__(
self,
code: str,
message: str = "",
severity: Severity = Severity.ERROR,
category: ErrorCategory = ErrorCategory.PERSISTENT,
cause: Optional[Exception] = None,
):
super().__init__(
code=code,
message=message,
severity=severity,
retryable=False,
category=category,
cause=cause,
)
2.2 链错误
class ChainError(RetryableError):
"""与 MSG Chain 节点通信时发生的错误"""
def __init__(
self,
code: str,
message: str = "",
tx_hash: Optional[str] = None,
gas_used: Optional[int] = None,
cause: Optional[Exception] = None,
):
self.tx_hash = tx_hash
self.gas_used = gas_used
super().__init__(
code=code,
message=message,
severity=Severity.ERROR,
category=ErrorCategory.TRANSITORY,
cause=cause,
)
class TxRevertError(ChainError):
"""交易回滚错误"""
def __init__(
self,
tx_hash: str,
revert_reason: str,
gas_used: int,
contract_address: Optional[str] = None,
):
self.revert_reason = revert_reason
self.contract_address = contract_address
super().__init__(
code="TX_REVERT",
message=f"Transaction {tx_hash} reverted: {revert_reason}",
tx_hash=tx_hash,
gas_used=gas_used,
)
# 合约 revert 通常不可重试（相同输入会再次失败）
self.retryable = False
self.category = ErrorCategory.PERSISTENT
class TxTimeoutError(ChainError):
"""交易超时（交易已广播但未在预期区块数内确认）"""
def __init__(
self,
tx_hash: str,
max_blocks: int,
current_height: int,
):
self.max_blocks = max_blocks
self.current_height = current_height
super().__init__(
code="TX_TIMEOUT",
message=f"Tx {tx_hash} not confirmed after {max_blocks} blocks",
tx_hash=tx_hash,
)
self.category = ErrorCategory.INDETERMINATE
class TxSequenceError(ChainError):
"""序列号（Nonce）错误——客户端序列号与链上状态不匹配"""
def __init__(self, expected: int, actual: int, address: str):
self.expected = expected
self.actual = actual
self.address = address
super().__init__(
code="TX_SEQUENCE",
message=f"Nonce mismatch for {address}: expected {expected}, got {actual}",
)
class GasEstimationError(ChainError):
"""Gas 估算失败"""
def __init__(self, tx_hash: str, gas_wanted: int, gas_used: int):
self.gas_wanted = gas_wanted
self.gas_used = gas_used
super().__init__(
code="GAS_ESTIMATION",
message=f"Gas underestimation: wanted {gas_wanted}, used {gas_used}",
tx_hash=tx_hash,
gas_used=gas_used,
)
class OutOfGasError(ChainError):
"""Gas 耗尽"""
def __init__(self, tx_hash: str, gas_limit: int):
self.gas_limit = gas_limit
super().__init__(
code="OUT_OF_GAS",
message=f"Out of gas at limit {gas_limit}",
tx_hash=tx_hash,
gas_used=gas_limit,
)
class InsufficientFundsError(ChainError):
"""余额不足"""
def __init__(self, address: str, required: str, available: str, denom: str):
self.address = address
self.required = required
self.available = available
self.denom = denom
super().__init__(
code="INSUFFICIENT_FUNDS",
message=f"Address {address} needs {required} {denom}, has {available} {denom}",
)
class AccountNotFoundError(ChainError):
"""链上未找到账户"""
def __init__(self, address: str):
self.address = address
super().__init__(
code="ACCOUNT_NOT_FOUND",
message=f"Account {address} does not exist on MSG Chain",
)
class IBCTransferError(ChainError):
"""IBC 跨链转账错误"""
def __init__(
self,
source_channel: str,
destination_chain: str,
packet_sequence: Optional[int] = None,
cause: Optional[Exception] = None,
):
self.source_channel = source_channel
self.destination_chain = destination_chain
self.packet_sequence = packet_sequence
super().__init__(
code="IBC_TRANSFER",
message=f"IBC transfer failed on channel {source_channel} to {destination_chain}",
cause=cause,
)
2.3 合约错误
class ContractError(NonRetryableError):
"""与智能合约交互时发生的错误"""
def __init__(
self,
code: str,
message: str = "",
contract_address: Optional[str] = None,
method: Optional[str] = None,
args: Optional[Dict[str, Any]] = None,
cause: Optional[Exception] = None,
):
self.contract_address = contract_address
self.method = method
self.args = args
super().__init__(
code=code,
message=message,
severity=Severity.ERROR,
category=ErrorCategory.PERSISTENT,
cause=cause,
)
class ContractRevertError(ContractError):
"""合约执行 revert"""
def __init__(
self,
contract_address: str,
method: str,
revert_message: str,
args: Optional[Dict[str, Any]] = None,
):
self.revert_message = revert_message
super().__init__(
code="CONTRACT_REVERT",
message=f"Contract {contract_address}.{method} reverted: {revert_message}",
contract_address=contract_address,
method=method,
args=args,
)
class ContractNotFoundError(ContractError):
"""链上未找到合约"""
def __init__(self, contract_address: str):
super().__init__(
code="CONTRACT_NOT_FOUND",
message=f"No contract at address {contract_address}",
contract_address=contract_address,
)
class ContractExecutionError(ContractError):
"""合约执行异常（非 revert 的其他错误）"""
def __init__(
self,
contract_address: str,
method: str,
output: str,
args: Optional[Dict[str, Any]] = None,
):
super().__init__(
code="CONTRACT_EXECUTION",
message=f"Contract {contract_address}.{method} execution error: {output}",
contract_address=contract_address,
method=method,
args=args,
)
class SlippageError(ContractError):
"""滑点超出限制"""
def __init__(
self,
expected_min: str,
actual: str,
denom: str,
pool_id: Optional[str] = None,
):
self.expected_min = expected_min
self.actual = actual
self.denom = denom
self.pool_id = pool_id
super().__init__(
code="SLIPPAGE_EXCEEDED",
message=f"Expected min {expected_min} {denom}, got {actual} {denom}",
)
class ApprovalError(ContractError):
"""通证授权失败"""
def __init__(
self,
spender: str,
token_address: str,
amount: str,
reason: str = "",
):
self.spender = spender
self.token_address = token_address
self.amount = amount
super().__init__(
code="APPROVAL_FAILED",
message=f"Failed to approve {amount} for {spender} on {token_address}: {reason}",
contract_address=token_address,
method="approve",
)
2.4 LLM 错误
class LLMError(RetryableError):
"""AI 推理服务错误"""
def __init__(
self,
code: str,
message: str = "",
provider: Optional[str] = None,
model: Optional[str] = None,
input_tokens: Optional[int] = None,
cause: Optional[Exception] = None,
):
self.provider = provider
self.model = model
self.input_tokens = input_tokens
super().__init__(
code=code,
message=message,
severity=Severity.WARNING,
category=ErrorCategory.TRANSITORY,
cause=cause,
)
def to_dict(self) -> Dict[str, Any]:
base = super().to_dict()
base.update({
"provider": self.provider,
"model": self.model,
"input_tokens": self.input_tokens,
})
return base
class LLMRateLimitError(LLMError):
"""LLM API 速率限制"""
def __init__(
self,
provider: str,
retry_after: float = 1.0,
model: Optional[str] = None,
):
self.retry_after = retry_after
super().__init__(
code="LLM_RATE_LIMIT",
message=f"Rate limited by {provider}, retry after {retry_after}s",
provider=provider,
model=model,
)
class LLMTimeoutError(LLMError):
"""LLM 推理超时"""
def __init__(self, provider: str, timeout: float, model: Optional[str] = None):
self.timeout = timeout
super().__init__(
code="LLM_TIMEOUT",
message=f"LLM {provider}/{model} timed out after {timeout}s",
provider=provider,
model=model,
)
class LLMResponseParseError(LLMError):
"""LLM 响应解析失败（输出格式不符合预期）"""
def __init__(
self,
raw_response: str,
expected_format: str,
provider: Optional[str] = None,
model: Optional[str] = None,
):
self.raw_response = raw_response[:500]
self.expected_format = expected_format
super().__init__(
code="LLM_PARSE_ERROR",
message=f"Failed to parse LLM response (expected {expected_format})",
provider=provider,
model=model,
)
class LLMContentFilterError(LLMError):
"""LLM 内容过滤器触发"""
def __init__(
self,
provider: str,
reason: str = "",
model: Optional[str] = None,
):
self.reason = reason
super().__init__(
code="LLM_CONTENT_FILTER",
message=f"Content filter triggered by {provider}: {reason}",
provider=provider,
model=model,
severity=Severity.ERROR,
category=ErrorCategory.PERSISTENT,
)
# 内容过滤错误通常不可重试（需要修改提示词）
self.retryable = False
class LLMTokenLimitError(LLMError):
"""Token 超出上下文窗口"""
def __init__(
self,
provider: str,
max_tokens: int,
requested_tokens: int,
model: Optional[str] = None,
):
self.max_tokens = max_tokens
self.requested_tokens = requested_tokens
super().__init__(
code="LLM_TOKEN_LIMIT",
message=f"Token limit {max_tokens} exceeded (requested {requested_tokens})",
provider=provider,
model=model,
)
2.5 网络错误
class NetworkError(RetryableError):
"""网络通信错误"""
def __init__(
self,
code: str,
message: str = "",
endpoint: Optional[str] = None,
cause: Optional[Exception] = None,
):
self.endpoint = endpoint
super().__init__(
code=code,
message=message,
severity=Severity.WARNING,
category=ErrorCategory.TRANSITORY,
cause=cause,
)
class RPCConnectionError(NetworkError):
"""RPC 节点连接失败"""
def __init__(self, endpoint: str, cause: Optional[Exception] = None):
super().__init__(
code="RPC_CONNECTION",
message=f"Cannot connect to RPC endpoint {endpoint}",
endpoint=endpoint,
cause=cause,
)
class RPCNodeUnhealthyError(NetworkError):
"""RPC 节点不健康（返回 5xx 或空响应）"""
def __init__(self, endpoint: str, status_code: int, body: str = ""):
self.status_code = status_code
self.body = body[:200]
super().__init__(
code="RPC_NODE_UNHEALTHY",
message=f"RPC {endpoint} returned status {status_code}",
endpoint=endpoint,
)
class DNSResolutionError(NetworkError):
"""DNS 解析失败"""
def __init__(self, hostname: str, cause: Optional[Exception] = None):
self.hostname = hostname
super().__init__(
code="DNS_FAILURE",
message=f"DNS resolution failed for {hostname}",
endpoint=hostname,
cause=cause,
)
class WebSocketDisconnectError(NetworkError):
"""WebSocket 连接意外断开"""
def __init__(self, endpoint: str, code: int = 0, reason: str = ""):
self.close_code = code
self.close_reason = reason
super().__init__(
code="WS_DISCONNECT",
message=f"WebSocket disconnected from {endpoint}: code={code} reason={reason}",
endpoint=endpoint,
)
class IBCRelayerError(NetworkError):
"""IBC 中继器错误"""
def __init__(
self,
channel_id: str,
sequence: int,
error_type: str,
endpoint: Optional[str] = None,
):
self.channel_id = channel_id
self.sequence = sequence
self.error_type = error_type
super().__init__(
code="IBC_RELAYER",
message=f"IBC relayer error on {channel_id} seq:{sequence} type:{error_type}",
endpoint=endpoint,
)
2.6 支付错误
class PaymentError(AgentError):
"""支付相关错误"""
def __init__(
self,
code: str,
message: str = "",
severity: Severity = Severity.ERROR,
retryable: bool = False,
category: ErrorCategory = ErrorCategory.PERSISTENT,
cause: Optional[Exception] = None,
):
super().__init__(
code=code,
message=message,
severity=severity,
retryable=retryable,
category=category,
cause=cause,
)
class InsufficientBalanceError(PaymentError):
"""余不足（包含 Gas 费用）"""
def __init__(self, address: str, needed: str, available: str, denom: str):
self.address = address
self.needed = needed
self.available = available
self.denom = denom
super().__init__(
code="BALANCE_INSUFFICIENT",
message=f"Address {address} needs {needed} {denom}, has {available} {denom}",
retryable=False,
)
class FeeEstimationError(PaymentError):
"""Gas 费用估算错误"""
def __init__(self, gas_limit: int, gas_price: str, multiplier: float):
self.gas_limit = gas_limit
self.gas_price = gas_price
self.multiplier = multiplier
super().__init__(
code="FEE_ESTIMATION",
message=f"Fee estimation failed: {gas_limit} * {gas_price} * {multiplier}",
retryable=True,
category=ErrorCategory.TRANSITORY,
)
class MaxFeeExceededError(PaymentError):
"""费用超过最大限制"""
def __init__(self, estimated: str, max_allowed: str, denom: str):
self.estimated = estimated
self.max_allowed = max_allowed
self.denom = denom
super().__init__(
code="MAX_FEE_EXCEEDED",
message=f"Estimated fee {estimated} {denom} exceeds max {max_allowed} {denom}",
retryable=True,
category=ErrorCategory.TRANSITORY,
)
class RefundError(PaymentError):
"""退款失败"""
def __init__(self, original_tx: str, reason: str):
self.original_tx = original_tx
super().__init__(
code="REFUND_FAILED",
message=f"Refund for tx {original_tx} failed: {reason}",
retryable=True,
category=ErrorCategory.TRANSITORY,
)
2.7 错误路由器
class ErrorRouter:
"""将异常路由到正确的处理策略"""
def __init__(self):
self._handlers: Dict[str, Callable[[AgentError], Awaitable[None]]] = {}
self._fallback: Optional[Callable[[AgentError], Awaitable[None]]] = None
def register(
self,
error_code: str,
handler: Callable[[AgentError], Awaitable[None]],
):
self._handlers[error_code] = handler
async def route(self, error: AgentError):
handler = self._handlers.get(error.code, self._fallback)
if handler:
await handler(error)
else:
logger.error(f"Unhandled error: {error.code} - {error}")
@staticmethod
def classify_error(exc: Exception) -> AgentError:
if isinstance(exc, AgentError):
return exc
if isinstance(exc, ConnectionError):
return RPCConnectionError(endpoint="unknown", cause=exc)
if isinstance(exc, TimeoutError):
return NetworkError(
code="GENERIC_TIMEOUT",
message=str(exc),
cause=exc,
)
return AgentError(
code="UNKNOWN",
message=str(exc),
severity=Severity.ERROR,
retryable=False,
cause=exc,
)
# 全局错误路由器
error_router = ErrorRouter()
3. 重试策略
3.1 基础重试策略
T = TypeVar("T")
class MaxRetriesExceeded(NonRetryableError):
"""超过最大重试次数"""
def __init__(
self,
last_error: Optional[Exception] = None,
attempts: int = 0,
total_duration: float = 0.0,
):
self.attempts = attempts
self.total_duration = total_duration
super().__init__(
code="MAX_RETRIES",
message=f"Exceeded max retries after {attempts} attempts ({total_duration:.2f}s)",
cause=last_error,
)
class RetryBudgetExhausted(NonRetryableError):
"""重试预算耗尽"""
def __init__(self, budget: "RetryBudget"):
self.budget = budget
super().__init__(
code="RETRY_BUDGET",
message=f"Retry budget exhausted: used {budget.used}/{budget.max_retries}",
)
class RetryStrategy:
"""可配置的重试策略，支持退避、抖动和预算控制"""
def __init__(
self,
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
backoff_factor: float = 2.0,
jitter_factor: float = 0.1,
retryable_exceptions: Tuple[type, ...] = (RetryableError,),
budget: Optional["RetryBudget"] = None,
):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.backoff_factor = backoff_factor
self.jitter_factor = jitter_factor
self.retryable_exceptions = retryable_exceptions
self.budget = budget
def compute_delay(self, attempt: int) -> float:
"""计算第 attempt 次重试的等待时间（指数退避 + 抖动）"""
delay = min(
self.base_delay * (self.backoff_factor ** attempt),
self.max_delay,
)
jitter = random.uniform(0, delay * self.jitter_factor)
return delay + jitter
def is_retryable(self, exc: Exception) -> bool:
"""判断异常是否可重试"""
return isinstance(exc, self.retryable_exceptions)
async def execute_with_retry(
self,
fn: Callable[..., Awaitable[T]],
*args: Any,
**kwargs: Any,
) -> T:
"""执行可重试的函数调用"""
last_error: Optional[Exception] = None
start_time = time.time()
for attempt in range(self.max_retries + 1):
try:
if attempt > 0 and self.budget is not None:
self.budget.record_retry()
return await fn(*args, **kwargs)
except Exception as e:
last_error = e
if not self.is_retryable(e):
logger.warning(
f"Non-retryable error on attempt {attempt + 1}: {e}"
)
raise
if attempt >= self.max_retries:
logger.error(
f"All {self.max_retries + 1} attempts failed"
)
raise MaxRetriesExceeded(
last_error=last_error,
attempts=attempt + 1,
total_duration=time.time() - start_time,
) from e
delay = self.compute_delay(attempt)
logger.info(
f"Retry attempt {attempt + 1}/{self.max_retries} "
f"after {delay:.2f}s delay. Error: {e}"
)
if self.budget is not None and self.budget.is_exhausted():
raise RetryBudgetExhausted(self.budget)
await asyncio.sleep(delay)
# 不应到达这里
raise MaxRetriesExceeded(
last_error=last_error,
attempts=self.max_retries + 1,
total_duration=time.time() - start_time,
)
class RetryStrategyBuilder:
"""构建器模式创建 RetryStrategy"""
def __init__(self):
self._max_retries = 3
self._base_delay = 1.0
self._max_delay = 60.0
self._backoff_factor = 2.0
self._jitter_factor = 0.1
self._retryable_exceptions = (RetryableError,)
self._budget = None
def with_max_retries(self, n: int) -> "RetryStrategyBuilder":
self._max_retries = n
return self
def with_base_delay(self, d: float) -> "RetryStrategyBuilder":
self._base_delay = d
return self
def with_max_delay(self, d: float) -> "RetryStrategyBuilder":
self._max_delay = d
return self
def with_exponential_backoff(self, factor: float) -> "RetryStrategyBuilder":
self._backoff_factor = factor
return self
def with_jitter(self, factor: float) -> "RetryStrategyBuilder":
self._jitter_factor = factor
return self
def with_budget(self, budget: "RetryBudget") -> "RetryStrategyBuilder":
self._budget = budget
return self
def build(self) -> RetryStrategy:
return RetryStrategy(
max_retries=self._max_retries,
base_delay=self._base_delay,
max_delay=self._max_delay,
backoff_factor=self._backoff_factor,
jitter_factor=self._jitter_factor,
retryable_exceptions=self._retryable_exceptions,
budget=self._budget,
)
3.2 重试预算
class RetryBudget:
"""全局重试预算——限制某个时间段内的重试总次数"""
def __init__(
self,
max_retries: int = 100,
window_seconds: float = 300.0,
):
self.max_retries = max_retries
self.window_seconds = window_seconds
self.used = 0
self.window_start = time.time()
def record_retry(self):
current = time.time()
if current - self.window_start > self.window_seconds:
self.used = 0
self.window_start = current
self.used += 1
def is_exhausted(self) -> bool:
current = time.time()
if current - self.window_start > self.window_seconds:
self.used = 0
self.window_start = current
return False
return self.used >= self.max_retries
def remaining(self) -> int:
if self.is_exhausted():
return 0
return self.max_retries - self.used
class PerEndpointRetryBudget:
"""每个 RPC 端点的重试预算"""
def __init__(self, max_retries_per_endpoint: int = 20, window: float = 60.0):
self.max_retries = max_retries_per_endpoint
self.window = window
self._endpoints: Dict[str, RetryBudget] = {}
def get_budget(self, endpoint: str) -> RetryBudget:
if endpoint not in self._endpoints:
self._endpoints[endpoint] = RetryBudget(
max_retries=self.max_retries,
window_seconds=self.window,
)
return self._endpoints[endpoint]
# 预定义的策略工厂
class RetryPolicies:
"""常见场景的预配置重试策略"""
@staticmethod
def chain_tx() -> RetryStrategy:
"""链交易：退避较快，重试较多"""
return RetryStrategyBuilder() \
.with_max_retries(5) \
.with_base_delay(0.5) \
.with_max_delay(30.0) \
.with_exponential_backoff(2.0) \
.with_jitter(0.2) \
.build()
@staticmethod
def rpc_call() -> RetryStrategy:
"""RPC 调用：快速重试"""
return RetryStrategyBuilder() \
.with_max_retries(3) \
.with_base_delay(0.2) \
.with_max_delay(5.0) \
.with_exponential_backoff(2.0) \
.with_jitter(0.1) \
.build()
@staticmethod
def llm_inference() -> RetryStrategy:
"""LLM 推理：退避较慢（避免触发速率限制）"""
return RetryStrategyBuilder() \
.with_max_retries(3) \
.with_base_delay(2.0) \
.with_max_delay(30.0) \
.with_exponential_backoff(3.0) \
.with_jitter(0.3) \
.build()
@staticmethod
def ibc_transfer() -> RetryStrategy:
"""IBC 转账：长时间退避"""
return RetryStrategyBuilder() \
.with_max_retries(10) \
.with_base_delay(6.0) \
.with_max_delay(300.0) \
.with_exponential_backoff(1.5) \
.with_jitter(0.2) \
.build()
@staticmethod
def no_retry() -> RetryStrategy:
"""不重试"""
return RetryStrategyBuilder() \
.with_max_retries(0) \
.build()
3.3 幂等性重试
class IdempotencyKey:
"""幂等性键，确保重试不会导致重复执行"""
def __init__(self, prefix: str = "msg"):
self.prefix = prefix
@staticmethod
def generate(agent_id: str, operation: str, nonce: Optional[str] = None) -> str:
"""生成幂等性键：agent_id + operation + nonce 的 HMAC"""
key_material = nonce or str(uuid.uuid4())
payload = f"{agent_id}:{operation}:{key_material}"
digest = hashlib.sha256(payload.encode()).hexdigest()[:16]
return f"msg-{digest}"
@staticmethod
def parse(key: str) -> Dict[str, str]:
parts = key.split("-", 1)
return {"prefix": parts[0], "hash": parts[1]} if len(parts) == 2 else {}
class IdempotentRetryStrategy(RetryStrategy):
"""带幂等性检查的重试策略"""
def __init__(
self,
idempotency_store: Dict[str, Any],
ttl_seconds: float = 3600.0,
**kwargs,
):
super().__init__(**kwargs)
self._store = idempotency_store
self._ttl = ttl_seconds
async def execute_idempotent(
self,
idempotency_key: str,
fn: Callable[..., Awaitable[T]],
*args: Any,
**kwargs: Any,
) -> T:
"""幂等地执行操作——如果已执行则返回缓存结果"""
if idempotency_key in self._store:
cached = self._store[idempotency_key]
if time.time() - cached["timestamp"] < self._ttl:
logger.info(f"Idempotency hit for key {idempotency_key}")
return cached["result"]
else:
# TTL 过期，清理
del self._store[idempotency_key]
result = await self.execute_with_retry(fn, *args, **kwargs)
self._store[idempotency_key] = {
"result": result,
"timestamp": time.time(),
}
return result
3.4 重试装饰器
def retryable(
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
retry_on: Tuple[type, ...] = (RetryableError,),
):
"""函数装饰器——为异步函数添加重试逻辑"""
strategy = RetryStrategy(
max_retries=max_retries,
base_delay=base_delay,
max_delay=max_delay,
retryable_exceptions=retry_on,
)
def decorator(fn: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
async def wrapper(*args: Any, **kwargs: Any) -> T:
return await strategy.execute_with_retry(fn, *args, **kwargs)
return wrapper
return decorator
# 使用示例
# @retryable(max_retries=5, base_delay=0.5)
# async def broadcast_tx(tx: SignedTx) -> TxResponse:
# ...
3.5 智能体级重试协调器
class AgentRetryCoordinator:
"""智能体级别的重试协调——管理所有子任务的重试状态"""
def __init__(self, agent_id: str):
self.agent_id = agent_id
self._retry_counts: Dict[str, int] = defaultdict(int)
self._max_global_retries = 50
self._global_retries = 0
self._budget = RetryBudget(max_retries=self._max_global_retries)
def can_retry(self, operation: str) -> bool:
"""检查特定操作和全局是否还能重试"""
if self._budget.is_exhausted():
logger.warning(f"Agent {self.agent_id} retry budget exhausted")
return False
return True
def record_attempt(self, operation: str):
self._retry_counts[operation] += 1
self._global_retries += 1
self._budget.record_retry()
def get_attempt_count(self, operation: str) -> int:
return self._retry_counts.get(operation, 0)
def reset(self):
self._retry_counts.clear()
self._global_retries = 0
def status(self) -> Dict[str, Any]:
return {
"agent_id": self.agent_id,
"global_retries": self._global_retries,
"max_global_retries": self._max_global_retries,
"operations": dict(self._retry_counts),
"budget_remaining": self._budget.remaining(),
}
4. 熔断器模式
4.1 基础熔断器
class CircuitBreakerState(Enum):
CLOSED = "CLOSED" # 正常工作
OPEN = "OPEN" # 熔断开启，请求快速失败
HALF_OPEN = "HALF_OPEN" # 半开状态，允许探测请求
class CircuitBreakerOpen(NonRetryableError):
"""熔断器开启时抛出的异常"""
def __init__(self, breaker_name: str, state: CircuitBreakerState):
self.breaker_name = breaker_name
self.breaker_state = state
super().__init__(
code="CIRCUIT_BREAKER_OPEN",
message=f"Circuit breaker '{breaker_name}' is OPEN, request rejected",
severity=Severity.WARNING,
)
class CircuitBreaker:
"""熔断器——防止级联故障"""
def __init__(
self,
name: str = "default",
failure_threshold: int = 5,
recovery_timeout: float = 30.0,
half_open_max_requests: int = 1,
consecutive_successes_to_close: int = 2,
):
self.name = name
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_requests = half_open_max_requests
self.consecutive_successes_to_close = consecutive_successes_to_close
self.state = CircuitBreakerState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time = 0.0
self.last_success_time = 0.0
self.half_open_requests = 0
self.total_failures = 0
self.total_successes = 0
self._lock = asyncio.Lock()
async def call(
self,
fn: Callable[..., Awaitable[T]],
*args: Any,
**kwargs: Any,
) -> T:
"""调用受熔断器保护的函数"""
async with self._lock:
if self.state == CircuitBreakerState.OPEN:
if time.time() - self.last_failure_time > self.recovery_timeout:
logger.info(
f"Breaker '{self.name}' transitioning OPEN -> HALF_OPEN"
)
self.state = CircuitBreakerState.HALF_OPEN
self.half_open_requests = 0
self.success_count = 0
else:
raise CircuitBreakerOpen(self.name, self.state)
if self.state == CircuitBreakerState.HALF_OPEN:
if self.half_open_requests >= self.half_open_max_requests:
raise CircuitBreakerOpen(self.name, self.state)
self.half_open_requests += 1
try:
result = await fn(*args, **kwargs)
async with self._lock:
self.total_successes += 1
self.last_success_time = time.time()
if self.state == CircuitBreakerState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.consecutive_successes_to_close:
logger.info(
f"Breaker '{self.name}' transitioning HALF_OPEN -> CLOSED"
)
self.state = CircuitBreakerState.CLOSED
self.failure_count = 0
self.success_count = 0
self.failure_count = max(0, self.failure_count - 1)
return result
except Exception as e:
async with self._lock:
self.total_failures += 1
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitBreakerState.HALF_OPEN:
logger.info(
f"Breaker '{self.name}' transitioning HALF_OPEN -> OPEN"
)
self.state = CircuitBreakerState.OPEN
elif (
self.state == CircuitBreakerState.CLOSED
and self.failure_count >= self.failure_threshold
):
logger.warning(
f"Breaker '{self.name}' transitioning CLOSED -> OPEN "
f"(failures: {self.failure_count}/{self.failure_threshold})"
)
self.state = CircuitBreakerState.OPEN
raise
def force_state(self, state: CircuitBreakerState):
"""强制设置熔断器状态（用于手动干预）"""
self.state = state
if state == CircuitBreakerState.CLOSED:
self.failure_count = 0
self.success_count = 0
@property
def is_available(self) -> bool:
return self.state != CircuitBreakerState.OPEN
def metrics(self) -> Dict[str, Any]:
return {
"name": self.name,
"state": self.state.value,
"failure_count": self.failure_count,
"failure_threshold": self.failure_threshold,
"total_failures": self.total_failures,
"total_successes": self.total_successes,
"last_failure_time": self.last_failure_time,
"last_success_time": self.last_success_time,
"available": self.is_available,
}
def __repr__(self) -> str:
return (
f"CircuitBreaker(name='{self.name}', state={self.state.value}, "
f"failures={self.failure_count}/{self.failure_threshold})"
)
4.2 熔断器注册表
class CircuitBreakerRegistry:
"""全局熔断器注册表——按服务名称管理多个熔断器"""
def __init__(self):
self._breakers: Dict[str, CircuitBreaker] = {}
self._default_config: Dict[str, Any] = {
"failure_threshold": 5,
"recovery_timeout": 30.0,
"half_open_max_requests": 1,
"consecutive_successes_to_close": 2,
}
def get_or_create(self, name: str, **overrides) -> CircuitBreaker:
if name not in self._breakers:
config = {**self._default_config, **overrides}
self._breakers[name] = CircuitBreaker(name=name, **config)
return self._breakers[name]
def get(self, name: str) -> Optional[CircuitBreaker]:
return self._breakers.get(name)
def all_metrics(self) -> Dict[str, Dict[str, Any]]:
return {name: cb.metrics() for name, cb in self._breakers.items()}
def reset_all(self):
for cb in self._breakers.values():
cb.force_state(CircuitBreakerState.CLOSED)
def __repr__(self) -> str:
return f"CircuitBreakerRegistry(breakers={list(self._breakers.keys())})"
# 预定义的熔断器端点
class BreakerEndpoints:
"""按服务划分的熔断器端点名称"""
MSG_RPC_PRIMARY = "msg_rpc_primary"
MSG_RPC_SECONDARY = "msg_rpc_secondary"
LLM_INFERENCE = "llm_inference"
INDEXER_QUERY = "indexer_query"
A2A_MESSAGING = "a2a_messaging"
IBC_RELAY = "ibc_relay"
PRICE_ORACLE = "price_oracle"
breaker_registry = CircuitBreakerRegistry()
4.3 熔断器装饰器
def circuit_breaker(breaker_name: str, **breaker_kwargs):
"""方法装饰器——使用指定熔断器保护异步函数"""
def decorator(fn: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
async def wrapper(*args: Any, **kwargs: Any) -> T:
breaker = breaker_registry.get_or_create(breaker_name, **breaker_kwargs)
return await breaker.call(fn, *args, **kwargs)
return wrapper
return decorator
class CircuitBreakerContext:
"""上下文管理器形式的熔断器"""
def __init__(self, breaker_name: str):
self.breaker = breaker_registry.get_or_create(breaker_name)
async def __aenter__(self) -> "CircuitBreakerContext":
return self
async def __aexit__(
self,
exc_type: Optional[type],
exc_val: Optional[Exception],
exc_tb: Any,
) -> bool:
if exc_type is None:
# 成功
self.breaker.failure_count = max(0, self.breaker.failure_count - 1)
self.breaker.total_successes += 1
self.breaker.last_success_time = time.time()
if (
self.breaker.state == CircuitBreakerState.HALF_OPEN
):
self.breaker.success_count += 1
if self.breaker.success_count >= self.breaker.consecutive_successes_to_close:
self.breaker.state = CircuitBreakerState.CLOSED
self.breaker.failure_count = 0
self.breaker.success_count = 0
return False
if isinstance(exc_val, CircuitBreakerOpen):
return False # 不处理——让调用者处理
# 失败
self.breaker.failure_count += 1
self.breaker.total_failures += 1
self.breaker.last_failure_time = time.time()
if self.breaker.state == CircuitBreakerState.HALF_OPEN:
self.breaker.state = CircuitBreakerState.OPEN
elif (
self.breaker.state == CircuitBreakerState.CLOSED
and self.breaker.failure_count >= self.breaker.failure_threshold
):
self.breaker.state = CircuitBreakerState.OPEN
return False
4.4 级联熔断器
class CascadingCircuitBreaker:
"""级联熔断器——下游熔断器触发时自动触发上游熔断"""
def __init__(self, name: str):
self.name = name
self.downstream: List[CircuitBreaker] = []
self.upstream: List[CircuitBreaker] = []
self._local_breaker = CircuitBreaker(name=f"{name}_local")
def add_downstream(self, breaker: CircuitBreaker):
self.downstream.append(breaker)
def add_upstream(self, breaker: CircuitBreaker):
self.upstream.append(breaker)
async def call(
self,
fn: Callable[..., Awaitable[T]],
*args: Any,
**kwargs: Any,
) -> T:
# 检查下游熔断器状态
for downstream in self.downstream:
if not downstream.is_available:
raise CircuitBreakerOpen(
f"{self.name} (downstream {downstream.name} OPEN)",
CircuitBreakerState.OPEN,
)
try:
result = await self._local_breaker.call(fn, *args, **kwargs)
return result
except Exception:
# 本地熔断器开启时，通知上游
if self._local_breaker.state == CircuitBreakerState.OPEN:
for upstream in self.upstream:
upstream.force_state(CircuitBreakerState.OPEN)
raise
5. 超时与截止时间
5.1 超时管理器
class TimeoutConfig:
"""超时配置——按操作类型定义超时时间"""
def __init__(
self,
rpc_call: float = 10.0,
tx_broadcast: float = 60.0,
tx_confirmation: float = 120.0,
llm_inference: float = 30.0,
indexer_query: float = 15.0,
a2a_message: float = 20.0,
ibc_ack: float = 300.0,
):
self.rpc_call = rpc_call
self.tx_broadcast = tx_broadcast
self.tx_confirmation = tx_confirmation
self.llm_inference = llm_inference
self.indexer_query = indexer_query
self.a2a_message = a2a_message
self.ibc_ack = ibc_ack
def get(self, operation: str) -> float:
return getattr(self, operation, 30.0)
DEFAULT_TIMEOUTS = TimeoutConfig()
class OperationTimeout:
"""操作超时——使用 asyncio.wait_for 实现"""
def __init__(
self,
timeout: float,
operation: str = "unknown",
error_code: str = "OPERATION_TIMEOUT",
):
self.timeout = timeout
self.operation = operation
self.error_code = error_code
async def run(
self,
fn: Callable[..., Awaitable[T]],
*args: Any,
**kwargs: Any,
) -> T:
try:
return await asyncio.wait_for(
fn(*args, **kwargs),
timeout=self.timeout,
)
except asyncio.TimeoutError:
raise NetworkError(
code=self.error_code,
message=f"Operation '{self.operation}' timed out after {self.timeout}s",
)
@asynccontextmanager
async def with_timeout(timeout: float, operation: str = "unknown"):
"""上下文管理器形式的超时"""
try:
yield await asyncio.wait_for(_null_context(), timeout=timeout)
except asyncio.TimeoutError:
raise NetworkError(
code="CONTEXT_TIMEOUT",
message=f"Context '{operation}' timed out after {timeout}s",
)
async def _null_context() -> None:
"""内部辅助——空协程"""
pass
5.2 截止时间传播
@dataclass
class Deadline:
"""截止时间——携带上下文传播的最后期限"""
value: float # Unix timestamp
source: str = "local"
metadata: Dict[str, Any] = field(default_factory=dict)
@staticmethod
def from_timeout(timeout: float, source: str = "local") -> "Deadline":
return Deadline(value=time.time() + timeout, source=source)
@property
def remaining(self) -> float:
return max(0.0, self.value - time.time())
@property
def expired(self) -> bool:
return time.time() >= self.value
def check(self):
"""检查截止时间是否已过期"""
if self.expired:
raise NetworkError(
code="DEADLINE_EXCEEDED",
message=f"Deadline from {self.source} exceeded",
)
def with_buffer(self, buffer: float) -> "Deadline":
"""创建一个更早的截止时间（为下游操作预留缓冲）"""
new_value = self.value - buffer
return Deadline(value=new_value, source=self.source)
def to_dict(self) -> Dict[str, Any]:
return {
"value": self.value,
"source": self.source,
"remaining": self.remaining,
"expired": self.expired,
}
class DeadlineAwareRetryStrategy(RetryStrategy):
"""感知截止时间的重试策略——如果在截止时间内无法完成重试则提前放弃"""
async def execute_with_deadline(
self,
deadline: Deadline,
fn: Callable[..., Awaitable[T]],
*args: Any,
**kwargs: Any,
) -> T:
deadline.check()
last_error = None
# 根据剩余时间动态计算最大重试次数
max_possible = 0
cumulative_delay = 0.0
for i in range(self.max_retries + 1):
if i > 0:
cumulative_delay += self.compute_delay(i - 1)
if cumulative_delay > deadline.remaining:
break
max_possible += 1
if max_possible == 0 and self.max_retries > 0:
raise MaxRetriesExceeded(
last_error=None,
attempts=0,
total_duration=0.0,
)
effective_max = min(self.max_retries, max_possible - 1)
strategy = RetryStrategy(
max_retries=effective_max,
base_delay=self.base_delay,
max_delay=self.max_delay,
backoff_factor=self.backoff_factor,
jitter_factor=self.jitter_factor,
)
return await strategy.execute_with_retry(fn, *args, **kwargs)
5.3 链交易超时处理
class TxTimeoutHandler:
"""链交易超时处理——监控交易确认状态"""
def __init__(
self,
rpc_client: Any,
max_blocks_wait: int = 50,
poll_interval: float = 1.0,
):
self.rpc_client = rpc_client
self.max_blocks_wait = max_blocks_wait
self.poll_interval = poll_interval
async def wait_for_tx(
self,
tx_hash: str,
deadline: Optional[Deadline] = None,
) -> Dict[str, Any]:
"""轮询等待交易确认，超时后抛出 TxTimeoutError"""
start_height = await self._get_current_height()
start_time = time.time()
for _ in range(self.max_blocks_wait):
if deadline and deadline.expired:
raise TxTimeoutError(
tx_hash=tx_hash,
max_blocks=self.max_blocks_wait,
current_height=await self._get_current_height(),
)
try:
tx_response = await self.rpc_client.query_tx(tx_hash)
if tx_response is not None:
if tx_response.get("code", 0) != 0:
raise TxRevertError(
tx_hash=tx_hash,
revert_reason=tx_response.get("raw_log", "unknown"),
gas_used=int(tx_response.get("gas_used", 0)),
)
return tx_response
except Exception as e:
if not isinstance(e, TxRevertError):
logger.warning(f"Polling tx {tx_hash}: {e}")
await asyncio.sleep(self.poll_interval)
current_height = await self._get_current_height()
blocks_elapsed = current_height - start_height
raise TxTimeoutError(
tx_hash=tx_hash,
max_blocks=self.max_blocks_wait,
current_height=current_height,
)
async def _get_current_height(self) -> int:
result = await self.rpc_client.query_latest_block()
return int(result.get("block", {}).get("header", {}).get("height", 0))
class AdaptiveTxTimeout:
"""自适应交易超时——根据网络拥堵情况动态调整"""
def __init__(self, rpc_client: Any):
self.rpc_client = rpc_client
self._base_blocks = 20
self._max_blocks = 100
self._current_multiplier = 1.0
async def estimate_blocks_needed(self, gas_limit: int) -> int:
"""根据最近区块的 Gas 使用情况估算交易确认需要的区块数"""
try:
recent_gas = await self._get_recent_gas_usage(10)
avg_gas_usage = sum(recent_gas) / max(len(recent_gas), 1)
block_gas_limit = await self._get_block_gas_limit()
congestion_ratio = avg_gas_usage / max(block_gas_limit, 1)
if congestion_ratio > 0.9:
self._current_multiplier = 3.0
elif congestion_ratio > 0.7:
self._current_multiplier = 2.0
else:
self._current_multiplier = 1.0
return int(self._base_blocks * self._current_multiplier)
except Exception:
return self._base_blocks
async def _get_recent_gas_usage(self, blocks: int) -> List[int]:
result = await self.rpc_client.query_blocks(blocks)
gas_usage = []
for block in result.get("blocks", []):
usage = sum(
tx.get("gas_used", 0)
for tx in block.get("data", {}).get("txs", [])
)
gas_usage.append(usage)
return gas_usage
async def _get_block_gas_limit(self) -> int:
result = await self.rpc_client.query_consensus_params()
return int(
result.get("consensus_params", {})
.get("block", {})
.get("max_gas", -1)
)
5.4 上下文传播
@dataclass
class AgentContext:
"""智能体上下文——贯穿请求生命周期的元数据"""
agent_id: str
trace_id: str = field(default_factory=lambda: str(uuid.uuid4()))
deadline: Optional[Deadline] = None
retry_count: int = 0
max_depth: int = 10
metadata: Dict[str, Any] = field(default_factory=dict)
def child(self, operation: str) -> "AgentContext":
"""创建子上下文（用于嵌套调用）"""
if self.retry_count >= self.max_depth:
raise AgentError(
code="MAX_DEPTH",
message=f"Max call depth {self.max_depth} exceeded",
severity=Severity.ERROR,
retryable=False,
)
child_deadline = (
self.deadline.with_buffer(5.0)
if self.deadline
else None
)
return AgentContext(
agent_id=self.agent_id,
trace_id=self.trace_id,
deadline=child_deadline,
retry_count=self.retry_count + 1,
max_depth=self.max_depth,
metadata={**self.metadata, "parent_operation": operation},
)
def to_dict(self) -> Dict[str, Any]:
return {
"agent_id": self.agent_id,
"trace_id": self.trace_id,
"deadline": self.deadline.to_dict() if self.deadline else None,
"retry_count": self.retry_count,
"max_depth": self.max_depth,
}
class ContextVars:
"""线程/任务局部上下文变量"""
current = asyncio.local() # type: ignore
def get_current_context() -> Optional[AgentContext]:
"""获取当前任务的上下文"""
return getattr(ContextVars.current, "context", None)
def set_current_context(ctx: AgentContext):
"""设置当前任务的上下文"""
ContextVars.current.context = ctx
6. 降级策略
6.1 降级管理器
class DegradationLevel(Enum):
FULL = "full" # 全部功能可用
REDUCED = "reduced" # 降级模式（无 LLM 推理）
MINIMAL = "minimal" # 最小模式（仅缓存）
OFFLINE = "offline" # 离线模式（紧急停止）
class ServiceType(Enum):
LLM = "llm"
CHAIN = "chain"
CACHE = "cache"
INDEXER = "indexer"
A2A = "a2a"
ORACLE = "oracle"
IBC = "ibc"
class DegradationManager:
"""服务降级管理器——在依赖服务不可用时优雅降级"""
def __init__(self, agent_id: str):
self.agent_id = agent_id
self._current_level = DegradationLevel.FULL
# 每个降级级别对应的可用服务
self._service_map: Dict[DegradationLevel, List[ServiceType]] = {
DegradationLevel.FULL: [
ServiceType.LLM,
ServiceType.CHAIN,
ServiceType.CACHE,
ServiceType.INDEXER,
ServiceType.A2A,
ServiceType.ORACLE,
ServiceType.IBC,
],
DegradationLevel.REDUCED: [
ServiceType.CHAIN,
ServiceType.CACHE,
ServiceType.INDEXER,
ServiceType.A2A,
ServiceType.ORACLE,
],
DegradationLevel.MINIMAL: [
ServiceType.CACHE,
ServiceType.CHAIN,
],
DegradationLevel.OFFLINE: [],
}
self._service_failures: Dict[ServiceType, int] = defaultdict(int)
self._failure_threshold: Dict[ServiceType, int] = {
ServiceType.LLM: 3,
ServiceType.INDEXER: 5,
ServiceType.CHAIN: 10,
ServiceType.A2A: 3,
ServiceType.ORACLE: 5,
ServiceType.IBC: 3,
}
self._service_breakers: Dict[ServiceType, CircuitBreaker] = {}
def _get_breaker(self, service: ServiceType) -> CircuitBreaker:
if service not in self._service_breakers:
self._service_breakers[service] = CircuitBreaker(
name=f"{self.agent_id}_{service.value}",
failure_threshold=self._failure_threshold.get(service, 5),
)
return self._service_breakers[service]
def is_service_available(self, service: ServiceType) -> bool:
"""检查特定服务在当前降级级别下是否可用"""
return service in self._service_map[self._current_level]
def record_failure(self, service: ServiceType):
"""记录服务失败，必要时自动降级"""
self._service_failures[service] += 1
breaker = self._get_breaker(service)
breaker.failure_count = self._service_failures[service]
if self._should_degrade(service):
self._auto_degrade(service)
def _should_degrade(self, service: ServiceType) -> bool:
failures = self._service_failures[service]
threshold = self._failure_threshold.get(service, 5)
return failures >= threshold
def _auto_degrade(self, failed_service: ServiceType):
"""根据失败的服务自动选择降级级别"""
if failed_service == ServiceType.LLM:
if self._current_level == DegradationLevel.FULL:
self.set_level(DegradationLevel.REDUCED, f"LLM failed after {self._service_failures[ServiceType.LLM]} attempts")
elif failed_service == ServiceType.INDEXER:
if self._current_level == DegradationLevel.FULL:
self.set_level(DegradationLevel.REDUCED, "Indexer unavailable")
elif self._current_level == DegradationLevel.REDUCED:
self.set_level(DegradationLevel.MINIMAL, "Indexer and LLM both unavailable")
elif failed_service == ServiceType.CHAIN:
self.set_level(DegradationLevel.OFFLINE, "Chain unavailable")
elif failed_service == ServiceType.A2A:
pass # A2A 失败不影响核心功能
def set_level(self, level: DegradationLevel, reason: str = ""):
"""手动设置降级级别"""
if level.value < self._current_level.value:
logger.warning(
f"Agent {self.agent_id} degrading: {self._current_level.value} -> {level.value}. "
f"Reason: {reason}"
)
else:
logger.info(
f"Agent {self.agent_id} upgrading: {self._current_level.value} -> {level.value}"
)
self._current_level = level
def available_services(self) -> List[ServiceType]:
"""获取当前可用的服务列表"""
return self._service_map[self._current_level]
@property
def is_online(self) -> bool:
return self._current_level != DegradationLevel.OFFLINE
@property
def can_use_llm(self) -> bool:
return ServiceType.LLM in self.available_services()
def status(self) -> Dict[str, Any]:
return {
"agent_id": self.agent_id,
"level": self._current_level.value,
"available_services": [s.value for s in self.available_services()],
"service_failures": {
k.value: v for k, v in self._service_failures.items()
},
"online": self.is_online,
}
6.2 优雅回退
class FallbackResult(Generic[T]):
"""回退操作的结果"""
def __init__(
self,
value: T,
source: str = "primary",
stale: bool = False,
latency_ms: float = 0.0,
):
self.value = value
self.source = source
self.stale = stale
self.latency_ms = latency_ms
def __repr__(self) -> str:
return f"FallbackResult(value={self.value}, source={self.source}, stale={self.stale})"
class FallbackStrategy(ABC, Generic[T]):
"""回退策略抽象基类"""
@abstractmethod
async def execute(self, *args: Any, **kwargs: Any) -> FallbackResult[T]:
pass
class ChainFallback(FallbackStrategy[Dict[str, Any]]):
"""链查询回退——主 RPC -> 备用 RPC -> 公共 RPC"""
def __init__(
self,
primary_rpc: str,
secondary_rpc: str,
public_rpc: Optional[str] = None,
timeout: float = 10.0,
):
self.endpoints = [primary_rpc, secondary_rpc]
if public_rpc:
self.endpoints.append(public_rpc)
async def execute(self, query_fn: Callable[..., Awaitable[Any]], *args, **kwargs) -> FallbackResult[Any]:
last_error = None
for i, endpoint in enumerate(self.endpoints):
try:
start = time.time()
result = await asyncio.wait_for(
query_fn(endpoint, *args, **kwargs),
timeout=self.timeout if hasattr(self, 'timeout') else 10.0,
)
latency = (time.time() - start) * 1000
return FallbackResult(
value=result,
source=endpoint,
stale=False,
latency_ms=latency,
)
except Exception as e:
last_error = e
logger.warning(f"Fallback endpoint {endpoint} failed: {e}")
raise NetworkError(
code="ALL_ENDPOINTS_FAILED",
message=f"All {len(self.endpoints)} endpoints failed",
cause=last_error,
)
class LLMFallback(FallbackStrategy[str]):
"""LLM 推理回退——主模型 -> 备用模型 -> 简化推理"""
def __init__(
self,
primary_model: str = "gpt-4",
fallback_model: str = "gpt-3.5-turbo",
fallback2_model: str = "claude-instant",
enable_simple_fallback: bool = True,
):
self.models = [primary_model, fallback_model]
if enable_simple_fallback:
self.models.append(fallback2_model)
async def execute(
self,
llm_client: Any,
prompt: str,
**kwargs,
) -> FallbackResult[str]:
last_error = None
for model in self.models:
try:
start = time.time()
response = await llm_client.generate(prompt, model=model, **kwargs)
latency = (time.time() - start) * 1000
return FallbackResult(
value=response,
source=model,
stale=False,
latency_ms=latency,
)
except Exception as e:
last_error = e
logger.warning(f"LLM model {model} failed: {e}")
# 最终回退：使用规则引擎（简单模式匹配，不依赖 LLM）
if hasattr(self, 'enable_simple_fallback') and self.enable_simple_fallback:
return FallbackResult(
value=self._rule_based_fallback(prompt),
source="rule_engine",
stale=False,
latency_ms=0.0,
)
raise LLMError(
code="ALL_MODELS_FAILED",
message=f"All {len(self.models)} LLM models failed",
cause=last_error,
)
def _rule_based_fallback(self, prompt: str) -> str:
"""基于规则的简单回退（无 LLM 推理）"""
prompt_lower = prompt.lower()
if "balance" in prompt_lower or "余额" in prompt:
return '{"action": "query_balance", "confidence": 0.6}'
if "transfer" in prompt_lower or "转账" in prompt:
return '{"action": "check_transfer_params", "confidence": 0.5}'
if "swap" in prompt_lower or "交换" in prompt_lower:
return '{"action": "query_pool", "confidence": 0.5}'
return '{"action": "unknown", "confidence": 0.1}'
6.3 缓存回退
class StaleCacheServing:
"""过期缓存服务——当后端不可用时提供过期数据"""
def __init__(self, ttl_normal: float = 60.0, ttl_stale_max: float = 3600.0):
self._cache: Dict[str, Tuple[Any, float]] = {}
self.ttl_normal = ttl_normal
self.ttl_stale_max = ttl_stale_max
async def get_or_fetch(
self,
key: str,
fetcher: Callable[..., Awaitable[Any]],
allow_stale: bool = True,
*args: Any,
**kwargs: Any,
) -> FallbackResult[Any]:
now = time.time()
# 检查缓存
if key in self._cache:
value, timestamp = self._cache[key]
age = now - timestamp
if age < self.ttl_normal:
# 正常缓存命中
return FallbackResult(
value=value,
source="cache",
stale=False,
latency_ms=0.0,
)
if allow_stale and age < self.ttl_stale_max:
# 允许使用过期缓存
try:
fresh = await fetcher(*args, **kwargs)
self._cache[key] = (fresh, now)
return FallbackResult(
value=fresh,
source="fetcher",
stale=False,
latency_ms=0.0,
)
except Exception:
logger.warning(f"Fetcher failed, serving stale cache for {key}")
return FallbackResult(
value=value,
source="stale_cache",
stale=True,
latency_ms=0.0,
)
# 无缓存，尝试获取
try:
fresh = await fetcher(*args, **kwargs)
self._cache[key] = (fresh, now)
return FallbackResult(
value=fresh,
source="fetcher",
stale=False,
latency_ms=0.0,
)
except Exception as e:
if allow_stale and key in self._cache:
value, timestamp = self._cache[key]
age = now - timestamp
if age < self.ttl_stale_max:
return FallbackResult(
value=value,
source="stale_cache",
stale=True,
latency_ms=0.0,
)
raise
def invalidate(self, key: str):
self._cache.pop(key, None)
def clear(self):
self._cache.clear()
6.4 降级事件
@dataclass
class DegradationEvent:
"""降级事件记录"""
timestamp: float = field(default_factory=time.time)
agent_id: str = ""
from_level: str = ""
to_level: str = ""
reason: str = ""
duration_seconds: float = 0.0
services_affected: List[str] = field(default_factory=list)
class DegradationHistory:
"""降级历史记录"""
def __init__(self, max_events: int = 100):
self.events: List[DegradationEvent] = []
self.max_events = max_events
def record(
self,
agent_id: str,
from_level: str,
to_level: str,
reason: str,
services: List[str],
):
event = DegradationEvent(
agent_id=agent_id,
from_level=from_level,
to_level=to_level,
reason=reason,
services_affected=services,
)
if len(self.events) >= self.max_events:
self.events.pop(0)
self.events.append(event)
def summary(self) -> Dict[str, Any]:
if not self.events:
return {"total_events": 0, "downtime_minutes": 0.0}
total_duration = sum(
e.duration_seconds for e in self.events if e.duration_seconds > 0
)
return {
"total_events": len(self.events),
"downtime_minutes": total_duration / 60.0,
"last_event": self.events[-1].to_dict() if hasattr(self.events[-1], 'to_dict') else {},
}
7. 事务补偿
7.1 Saga 模式基础
class SagaStep(ABC):
"""Saga 模式中的一步"""
@abstractmethod
async def execute(self) -> Any:
pass
@abstractmethod
async def compensate(self) -> None:
pass
@property
@abstractmethod
def name(self) -> str:
pass
class SagaFailed(NonRetryableError):
"""Saga 执行失败"""
def __init__(self, cause: Exception, step_index: int = 0, step_name: str = ""):
self.step_index = step_index
self.step_name = step_name
super().__init__(
code="SAGA_FAILED",
message=f"Saga failed at step {step_index} '{step_name}': {cause}",
cause=cause,
)
class CompensationFailed(AgentError):
"""补偿操作失败——需要人工介入"""
def __init__(self, step_name: str, cause: Exception):
self.step_name = step_name
super().__init__(
code="COMPENSATION_FAILED",
message=f"Compensation for '{step_name}' failed: {cause}",
severity=Severity.CRITICAL,
retryable=True,
category=ErrorCategory.TRANSITORY,
)
class CompensationManager:
"""Saga 模式的事务补偿管理器"""
def __init__(self, saga_id: Optional[str] = None):
self.saga_id = saga_id or f"saga-{uuid.uuid4().hex[:12]}"
self.completed_steps: List[Tuple[int, str, Callable[[], Awaitable[None]]]] = []
self._status: str = "pending"
self._errors: List[Dict[str, Any]] = []
async def execute_saga(
self,
steps: List[SagaStep],
abort_on_failure: bool = True,
) -> List[Any]:
"""执行 Saga——每步执行成功记录补偿，失败则回滚"""
results = []
self._status = "running"
for i, step in enumerate(steps):
try:
result = await step.execute()
results.append(result)
self.completed_steps.append((i, step.name, step.compensate))
logger.info(f"Saga {self.saga_id}: step {i} '{step.name}' succeeded")
except Exception as e:
logger.error(
f"Saga {self.saga_id}: step {i} '{step.name}' failed: {e}"
)
self._errors.append({
"step_index": i,
"step_name": step.name,
"error": str(e),
})
if abort_on_failure:
await self._rollback(f"Step {i} '{step.name}' failed: {e}")
raise SagaFailed(
cause=e,
step_index=i,
step_name=step.name,
)
# 即使失败也继续执行（尽力而为模式）
results.append(None)
self._status = "completed"
return results
async def _rollback(self, reason: str):
"""反向执行所有已完成的补偿操作"""
self._status = "rolling_back"
logger.warning(
f"Saga {self.saga_id} rolling back {len(self.completed_steps)} steps. "
f"Reason: {reason}"
)
compensation_errors = []
for step_index, step_name, compensate in reversed(self.completed_steps):
try:
await compensate()
logger.info(f"Compensation for '{step_name}' succeeded")
except Exception as e:
logger.critical(
f"Compensation for '{step_name}' failed: {e}"
)
compensation_errors.append({
"step_index": step_index,
"step_name": step_name,
"error": str(e),
})
if compensation_errors:
self._status = "compensation_failed"
raise CompensationFailed(
step_name=compensation_errors[0]["step_name"],
cause=Exception(compensation_errors[0]["error"]),
)
self._status = "rolled_back"
async def execute_with_retry_compensation(
self,
step_name: str,
compensate_fn: Callable[[], Awaitable[None]],
max_retries: int = 3,
):
"""带重试的补偿执行"""
for attempt in range(max_retries):
try:
await compensate_fn()
return
except Exception as e:
if attempt < max_retries - 1:
delay = (2 ** attempt) * 1.0
logger.warning(
f"Compensation retry {attempt + 1}/{max_retries} "
f"for '{step_name}' after {delay}s"
)
await asyncio.sleep(delay)
else:
raise CompensationFailed(step_name=step_name, cause=e)
def status(self) -> Dict[str, Any]:
return {
"saga_id": self.saga_id,
"status": self._status,
"completed_steps": len(self.completed_steps),
"errors": self._errors,
}
7.2 链上操作的具体 Saga 步骤
class TransferStep(SagaStep):
"""通证转账步骤"""
def __init__(
self,
client: Any,
from_address: str,
to_address: str,
amount: str,
denom: str,
memo: str = "",
):
self.client = client
self.from_address = from_address
self.to_address = to_address
self.amount = amount
self.denom = denom
self.memo = memo
self._tx_hash: Optional[str] = None
self._idempotency_key: Optional[str] = None
@property
def name(self) -> str:
return f"transfer_{self.amount}{self.denom}_{self.from_address[:8]}"
async def execute(self) -> Dict[str, Any]:
self._idempotency_key = IdempotencyKey.generate(
agent_id=self.from_address,
operation=f"transfer_{self.amount}_{self.denom}_{self.to_address}",
)
result = await self.client.send_tokens(
from_address=self.from_address,
to_address=self.to_address,
amount=self.amount,
denom=self.denom,
memo=self.memo,
idempotency_key=self._idempotency_key,
)
self._tx_hash = result.get("tx_hash")
return result
async def compensate(self) -> None:
"""补偿：反向转账（需要确保有足够的资金）"""
if self._tx_hash:
# 检查转账是否已确认
tx_status = await self.client.query_tx(self._tx_hash)
if tx_status and tx_status.get("code", 0) == 0:
# 已确认，发反向转账
await self.client.send_tokens(
from_address=self.to_address,
to_address=self.from_address,
amount=self.amount,
denom=self.denom,
memo=f"compensation for {self._tx_hash[:16]}",
idempotency_key=f"comp-{self._idempotency_key}",
)
class SwapStep(SagaStep):
"""DEX 交换步骤"""
def __init__(
self,
client: Any,
pool_id: str,
sender: str,
token_in: str,
token_out: str,
amount_in: str,
min_amount_out: str,
):
self.client = client
self.pool_id = pool_id
self.sender = sender
self.token_in = token_in
self.token_out = token_out
self.amount_in = amount_in
self.min_amount_out = min_amount_out
self._actual_out: Optional[str] = None
self._tx_hash: Optional[str] = None
@property
def name(self) -> str:
return f"swap_{self.amount_in}{self.token_in}_to_{self.token_out}"
async def execute(self) -> Dict[str, Any]:
result = await self.client.swap_exact_amount_in(
pool_id=self.pool_id,
sender=self.sender,
token_in=self.token_in,
token_out=self.token_out,
amount_in=self.amount_in,
min_amount_out=self.min_amount_out,
)
self._tx_hash = result.get("tx_hash")
# 获取实际输出量
tx_detail = await self.client.query_tx(self._tx_hash)
self._actual_out = self._parse_swap_output(tx_detail)
return result
async def compensate(self) -> None:
"""补偿：反向交换（如果交换已确认）"""
if self._tx_hash and self._actual_out:
tx_status = await self.client.query_tx(self._tx_hash)
if tx_status and tx_status.get("code", 0) == 0:
await self.client.swap_exact_amount_in(
pool_id=self.pool_id,
sender=self.sender,
token_in=self.token_out,
token_out=self.token_in,
amount_in=self._actual_out,
min_amount_out="1", # 尽力返回
)
def _parse_swap_output(self, tx_detail: Dict) -> Optional[str]:
"""从交易详情中解析实际输出量"""
try:
events = tx_detail.get("events", [])
for event in events:
if event.get("type") == "token_swapped":
for attr in event.get("attributes", []):
if attr.get("key") == "tokens_out":
return attr.get("value").split(",")[0]
except Exception:
pass
return None
class ContractCallStep(SagaStep):
"""合约调用步骤"""
def __init__(
self,
client: Any,
contract_address: str,
method: str,
args: Dict[str, Any],
sender: str,
funds: Optional[str] = None,
):
self.client = client
self.contract_address = contract_address
self.method = method
self.args = args
self.sender = sender
self.funds = funds
self._tx_hash: Optional[str] = None
self._state_before: Optional[Dict] = None
@property
def name(self) -> str:
return f"contract_{self.contract_address[:8]}.{self.method}"
async def execute(self) -> Dict[str, Any]:
# 记录调用前的状态（如果合约提供查询接口）
self._state_before = await self._capture_state()
result = await self.client.execute_contract(
contract_address=self.contract_address,
method=self.method,
args=self.args,
sender=self.sender,
funds=self.funds,
)
self._tx_hash = result.get("tx_hash")
return result
async def compensate(self) -> None:
"""补偿：如果可能，恢复调用前的状态"""
if self._tx_hash and self._state_before:
tx_status = await self.client.query_tx(self._tx_hash)
if tx_status and tx_status.get("code", 0) == 0:
logger.info(
f"Contract call {self._tx_hash} confirmed, "
f"manual reversal may be needed for {self.contract_address}.{self.method}"
)
async def _capture_state(self) -> Optional[Dict]:
"""捕获调用前的相关状态"""
try:
return await self.client.query_contract_state(self.contract_address)
except Exception:
return None
7.3 两步提交
class TwoPhaseCommitStep:
"""两步提交（2PC）——准备阶段 + 提交/回滚阶段"""
def __init__(self, name: str, timeout: float = 30.0):
self.name = name
self.timeout = timeout
self._prepared = False
self._prepared_data: Optional[Dict[str, Any]] = None
async def prepare(self, *args, **kwargs) -> bool:
"""准备阶段——预留资源并验证"""
try:
result = await asyncio.wait_for(
self._do_prepare(*args, **kwargs),
timeout=self.timeout,
)
self._prepared = True
self._prepared_data = result
return True
except Exception as e:
logger.error(f"2PC prepare failed for '{self.name}': {e}")
return False
async def commit(self) -> bool:
"""提交阶段"""
if not self._prepared:
raise AgentError(
code="2PC_NOT_PREPARED",
message=f"Cannot commit '{self.name}' without prepare",
severity=Severity.ERROR,
retryable=False,
)
try:
await asyncio.wait_for(
self._do_commit(self._prepared_data),
timeout=self.timeout,
)
return True
except Exception as e:
logger.error(f"2PC commit failed for '{self.name}': {e}")
return False
async def rollback(self) -> bool:
"""回滚阶段——释放预留的资源"""
if not self._prepared:
return True
try:
await asyncio.wait_for(
self._do_rollback(self._prepared_data),
timeout=self.timeout,
)
return True
except Exception as e:
logger.error(f"2PC rollback failed for '{self.name}': {e}")
return False
async def _do_prepare(self, *args, **kwargs) -> Dict[str, Any]:
"""子类重写——实际准备逻辑"""
return {}
async def _do_commit(self, data: Optional[Dict]) -> None:
"""子类重写——实际提交逻辑"""
pass
async def _do_rollback(self, data: Optional[Dict]) -> None:
"""子类重写——实际回滚逻辑"""
pass
class ReserveGasStep(TwoPhaseCommitStep):
"""预留 Gas 费用的两步提交"""
def __init__(self, client: Any, address: str, gas_amount: str, denom: str):
super().__init__(name=f"reserve_gas_{address[:8]}", timeout=15.0)
self.client = client
self.address = address
self.gas_amount = gas_amount
self.denom = denom
async def _do_prepare(self) -> Dict[str, Any]:
balance = await self.client.query_balance(self.address, self.denom)
available = int(balance.get("amount", "0"))
required = int(self.gas_amount)
if available < required:
raise InsufficientFundsError(
address=self.address,
required=self.gas_amount,
available=balance.get("amount", "0"),
denom=self.denom,
)
return {
"balance_before": balance,
"reserved_amount": self.gas_amount,
}
async def _do_commit(self, data: Optional[Dict]) -> None:
logger.info(f"Gas reservation committed for {self.address}")
async def _do_rollback(self, data: Optional[Dict]) -> None:
logger.info(f"Gas reservation rolled back for {self.address}")
7.4 幂等性注册表
class IdempotencyRegistry:
"""幂等性注册表——防止补偿操作重复执行"""
def __init__(self, backend: Optional[Dict[str, Any]] = None):
self._store: Dict[str, Dict[str, Any]] = backend or {}
async def check_and_set(
self,
key: str,
ttl: float = 3600.0,
) -> Tuple[bool, Optional[Any]]:
"""检查键是否存在。如果不存在则设置并返回可执行。如果存在返回已有结果。"""
if key in self._store:
entry = self._store[key]
if time.time() - entry["timestamp"] < ttl:
return False, entry.get("result")
else:
del self._store[key]
self._store[key] = {
"timestamp": time.time(),
"status": "pending",
"result": None,
}
return True, None
async def complete(self, key: str, result: Any):
"""标记操作为完成"""
if key in self._store:
self._store[key]["status"] = "completed"
self._store[key]["result"] = result
self._store[key]["timestamp"] = time.time()
async def is_completed(self, key: str) -> bool:
entry = self._store.get(key)
return entry is not None and entry["status"] == "completed"
def clear_expired(self, max_age: float = 7200.0):
now = time.time()
expired = [
k for k, v in self._store.items()
if now - v["timestamp"] > max_age
]
for k in expired:
del self._store[k]
# 全局幂等性注册表
idempotency_registry = IdempotencyRegistry()
8. 健康检查与自愈
8.1 健康状态模型
@dataclass
class ServiceHealth:
"""单个服务的健康状态"""
service: str
healthy: bool
latency_ms: float
last_check: float = field(default_factory=time.time)
error: Optional[str] = None
details: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
return {
"service": self.service,
"healthy": self.healthy,
"latency_ms": self.latency_ms,
"last_check": self.last_check,
"error": self.error,
"details": self.details,
}
@dataclass
class HealthStatus:
"""聚合健康状态"""
chain: ServiceHealth
indexer: ServiceHealth
llm: ServiceHealth
a2a: ServiceHealth
timestamp: float = field(default_factory=time.time)
@property
def all_healthy(self) -> bool:
return all([
self.chain.healthy,
self.indexer.healthy,
self.llm.healthy,
self.a2a.healthy,
])
@property
def critical_healthy(self) -> bool:
"""核心服务（链+索引器）是否健康"""
return self.chain.healthy and self.indexer.healthy
@property
def healthy_count(self) -> int:
return sum([
self.chain.healthy,
self.indexer.healthy,
self.llm.healthy,
self.a2a.healthy,
])
def to_dict(self) -> Dict[str, Any]:
return {
"all_healthy": self.all_healthy,
"critical_healthy": self.critical_healthy,
"healthy_count": self.healthy_count,
"timestamp": self.timestamp,
"services": {
"chain": self.chain.to_dict(),
"indexer": self.indexer.to_dict(),
"llm": self.llm.to_dict(),
"a2a": self.a2a.to_dict(),
},
}
def __repr__(self) -> str:
status = "✅ ALL HEALTHY" if self.all_healthy else "⚠️ DEGRADED"
return (
f"HealthStatus({status}, "
f"chain={'✅' if self.chain.healthy else '❌'}, "
f"indexer={'✅' if self.indexer.healthy else '❌'}, "
f"llm={'✅' if self.llm.healthy else '❌'}, "
f"a2a={'✅' if self.a2a.healthy else '❌'})"
)
8.2 健康检查器
class HealthChecker:
"""健康检查器——定期检查所有依赖服务的状态"""
def __init__(
self,
rpc_endpoints: List[str],
indexer_endpoint: str,
llm_endpoints: Dict[str, str],
a2a_endpoints: Dict[str, str],
check_interval: float = 30.0,
):
self.rpc_endpoints = rpc_endpoints
self.indexer_endpoint = indexer_endpoint
self.llm_endpoints = llm_endpoints
self.a2a_endpoints = a2a_endpoints
self.check_interval = check_interval
self._last_status: Optional[HealthStatus] = None
async def check_chain(self) -> ServiceHealth:
"""检查 MSG Chain RPC 节点健康状态"""
start = time.time()
errors = []
for endpoint in self.rpc_endpoints:
try:
async with aiohttp.ClientSession() as session:
async with session.get(
f"{endpoint}/status",
timeout=aiohttp.ClientTimeout(total=5),
) as resp:
if resp.status == 200:
data = await resp.json()
latest_block_height = data.get("result", {}).get(
"sync_info", {}
).get("latest_block_height", "0")
latency = (time.time() - start) * 1000
return ServiceHealth(
service="chain",
healthy=True,
latency_ms=latency,
details={
"endpoint": endpoint,
"block_height": latest_block_height,
"catching_up": data.get("result", {}).get(
"sync_info", {}
).get("catching_up", False),
},
)
errors.append(f"{endpoint}: HTTP {resp.status}")
except Exception as e:
errors.append(f"{endpoint}: {e}")
latency = (time.time() - start) * 1000
return ServiceHealth(
service="chain",
healthy=False,
latency_ms=latency,
error="; ".join(errors),
)
async def check_indexer(self) -> ServiceHealth:
"""检查索引器健康状态"""
start = time.time()
try:
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.indexer_endpoint}/health",
timeout=aiohttp.ClientTimeout(total=10),
) as resp:
latency = (time.time() - start) * 1000
if resp.status == 200:
data = await resp.json()
return ServiceHealth(
service="indexer",
healthy=True,
latency_ms=latency,
details=data,
)
return ServiceHealth(
service="indexer",
healthy=False,
latency_ms=latency,
error=f"HTTP {resp.status}",
)
except Exception as e:
latency = (time.time() - start) * 1000
return ServiceHealth(
service="indexer",
healthy=False,
latency_ms=latency,
error=str(e),
)
async def check_llm(self) -> ServiceHealth:
"""检查 LLM 服务健康状态"""
start = time.time()
errors = []
for name, endpoint in self.llm_endpoints.items():
try:
async with aiohttp.ClientSession() as session:
async with session.get(
f"{endpoint}/health",
timeout=aiohttp.ClientTimeout(total=10),
) as resp:
if resp.status == 200:
latency = (time.time() - start) * 1000
return ServiceHealth(
service="llm",
healthy=True,
latency_ms=latency,
details={"provider": name, "endpoint": endpoint},
)
errors.append(f"{name}: HTTP {resp.status}")
except Exception as e:
errors.append(f"{name}: {e}")
latency = (time.time() - start) * 1000
return ServiceHealth(
service="llm",
healthy=False,
latency_ms=latency,
error="; ".join(errors),
)
async def check_a2a(self) -> ServiceHealth:
"""检查 Agent-to-Agent 通信服务健康状态"""
start = time.time()
errors = []
for name, endpoint in self.a2a_endpoints.items():
try:
async with aiohttp.ClientSession() as session:
async with session.get(
f"{endpoint}/ping",
timeout=aiohttp.ClientTimeout(total=5),
) as resp:
if resp.status == 200:
latency = (time.time() - start) * 1000
return ServiceHealth(
service="a2a",
healthy=True,
latency_ms=latency,
details={"relay": name},
)
errors.append(f"{name}: HTTP {resp.status}")
except Exception as e:
errors.append(f"{name}: {e}")
latency = (time.time() - start) * 1000
return ServiceHealth(
service="a2a",
healthy=False,
latency_ms=latency,
error="; ".join(errors),
)
async def check_all(self) -> HealthStatus:
"""并发检查所有服务"""
results = await asyncio.gather(
self.check_chain(),
self.check_indexer(),
self.check_llm(),
self.check_a2a(),
return_exceptions=True,
)
chain = results[0] if isinstance(results[0], ServiceHealth) else ServiceHealth(
service="chain", healthy=False, latency_ms=0, error=str(results[0])
)
indexer = results[1] if isinstance(results[1], ServiceHealth) else ServiceHealth(
service="indexer", healthy=False, latency_ms=0, error=str(results[1])
)
llm = results[2] if isinstance(results[2], ServiceHealth) else ServiceHealth(
service="llm", healthy=False, latency_ms=0, error=str(results[2])
)
a2a = results[3] if isinstance(results[3], ServiceHealth) else ServiceHealth(
service="a2a", healthy=False, latency_ms=0, error=str(results[3])
)
status = HealthStatus(chain=chain, indexer=indexer, llm=llm, a2a=a2a)
self._last_status = status
return status
def last_status(self) -> Optional[HealthStatus]:
return self._last_status
8.3 自动修复引擎
class AutoHealAction(ABC):
"""自愈操作基类"""
@abstractmethod
async def heal(self, service: str, details: Dict[str, Any]) -> bool:
pass
@property
@abstractmethod
def name(self) -> str:
pass
class RestartRPCConnection(AutoHealAction):
"""重启 RPC 连接池"""
def __init__(self, rpc_client: Any):
self.rpc_client = rpc_client
@property
def name(self) -> str:
return "restart_rpc_connection"
async def heal(self, service: str, details: Dict[str, Any]) -> bool:
if service != "chain":
return False
try:
await self.rpc_client.reconnect()
logger.info("RPC connection pool restarted")
return True
except Exception as e:
logger.error(f"Failed to restart RPC connection: {e}")
return False
class SwitchRPCEndpoint(AutoHealAction):
"""切换到备用 RPC 端点"""
def __init__(self, rpc_client: Any, fallback_endpoints: List[str]):
self.rpc_client = rpc_client
self.fallback_endpoints = fallback_endpoints
@property
def name(self) -> str:
return "switch_rpc_endpoint"
async def heal(self, service: str, details: Dict[str, Any]) -> bool:
if service != "chain":
return False
current = getattr(self.rpc_client, "endpoint", "")
for endpoint in self.fallback_endpoints:
if endpoint != current:
try:
await self.rpc_client.set_endpoint(endpoint)
logger.info(f"Switched RPC endpoint to {endpoint}")
return True
except Exception:
continue
return False
class ClearLLMCache(AutoHealAction):
"""清除 LLM 响应缓存"""
def __init__(self, cache: Dict[str, Any]):
self.cache = cache
@property
def name(self) -> str:
return "clear_llm_cache"
async def heal(self, service: str, details: Dict[str, Any]) -> bool:
if service != "llm":
return False
self.cache.clear()
logger.info("LLM cache cleared")
return True
class RetryIndexerConnection(AutoHealAction):
"""重试索引器连接"""
def __init__(self, indexer_client: Any):
self.indexer_client = indexer_client
@property
def name(self) -> str:
return "retry_indexer"
async def heal(self, service: str, details: Dict[str, Any]) -> bool:
if service != "indexer":
return False
try:
await self.indexer_client.reconnect()
logger.info("Indexer connection retried")
return True
except Exception as e:
logger.error(f"Indexer reconnect failed: {e}")
return False
class AutoHealEngine:
"""自动修复引擎——持续监控并修复服务"""
def __init__(
self,
checker: HealthChecker,
max_repair_attempts: int = 3,
cooldown_seconds: float = 60.0,
):
self.checker = checker
self.max_repair_attempts = max_repair_attempts
self.cooldown_seconds = cooldown_seconds
self.actions: List[AutoHealAction] = []
self._repair_counts: Dict[str, int] = defaultdict(int)
self._last_repair: Dict[str, float] = defaultdict(float)
self._is_running = False
def add_action(self, action: AutoHealAction):
self.actions.append(action)
async def heal(self, status: HealthStatus) -> Dict[str, bool]:
"""尝试修复不健康的服务"""
results: Dict[str, bool] = {}
for service_health in [status.chain, status.indexer, status.llm, status.a2a]:
if service_health.healthy:
continue
service = service_health.service
now = time.time()
# 检查冷却期
if now - self._last_repair[service] < self.cooldown_seconds:
logger.info(f"Service '{service}' in cooldown, skipping repair")
results[service] = False
continue
# 检查修复次数
if self._repair_counts[service] >= self.max_repair_attempts:
logger.warning(
f"Service '{service}' exceeded max repair attempts "
f"({self.max_repair_attempts})"
)
# 通知人工介入
await self._notify_human_intervention(service)
results[service] = False
continue
# 按顺序尝试修复操作
for action in self.actions:
try:
success = await action.heal(service, service_health.details)
if success:
self._repair_counts[service] += 1
self._last_repair[service] = now
results[service] = True
logger.info(
f"Repair '{action.name}' succeeded for {service}"
)
break
except Exception as e:
logger.error(f"Repair '{action.name}' failed for {service}: {e}")
else:
results[service] = False
self._repair_counts[service] += 1
self._last_repair[service] = now
return results
async def run_periodically(self, interval: Optional[float] = None):
"""定期执行健康检查和自动修复"""
self._is_running = True
interval = interval or self.checker.check_interval
while self._is_running:
try:
status = await self.checker.check_all()
logger.info(f"Health check: {status}")
if not status.all_healthy:
results = await self.heal(status)
logger.info(f"Auto-heal results: {results}")
if not results:
# 所有修复尝试失败——触发降级
degraded_services = [
s.service
for s in [status.chain, status.indexer, status.llm, status.a2a]
if not s.healthy
]
logger.warning(
f"Auto-heal failed for: {degraded_services}. "
f"Triggering degradation."
)
except Exception as e:
logger.error(f"Health check cycle failed: {e}")
await asyncio.sleep(interval)
def stop(self):
self._is_running = False
async def _notify_human_intervention(self, service: str):
"""通知人工介入——发送告警"""
logger.critical(
f"HUMAN INTERVENTION REQUIRED: Service '{service}' "
f"failed after {self.max_repair_attempts} auto-repair attempts"
)
8.4 链上心跳
class OnChainHeartbeat:
"""链上心跳——定期在 MSG Chain 上记录 Agent 活跃状态"""
def __init__(
self,
client: Any,
agent_address: str,
contract_address: str,
interval_blocks: int = 100,
):
self.client = client
self.agent_address = agent_address
self.contract_address = contract_address
self.interval_blocks = interval_blocks
self._last_heartbeat_block = 0
self._is_running = False
async def beat(self) -> Dict[str, Any]:
"""发送心跳交易"""
current_block = await self._get_current_block()
if current_block - self._last_heartbeat_block < self.interval_blocks:
return {"skipped": True, "block": current_block}
try:
result = await self.client.execute_contract(
contract_address=self.contract_address,
method="heartbeat",
args={
"agent": self.agent_address,
"timestamp": int(time.time()),
"block_height": current_block,
"version": "1.0.0",
},
sender=self.agent_address,
)
self._last_heartbeat_block = current_block
return {
"sent": True,
"block": current_block,
"tx_hash": result.get("tx_hash"),
}
except Exception as e:
logger.error(f"Heartbeat failed: {e}")
return {"sent": False, "block": current_block, "error": str(e)}
async def run_periodically(self):
"""定期发送心跳"""
self._is_running = True
while self._is_running:
try:
await self.beat()
except Exception as e:
logger.error(f"Heartbeat cycle error: {e}")
await asyncio.sleep(self.interval_blocks * 6) # ~6s per MSG block
def stop(self):
self._is_running = False
async def _get_current_block(self) -> int:
result = await self.client.query_latest_block()
return int(result.get("block", {}).get("header", {}).get("height", 0))
9. 监控与告警
9.1 错误率指标
class ErrorMetricsCollector:
"""错误指标收集器——按类型、严重级别、服务统计错误率"""
def __init__(self, window_seconds: float = 300.0):
self.window_seconds = window_seconds
self._errors: List[Dict[str, Any]] = []
self._error_counts: Dict[str, int] = defaultdict(int)
self._severity_counts: Dict[str, int] = defaultdict(int)
self._service_counts: Dict[str, int] = defaultdict(int)
self._last_cleanup = time.time()
def record(self, error: AgentError):
"""记录一个错误"""
now = time.time()
self._cleanup(now)
entry = {
"code": error.code,
"severity": error.severity.name,
"retryable": error.retryable,
"timestamp": error.timestamp,
"message": str(error)[:200],
}
self._errors.append(entry)
self._error_counts[error.code] += 1
self._severity_counts[error.severity.name] += 1
# 尝试推断服务类型
if isinstance(error, ChainError):
self._service_counts["chain"] += 1
elif isinstance(error, ContractError):
self._service_counts["contract"] += 1
elif isinstance(error, LLMError):
self._service_counts["llm"] += 1
elif isinstance(error, NetworkError):
self._service_counts["network"] += 1
elif isinstance(error, PaymentError):
self._service_counts["payment"] += 1
else:
self._service_counts["unknown"] += 1
def error_rate(self, code: Optional[str] = None) -> float:
"""计算错误率（每秒）"""
self._cleanup(time.time())
total = self._error_counts.get(code, 0) if code else len(self._errors)
return total / max(self.window_seconds, 1)
def top_errors(self, n: int = 5) -> List[Dict[str, Any]]:
"""返回最常见的 n 个错误"""
sorted_codes = sorted(
self._error_counts.items(),
key=lambda x: x[1],
reverse=True,
)[:n]
return [
{"code": code, "count": count}
for code, count in sorted_codes
]
def snapshot(self) -> Dict[str, Any]:
"""当前快照"""
self._cleanup(time.time())
return {
"window_seconds": self.window_seconds,
"total_errors": len(self._errors),
"error_rate": self.error_rate(),
"by_code": dict(self._error_counts),
"by_severity": dict(self._severity_counts),
"by_service": dict(self._service_counts),
"top_errors": self.top_errors(),
}
def _cleanup(self, now: float):
"""清理过期记录"""
cutoff = now - self.window_seconds
self._errors = [e for e in self._errors if e["timestamp"] > cutoff]
# 重建计数
self._error_counts.clear()
self._severity_counts.clear()
self._service_counts.clear()
for e in self._errors:
self._error_counts[e["code"]] += 1
self._severity_counts[e["severity"]] += 1
# 全局错误指标收集器
error_metrics = ErrorMetricsCollector()
9.2 告警阈值与触发器
class AlertSeverity(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class AlertRule:
"""告警规则"""
name: str
description: str
condition: Callable[[ErrorMetricsCollector], bool]
severity: AlertSeverity
cooldown_seconds: float = 300.0
last_triggered: float = 0.0
def should_trigger(self, metrics: ErrorMetricsCollector) -> bool:
now = time.time()
if now - self.last_triggered < self.cooldown_seconds:
return False
try:
if self.condition(metrics):
self.last_triggered = now
return True
except Exception:
pass
return False
@dataclass
class Alert:
"""告警实例"""
rule_name: str
severity: AlertSeverity
message: str
timestamp: float = field(default_factory=time.time)
metrics_snapshot: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
return {
"rule_name": self.rule_name,
"severity": self.severity.value,
"message": self.message,
"timestamp": self.timestamp,
}
class AlertManager:
"""告警管理器——评估规则并发送告警"""
def __init__(self, metrics: ErrorMetricsCollector):
self.metrics = metrics
self.rules: List[AlertRule] = []
self.alerts: List[Alert] = []
self._alert_handlers: List[Callable[[Alert], Awaitable[None]]] = []
def add_rule(self, rule: AlertRule):
self.rules.append(rule)
def add_handler(self, handler: Callable[[Alert], Awaitable[None]]):
self._alert_handlers.append(handler)
async def evaluate(self) -> List[Alert]:
"""评估所有告警规则"""
triggered: List[Alert] = []
for rule in self.rules:
if rule.should_trigger(self.metrics):
alert = Alert(
rule_name=rule.name,
severity=rule.severity,
message=rule.description,
metrics_snapshot=self.metrics.snapshot(),
)
self.alerts.append(alert)
triggered.append(alert)
logger.warning(
f"ALERT [{rule.severity.value.upper()}] {rule.name}: "
f"{rule.description}"
)
# 触发告警处理器
for handler in self._alert_handlers:
try:
await handler(alert)
except Exception as e:
logger.error(f"Alert handler failed: {e}")
return triggered
def recent_alerts(self, n: int = 10) -> List[Alert]:
return self.alerts[-n:]
# 预定义的告警规则
def high_error_rate_rule(threshold: float = 0.5) -> AlertRule:
"""错误率过高的告警规则"""
return AlertRule(
name="high_error_rate",
description=f"Error rate exceeds {threshold}/s over 5min window",
condition=lambda m: m.error_rate() > threshold,
severity=AlertSeverity.HIGH,
)
def critical_error_rule() -> AlertRule:
"""严重错误出现的告警规则"""
return AlertRule(
name="critical_error_detected",
description="Critical severity error detected",
condition=lambda m: m._severity_counts.get("CRITICAL", 0) > 0,
severity=AlertSeverity.CRITICAL,
)
def chain_disconnected_rule() -> AlertRule:
"""链断连的告警规则"""
return AlertRule(
name="chain_disconnected",
description="Chain RPC connection errors detected",
condition=lambda m: m._service_counts.get("chain", 0) >= 3,
severity=AlertSeverity.HIGH,
)
def llm_degradation_rule() -> AlertRule:
"""LLM 服务降级的告警规则"""
return AlertRule(
name="llm_degradation",
description="LLM service errors detected (rate limit/timeout)",
condition=lambda m: m._service_counts.get("llm", 0) >= 5,
severity=AlertSeverity.MEDIUM,
)
9.3 链上错误报告
class OnChainErrorReporter:
"""链上错误报告——将关键错误记录到 MSG Chain 上"""
def __init__(
self,
client: Any,
reporter_address: str,
reporting_contract: str,
min_severity: Severity = Severity.ERROR,
):
self.client = client
self.reporter_address = reporter_address
self.reporting_contract = reporting_contract
self.min_severity = min_severity
self._rate_limiter = RetryBudget(max_retries=10, window_seconds=60.0)
async def report(self, error: AgentError) -> Optional[str]:
"""报告错误到链上"""
if error.severity.value < self.min_severity.value:
return None
if self._rate_limiter.is_exhausted():
logger.warning("On-chain error report rate limit exceeded")
return None
self._rate_limiter.record_retry()
try:
result = await self.client.execute_contract(
contract_address=self.reporting_contract,
method="report_error",
args={
"error_code": error.code,
"severity": error.severity.name,
"message": str(error)[:500],
"timestamp": int(error.timestamp),
"retryable": error.retryable,
"trace": error.trace[:2000] if error.trace else "",
},
sender=self.reporter_address,
)
tx_hash = result.get("tx_hash")
logger.info(f"Error reported on-chain: {error.code} tx={tx_hash}")
return tx_hash
except Exception as e:
logger.error(f"Failed to report error on-chain: {e}")
return None
async def report_batch(self, errors: List[AgentError]) -> List[str]:
"""批量报告错误"""
txs = []
for error in errors[:5]: # 每批最多 5 个
tx = await self.report(error)
if tx:
txs.append(tx)
await asyncio.sleep(1) # 避免速率限制
return txs
9.4 结构化日志
class ResilienceLogger:
"""弹性工程的结构化日志记录器"""
def __init__(self, agent_id: str, log_dir: str = "/var/log/msg-agent"):
self.agent_id = agent_id
self.log_dir = log_dir
os.makedirs(log_dir, exist_ok=True)
self._log_file = os.path.join(log_dir, f"{agent_id}.jsonl")
def log_retry(self, operation: str, attempt: int, delay: float, error: str):
entry = {
"type": "retry",
"timestamp": time.time(),
"agent_id": self.agent_id,
"operation": operation,
"attempt": attempt,
"delay": delay,
"error": error,
}
self._write(entry)
def log_circuit_break(
self, breaker: str, state: str, failure_count: int
):
entry = {
"type": "circuit_breaker",
"timestamp": time.time(),
"agent_id": self.agent_id,
"breaker": breaker,
"state": state,
"failure_count": failure_count,
}
self._write(entry)
def log_degradation(
self, from_level: str, to_level: str, reason: str
):
entry = {
"type": "degradation",
"timestamp": time.time(),
"agent_id": self.agent_id,
"from_level": from_level,
"to_level": to_level,
"reason": reason,
}
self._write(entry)
def log_saga(self, saga_id: str, status: str, steps: int, error: str = ""):
entry = {
"type": "saga",
"timestamp": time.time(),
"agent_id": self.agent_id,
"saga_id": saga_id,
"status": status,
"steps": steps,
"error": error,
}
self._write(entry)
def log_health(self, status: HealthStatus):
entry = {
"type": "health",
"timestamp": time.time(),
"agent_id": self.agent_id,
"all_healthy": status.all_healthy,
"services": {
"chain": status.chain.healthy,
"indexer": status.indexer.healthy,
"llm": status.llm.healthy,
"a2a": status.a2a.healthy,
},
}
self._write(entry)
def _write(self, entry: Dict[str, Any]):
try:
with open(self._log_file, "a") as f:
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
except Exception as e:
logger.error(f"Failed to write resilience log: {e}")
附录
A. 快速参考：错误代码速查表
| 错误代码 | 类型 | 严重度 | 可重试 | 说明 |
|---|---|---|---|---|
TX_REVERT |
ChainError | ERROR | ❌ | 交易被链回滚 |
TX_TIMEOUT |
ChainError | ERROR | ✅ | 交易超时未确认 |
TX_SEQUENCE |
ChainError | WARNING | ✅ | Nonce 不匹配 |
GAS_ESTIMATION |
ChainError | ERROR | ✅ | Gas 估算不足 |
OUT_OF_GAS |
ChainError | ERROR | ❌ | Gas 耗尽 |
INSUFFICIENT_FUNDS |
ChainError | ERROR | ❌ | 余额不足 |
IBC_TRANSFER |
ChainError | ERROR | ✅ | IBC 转账失败 |
CONTRACT_REVERT |
ContractError | ERROR | ❌ | 合约 revert |
CONTRACT_NOT_FOUND |
ContractError | ERROR | ❌ | 合约未找到 |
SLIPPAGE_EXCEEDED |
ContractError | ERROR | ❌ | 滑点超出限制 |
LLM_RATE_LIMIT |
LLMError | WARNING | ✅ | LLM 速率限制 |
LLM_TIMEOUT |
LLMError | WARNING | ✅ | LLM 超时 |
LLM_PARSE_ERROR |
LLMError | WARNING | ✅ | 响应解析失败 |
LLM_CONTENT_FILTER |
LLMError | ERROR | ❌ | 内容过滤器触发 |
RPC_CONNECTION |
NetworkError | WARNING | ✅ | RPC 连接失败 |
RPC_NODE_UNHEALTHY |
NetworkError | WARNING | ✅ | RPC 节点不健康 |
WS_DISCONNECT |
NetworkError | WARNING | ✅ | WebSocket 断连 |
IBC_RELAYER |
NetworkError | ERROR | ✅ | IBC 中继器错误 |
BALANCE_INSUFFICIENT |
PaymentError | ERROR | ❌ | 支付余额不足 |
MAX_FEE_EXCEEDED |
PaymentError | WARNING | ✅ | 费用超出限制 |
CIRCUIT_BREAKER_OPEN |
NonRetryableError | WARNING | ❌ | 熔断器开启 |
MAX_RETRIES |
NonRetryableError | ERROR | ❌ | 重试耗尽 |
SAGA_FAILED |
NonRetryableError | CRITICAL | ❌ | Saga 执行失败 |
B. 配置示例
# msg_agent_resilience_config.py
# MSG Chain Agent 弹性工程配置模板
RESILIENCE_CONFIG = {
"agent": {
"id": "msg-agent-001",
"address_prefix": "msg",
"chain_id": "msg-chain-1",
},
"retry": {
"global_budget": {
"max_retries": 100,
"window_seconds": 300,
},
"policies": {
"chain_tx": {
"max_retries": 5,
"base_delay": 0.5,
"max_delay": 30.0,
"backoff_factor": 2.0,
"jitter_factor": 0.2,
},
"rpc_call": {
"max_retries": 3,
"base_delay": 0.2,
"max_delay": 5.0,
},
"llm_inference": {
"max_retries": 3,
"base_delay": 2.0,
"max_delay": 30.0,
"backoff_factor": 3.0,
},
},
},
"circuit_breakers": {
"failure_threshold": 5,
"recovery_timeout": 30.0,
"half_open_max_requests": 1,
"consecutive_successes_to_close": 2,
},
"timeouts": {
"rpc_call": 10.0,
"tx_broadcast": 60.0,
"tx_confirmation": 120.0,
"llm_inference": 30.0,
"indexer_query": 15.0,
"a2a_message": 20.0,
"ibc_ack": 300.0,
},
"degradation": {
"failure_thresholds": {
"llm": 3,
"indexer": 5,
"chain": 10,
"a2a": 3,
"oracle": 5,
"ibc": 3,
},
},
"health_check": {
"interval_seconds": 30.0,
"max_repair_attempts": 3,
"cooldown_seconds": 60.0,
},
"monitoring": {
"error_window_seconds": 300.0,
"alert_rules": {
"high_error_rate": {"threshold": 0.5, "severity": "high"},
"chain_disconnected": {"min_errors": 3, "severity": "high"},
"llm_degradation": {"min_errors": 5, "severity": "medium"},
},
},
}
C. 典型故障场景与处理流程
C.1 RPC 节点故障
1. Agent 广播交易
2. RPC 连接超时 → RPCConnectionError
3. ErrorRouter 识别为可重试错误
4. RetryStrategy 切换备用 RPC 端点
5. CircuitBreaker 记录失败
6. 连续 5 次失败 → 熔断器开启
7. AutoHealEngine 尝试修复
8. 修复失败 → DegradationManager 降级到 REDUCED
9. 触发告警: HIGH - chain_disconnected
C.2 交易回滚
1. Agent 提交 Swap 交易
2. 交易被回滚 → TxRevertError(revert_reason="insufficient_liquidity")
3. ErrorRouter 标记为不可重试（PERSISTENT）
4. CompensationManager 回滚 Sag 中的前置步骤
5. 记录错误到 OnChainErrorReporter
6. 更新 ErrorMetrics
7. 评估告警规则
C.3 LLM 推理超时
1. Agent 调用 LLM 推理
2. 等待 30s → LLMTimeoutError
3. RetryStrategy 尝试退避重试（LLM 策略）
4. 3 次重试都失败 → MaxRetriesExceeded
5. FallbackStrategy 切换到备用模型
6. 备用模型成功 → FallbackResult(source="gpt-3.5-turbo")
7. CircuitBreaker 记录服务降级
8. 如果所有模型失败 → 触发 DegradationManager 降级
C.4 跨链 IBC 超时
1. Agent 发起 IBC 转账
2. 等待 300s 未收到确认 → IBCRelayerError
3. IdempotentRetryStrategy 检查幂等性键
4. 已存在的键 → 查询链上状态确认
5. 状态为 "pending" → 继续等待
6. 状态为 "failed" → 触发补偿操作
7. CompensationManager 发起反向 IBC 转账
8. 补偿失败 → CRITICAL 告警，人工介入
文档版本: v1.0.0
维护者: MSG Chain Engineering Team
许可: MIT
本文档内容基于 MSGChain 代码库真实状态编写，非 AI 自动生成。
主网状态: No-Go | 白皮书: https://msgchain.org/whitepaper/