企业官网建设流程全解析

# 反面教材，别这么干 for i in range(3): try: result = call_api() break except Exception: time.sleep(3)

import asyncio import random from dataclasses import dataclass from typing import TypeVar, Callable, Optional, Tuple

T = TypeVar(‘T’)

@dataclass
class RetryConfig:
max_retries: int = 3
base_delay: float = 1.0
max_delay: float = 60.0
exponential_base: float = 2.0
jitter: bool = True
# 关键：只重试这些异常类型
retryable_exceptions: tuple = (TimeoutError, ConnectionError,
ConnectionRefusedError)

async def smart_retry(
func: Callable[…, T],
config: RetryConfig = RetryConfig(),
*args, **kwargs
) -> T:
"""带失败分类的智能重试"""
last_exception = None

for attempt in range(config.max_retries + 1): try: return await func(*args, **kwargs) except config.retryable_exceptions as e: # 基础设施错误 -> 指数退避重试 last_exception = e if attempt == config.max_retries: break delay = min( config.base_delay * (config.exponential_base ** attempt), config.max_delay ) if config.jitter: delay = delay * (0.5 + random.random()) print(f"[重试] 第{attempt+1}次失败: {e}，" f"{delay:.1f}秒后重试...") await asyncio.sleep(delay) except ValueError as e: # 语义错误 -> 直接抛给上层处理，不重试 print(f"[跳过重试] 语义错误: {e}") raise raise last_exception

这里有几个细节值得注意：

1. retryable_exceptions白名单：只重试网络层面的错误。如果API返回了200但内容是空的，那不是重试能解决的问题。
2. jitter（随机抖动）：生产环境多个实例同时重试时，固定间隔会导致它们永远同步。加上随机因子可以打散。
3. 语义错误不重试：ValueError、KeyError这类错误说明逻辑有问题，重试只会浪费钱。

什么时候用、什么时候不用

适合：调用外部API、数据库连接、模型推理接口等可能因临时故障失败的场景。
不适合：工具返回空结果、模型输出格式错误、业务逻辑异常——这些需要换策略，不是等一等就能好的。

三、模式二：断路器——该断就断，别当老好人

为什么需要断路器

假设你的Agent依赖三个外部工具：向量数据库、OCR服务、邮件API。某天OCR服务挂了，你的Agent会怎么做？

如果没有断路器，它会一次又一次调用OCR，每次都等超时，每次都失败。更糟的是，如果有10个并发任务都在调OCR，你的Agent会同时发出10个注定失败的请求，白白消耗资源。

断路器就是用来解决这个问题的：当某个服务连续失败超过阈值时，直接"断开"，后续请求不再发出去，立即返回降级结果。等一段时间后，再试探性地发一个请求看看服务恢复了没有。

完整实现

from datetime import datetime, timedelta from enum import Enum from dataclasses import dataclass, field import logging

logger = logging.getLogger(name)

class CircuitState(Enum):
CLOSED = "closed" # 正常，请求通过
OPEN = "open" # 断开，请求直接走降级
HALF_OPEN = "half_open" # 试探，放一个请求过去试试

@dataclass
class CircuitBreaker:
name: str
failure_threshold: int = 5 # 连续失败几次后断开
recovery_timeout: int = 30 # 断开后多少秒尝试恢复
half_open_max_calls: int = 1 # 半开状态最多放几个请求

state: CircuitState = CircuitState.CLOSED failure_count: int = 0 success_count: int = 0 last_failure_time: Optional[datetime] = None half_open_calls: int = 0 def can_execute(self) -> bool: if self.state == CircuitState.CLOSED: return True if self.state == CircuitState.OPEN: if self._recovery_expired(): self.state = CircuitState.HALF_OPEN self.half_open_calls = 0 logger.info(f"[{self.name}] 断路器进入半开状态") return True return False if self.state == CircuitState.HALF_OPEN: return self.half_open_calls < self.half_open_max_calls return False def record_success(self): if self.state == CircuitState.HALF_OPEN: self.success_count += 1 if self.success_count >= self.half_open_max_calls: self._reset() logger.info(f"[{self.name}] 断路器关闭，服务已恢复") else: self.failure_count = 0 def record_failure(self): self.failure_count += 1 self.last_failure_time = datetime.now() if self.state == CircuitState.HALF_OPEN: self._trip() elif self.failure_count >= self.failure_threshold: self._trip() def _trip(self): self.state = CircuitState.OPEN logger.warning( f"[{self.name}] 断路器打开！" f"连续失败{self.failure_count}次，" f"{self.recovery_timeout}秒后尝试恢复" ) def _reset(self): self.state = CircuitState.CLOSED self.failure_count = 0 self.success_count = 0 self.half_open_calls = 0 def _recovery_expired(self) -> bool: if not self.last_failure_time: return True return (datetime.now() - self.last_failure_time > timedelta(seconds=self.recovery_timeout))

用法很简单，在调用工具之前检查断路器状态：

# 每个外部工具配一个断路器 breakers = { "ocr": CircuitBreaker("OCR服务", failure_threshold=3), "vector_db": CircuitBreaker("向量数据库", failure_threshold=5), "email": CircuitBreaker("邮件API", failure_threshold=3), }

async def call_tool_with_breaker(tool_name: str, func, *args, **kwargs):
breaker = breakers[tool_name]

if not breaker.can_execute(): # 断路器开着，直接走降级 logger.warning(f"[{tool_name}] 断路器打开，执行降级逻辑") return get_fallback(tool_name) try: result = await func(*args, **kwargs) breaker.record_success() return result except Exception as e: breaker.record_failure() raise

一个容易踩的坑

断路器的recovery_timeout不要设太短。我见过有人设成5秒——服务挂了之后每5秒放一个请求过去，结果服务本来都快恢复了，被这几发试探请求直接打崩。生产环境建议至少30秒，慢依赖可以设到60-120秒。

四、模式三：检查点与恢复——干到一半崩了，不用从头来

为什么Agent特别需要检查点

传统程序崩了就重启，从头跑一遍也没什么。但Agent不一样——一次Agent执行可能涉及几十次LLM调用，每次调用都在烧钱。如果一个提取200个字段的Agent在第150个字段时崩了，你让它从第1个字段重新开始？那前149个字段调用LLM的钱就白花了。

LangGraph从1.2版本开始原生支持Checkpointer机制，在每个super-step自动保存状态快照。配合PostgresSaver可以实现持久化。但即使不用LangGraph，你也可以自己实现一套轻量级的检查点逻辑。

实战代码

import json import os from dataclasses import dataclass, field, asdict from typing import Dict, Any, Optional, List from pathlib import Path

@dataclass
class Checkpoint:
"""检查点数据结构"""
task_id: str
current_step: int
total_steps: int
completed_fields: Dict[str, Any] = field(default_factory=dict)
last_error: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)

class AgentCheckpointManager:
"""轻量级Agent检查点管理器"""

def __init__(self, checkpoint_dir: str = "./checkpoints"): self.checkpoint_dir = Path(checkpoint_dir) self.checkpoint_dir.mkdir(parents=True, exist_ok=True) def save(self, checkpoint: Checkpoint): """保存检查点""" path = self.checkpoint_dir / f"{checkpoint.task_id}.json" data = asdict(checkpoint) path.write_text(json.dumps(data, ensure_ascii=False, indent=2)) def load(self, task_id: str) -> Optional[Checkpoint]: """加载最近的检查点""" path = self.checkpoint_dir / f"{task_id}.json" if not path.exists(): return None data = json.loads(path.read_text()) return Checkpoint(**data) def cleanup(self, task_id: str): """任务完成后清理检查点""" path = self.checkpoint_dir / f"{task_id}.json" if path.exists(): path.unlink()

============ 使用示例 ============

async def extract_fields_with_checkpoint(
documents: List[dict],
fields_to_extract: List[str],
task_id: str
):
"""
从文档中提取字段，支持检查点恢复。
假设某个字段提取到一半崩了，下次可以从断点继续。
"""
manager = AgentCheckpointManager()

# 尝试加载已有检查点 checkpoint = manager.load(task_id) if checkpoint: print(f"[恢复] 从第{checkpoint.current_step}步继续，" f"已完成{len(checkpoint.completed_fields)}个字段") else: checkpoint = Checkpoint( task_id=task_id, current_step=0, total_steps=len(fields_to_extract) ) for i, field_name in enumerate(fields_to_extract): # 跳过已完成的字段 if field_name in checkpoint.completed_fields: continue try: # 这里替换成实际的LLM提取逻辑 value = await extract_single_field(documents, field_name) checkpoint.completed_fields[field_name] = value checkpoint.current_step = i + 1 # 每完成一个字段就保存检查点 manager.save(checkpoint) print(f"[进度] {i+1}/{len(fields_to_extract)}: " f"{field_name} = {value}") except Exception as e: checkpoint.last_error = str(e) manager.save(checkpoint) # 保存错误状态 print(f"[错误] 第{i+1}步失败: {e}") raise # 全部完成，清理检查点 manager.cleanup(task_id) return checkpoint.completed_fields

这段代码的核心思路：每完成一个最小工作单元就保存一次状态。崩溃恢复时，读检查点跳过已完成的部分，从断点继续。

在实际项目中，这个检查点可以存在Postgres里（LangGraph的PostgresSaver就是这么干的），也可以用Redis。如果任务涉及多个Agent并发写入同一个状态，还需要考虑冲突解决策略——一般用last-write-wins或者版本号机制就够了。

五、模式四：降级兜底——最后的防线

为什么要设计"体面的失败"

前面三种模式都在努力让Agent继续工作。但有些情况下，继续工作是不可能的——所有重试都失败了，断路器也开了，检查点恢复也没用。这时候怎么办？

最差的方案是让Agent硬编一个答案出来。这时候它大概率会幻觉，给你一个看起来很对但完全错误的答案。

正确的做法是设计一个明确的降级路径：告诉用户"我做不到"，同时给出一个有限的、但可靠的替代方案。

代码实现

from typing import Any, Optional, Callable import logging

logger = logging.getLogger(name)

class DegradationPolicy:
"""降级策略管理器"""

@staticmethod async def execute_with_fallback( primary: Callable, fallbacks: list, safe_default: Any, context: dict = None ) -> dict: """ 按优先级依次尝试多个策略，全部失败则返回安全默认值。 Args: primary: 主策略函数 fallbacks: 降级策略列表，按优先级排序 safe_default: 所有策略都失败时的安全默认值 context: 执行上下文信息 """ errors = [] # 尝试主策略 try: result = await primary(context) if result and not _is_empty_result(result): return {"status": "success", "data": result, "source": "primary"} except Exception as e: errors.append(f"primary: {e}") logger.warning(f"主策略失败: {e}") # 依次尝试降级策略 for i, fallback in enumerate(fallbacks): try: result = await fallback(context) if result and not _is_empty_result(result): return {"status": "degraded", "data": result, "source": f"fallback_{i+1}"} except Exception as e: errors.append(f"fallback_{i+1}: {e}") logger.warning(f"降级策略{i+1}失败: {e}") # 全部失败 -> 返回安全默认值，而不是让Agent编一个 logger.error(f"所有策略失败，返回安全默认值。错误链: {errors}") return { "status": "failed", "data": safe_default, "source": "safe_default", "errors": errors }

def _is_empty_result(result) -> bool:
"""判断结果是否实质为空"""
if result is None:
return True
if isinstance(result, (list, dict, str)) and len(result) == 0:
return True
return False

============ 实际使用 ============

async def answer_user_question(query: str):
"""用户提问的完整降级链"""

async def primary_search(ctx): # 主策略：向量数据库检索 + LLM生成 docs = await vector_db.search(ctx["query"], top_k=5) return await llm.generate(ctx["query"], context=docs) async def fallback_cached(ctx): # 降级1：用本地缓存的相似问答 cached = await cache.get_similar(ctx["query"], threshold=0.85) if cached: return cached.answer return None async def fallback_keyword(ctx): # 降级2：用关键词匹配兜底 keyword_result = await keyword_search(ctx["query"]) if keyword_result: return f"找到以下相关内容：{keyword_result}" return None result = await DegradationPolicy.execute_with_fallback( primary=lambda ctx: primary_search(ctx), fallbacks=[ lambda ctx: fallback_cached(ctx), lambda ctx: fallback_keyword(ctx), ], safe_default="抱歉，当前无法获取准确信息，请稍后再试。", context={"query": query} ) # 根据来源做不同处理 if result["status"] == "success": return result["data"] elif result["status"] == "degraded": # 降级结果要标注来源，让用户知道这不是最佳答案 return f"[结果可能不完整] {result['data']}" else: return result["data"]

降级设计的关键原则

1. 安全默认值绝对不能是Agent生成的。必须是硬编码的、经过人工审核的文案。让一个正在"幻觉"的Agent来生成"我做不到"的回答，这本身就很有讽刺意味。

2. 降级结果要标注来源。用户有权知道自己拿到的是最佳答案还是退而求其次的结果。

3. 错误链要记录。出了问题你得知道是哪一层先挂的，不然排查起来跟猜谜一样。

在之前的项目中，黑箭科技的工程团队在做Agent可靠性测试时总结了一个经验：降级策略的设计应该跟功能开发同步进行，而不是上线前补一个try-catch了事。每个工具调用在写主逻辑的时候，就应该想好"这个工具挂了，我拿什么顶上"。

六、四种模式的协同：一个完整的容错Agent

单独用每种模式都不够，真正靠谱的做法是把它们组合起来。下面是一个把四种模式整合到一起的Agent执行框架：

import asyncio import logging from dataclasses import dataclass from typing import Dict, Any

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(name)

@dataclass
class AgentExecutionConfig:
"""Agent执行配置"""
max_tool_calls: int = 50 # 单次执行最大工具调用次数
max_self_reflections: int = 3 # 最大自我反思次数
token_budget: int = 100000 # 单次执行token预算
max_consecutive_failures: int = 5 # 最大连续失败次数

class ResilientAgent:
"""容错Agent执行框架"""

def __init__(self, config: AgentExecutionConfig = None): self.config = config or AgentExecutionConfig() self.breakers: Dict[str, CircuitBreaker] = {} self.checkpoint_mgr = AgentCheckpointManager() # 运行时统计 self.tool_call_count = 0 self.reflection_count = 0 self.token_used = 0 self.consecutive_failures = 0 def get_or_create_breaker(self, tool_name: str) -> CircuitBreaker: if tool_name not in self.breakers: self.breakers[tool_name] = CircuitBreaker( name=tool_name, failure_threshold=3, recovery_timeout=60 ) return self.breakers[tool_name] def _check_budget(self): """检查是否超出预算""" if self.tool_call_count >= self.config.max_tool_calls: raise RuntimeError( f"工具调用次数超限: {self.tool_call_count}" f"/{self.config.max_tool_calls}" ) if self.token_used >= self.config.token_budget: raise RuntimeError( f"Token预算超限: {self.token_used}" f"/{self.config.token_budget}" ) if self.consecutive_failures >= self.config.max_consecutive_failures: raise RuntimeError( f"连续失败超限: {self.consecutive_failures}" f"/{self.config.max_consecutive_failures}" ) async def call_tool(self, tool_name: str, func, *args, **kwargs): """带全套容错的工具调用""" self._check_budget() self.tool_call_count += 1 breaker = self.get_or_create_breaker(tool_name) # 1. 检查断路器 if not breaker.can_execute(): logger.warning(f"[{tool_name}] 断路器打开，执行降级") return self._get_tool_fallback(tool_name) # 2. 带重试的执行 retry_config = RetryConfig( max_retries=2, base_delay=1.0, retryable_exceptions=(TimeoutError, ConnectionError) ) try: result = await smart_retry(func, retry_config, *args, **kwargs) breaker.record_success() self.consecutive_failures = 0 # 重置连续失败计数 # 3. 结果校验（防止空结果污染上下文） if _is_empty_result(result): logger.warning(f"[{tool_name}] 返回空结果") self.consecutive_failures += 1 return self._get_tool_fallback(tool_name) return result except Exception as e: breaker.record_failure() self.consecutive_failures += 1 logger.error(f"[{tool_name}] 调用失败: {e}") return self._get_tool_fallback(tool_name) def _get_tool_fallback(self, tool_name: str) -> dict: """工具降级默认值""" fallbacks = { "search": {"status": "no_results", "message": "未找到相关信息"}, "ocr": {"status": "ocr_unavailable", "message": "OCR服务暂不可用"}, "llm_generate": {"status": "generation_failed", "message": "内容生成失败，请稍后重试"}, } return fallbacks.get(tool_name, {"status": "unknown_error"}) def get_status(self) -> dict: """获取当前执行状态（用于监控和调试）""" return { "tool_calls": self.tool_call_count, "reflections": self.reflection_count, "tokens_used": self.token_used, "consecutive_failures": self.consecutive_failures, "breaker_states": { name: b.state.value for name, b in self.breakers.items() } }

这个框架做了三件关键的事情：

1. 所有工具调用都经过统一入口（call_tool），不可能绕过容错机制。
2. 预算控制是硬性的，超限直接报错，不给Agent"再试一次"的机会。
3. 空结果和异常走同一条降级路径，避免空数据污染下游。

七、几句掏心窝子的话

写到这里，有些经验想分享一下。

第一，别迷信"让Agent自己处理错误"。很多开发者觉得给Agent一段prompt说"如果遇到错误就停下来想一想"就行了。实测下来，这招的成功率大概跟掷骰子差不多。错误处理逻辑必须是确定性的代码，不能交给概率模型。

第二，监控比容错更重要。你设计了一百种容错模式，但如果Agent降级的时候你不知道，那跟没设计一样。建议至少监控这几个指标：工具调用成功率、断路器状态变化、降级触发频率、单次任务token消耗。

第三，容错不等于无限重试。我之前见过一个系统，重试逻辑嵌套了三层——外层Agent重试，中间件重试，内层API客户端也重试。结果一个超时请求被放大了27次。重试要有边界，预算要有上限。

2026年了，AI Agent的开发框架从LangGraph到CrewAI到OpenAI Agents SDK百花齐放，MCP协议也把工具互操作的标准化推进了一大步。但不管框架怎么变，上面这些容错模式都是绕不过去的基础功。Demo阶段你可以忽略它们，但上生产的那天，它们就是你的救命稻草。

参考资料：

1. InfoQ - The Platform Engineering Playbook for Production LLMs（2026.10）
2. DEV Community - Building Fault-Tolerant AI Agent Loops: Patterns Beyond While(True)（2026.10.07）
3. DEV Community - How to Build Resilient AI Agents with Search Fallback Loops（2026.10.06）
4. Microsoft Learn - 测试AI Agent工具失败场景（Dev Proxy文档，2026.10.06）
5. LangGraph官方文档 - Checkpointer与Memory机制

如果这篇文章对你有帮助，欢迎点赞、收藏，也欢迎在评论区聊聊你的Agent在生产环境里遇到过什么奇葩的翻车经历。有问题的话直接评论区见，看到都会回。