LangGraph重试策略构建稳定AI工作流的终极指南【免费下载链接】langgraphBuild resilient agents.项目地址: https://gitcode.com/GitHub_Trending/la/langgraph在当今复杂多变的AI应用环境中网络抖动、API限流、服务暂时不可用等问题已成为常态。LangGraph作为构建智能工作流的强大框架其重试策略机制为开发者提供了构建稳定可靠AI系统的关键能力。本文将深入探讨LangGraph的重试机制帮助您掌握如何在实际项目中实现自动恢复和容错处理。为什么AI工作流需要智能重试现代AI应用通常涉及多个外部服务调用每个环节都可能面临失败风险。LangGraph重试策略正是为解决这些挑战而设计网络波动API调用超时、连接中断服务限流第三方API的速率限制和配额管理资源竞争数据库连接池耗尽、内存不足暂时性错误服务重启、负载均衡切换LangGraph UI界面展示工作流可视化重试策略确保节点执行可靠性LangGraph重试策略核心架构RetryPolicy类详解LangGraph通过RetryPolicy类提供灵活的重试配置这是构建稳定工作流的基础from langgraph.types import RetryPolicy # 基础重试策略配置 basic_retry RetryPolicy( max_attempts3, # 最大重试次数包含首次尝试 initial_interval0.5, # 初始重试间隔秒 backoff_factor2.0, # 退避因子 max_interval128.0, # 最大重试间隔 jitterTrue, # 是否添加随机抖动 retry_ondefault_retry_on # 默认重试条件 ) # 自定义异常处理策略 custom_retry RetryPolicy( max_attempts5, initial_interval1.0, retry_onlambda exc: ( isinstance(exc, ConnectionError) or (hasattr(exc, status_code) and exc.status_code 500) ) )内置智能异常分类LangGraph内置了智能的异常处理逻辑自动识别可恢复错误# 默认重试条件实现 def default_retry_on(exc: Exception) - bool: import httpx import requests # 网络连接错误总是重试 if isinstance(exc, ConnectionError): return True # HTTP 5xx服务器错误重试 if isinstance(exc, httpx.HTTPStatusError): return 500 exc.response.status_code 600 # Requests库的HTTP错误 if isinstance(exc, requests.HTTPError): return 500 exc.response.status_code 600 if exc.response else True # 以下错误类型不重试 non_retryable ( ValueError, TypeError, ArithmeticError, ImportError, LookupError, NameError, SyntaxError, RuntimeError, ReferenceError, StopIteration, StopAsyncIteration, OSError ) if isinstance(exc, non_retryable): return False # 其他异常默认重试 return True实战配置节点级重试策略基础工作流重试配置from langgraph.graph import StateGraph from langgraph.prebuilt import ToolNode def unreliable_api_call(input_data): 模拟可能失败的API调用 import random if random.random() 0.4: # 40%失败率 raise ConnectionError(API服务暂时不可用) return {result: success, data: input_data} # 创建带重试策略的工作流 builder StateGraph(dict) # 配置带重试的节点 api_node ToolNode( tools[unreliable_api_call], retry_policyRetryPolicy( max_attempts4, initial_interval1.0, backoff_factor1.5, jitterTrue ) ) builder.add_node(api_processor, api_node) builder.set_entry_point(api_processor) builder.set_finish_point(api_processor) workflow builder.compile() # 执行工作流 result workflow.invoke({input: test_data})多节点差异化重试策略from langgraph.types import RetryPolicy # 为不同节点配置不同的重试策略 database_retry RetryPolicy( max_attempts3, initial_interval0.5, retry_on(ConnectionError, TimeoutError) ) external_api_retry RetryPolicy( max_attempts5, initial_interval2.0, backoff_factor2.0, retry_onlambda exc: isinstance(exc, (ConnectionError, TimeoutError)) ) llm_service_retry RetryPolicy( max_attempts2, initial_interval3.0, retry_on(ConnectionError,) ) # 构建复杂工作流 builder StateGraph(dict) builder.add_node(db_query, database_query_node, retry_policydatabase_retry) builder.add_node(api_call, external_api_node, retry_policyexternal_api_retry) builder.add_node(llm_process, llm_service_node, retry_policyllm_service_retry)高级重试模式实现指数退避与随机抖动import random import time class SmartRetryPolicy(RetryPolicy): 智能重试策略指数退避 随机抖动 def calculate_delay(self, attempt: int) - float: 计算重试延迟时间 # 指数退避delay initial * (backoff_factor ^ (attempt-1)) delay self.initial_interval * (self.backoff_factor ** (attempt - 1)) # 应用最大间隔限制 delay min(delay, self.max_interval) # 添加随机抖动避免重试风暴 if self.jitter: jitter_factor random.uniform(0.8, 1.2) delay * jitter_factor return delay def should_retry(self, exc: Exception, attempt: int) - bool: 判断是否应该重试 if attempt self.max_attempts: return False # 自定义重试条件判断 if callable(self.retry_on): return self.retry_on(exc) elif isinstance(self.retry_on, (list, tuple)): return any(isinstance(exc, exc_type) for exc_type in self.retry_on) else: return isinstance(exc, self.retry_on)熔断器模式集成class CircuitBreakerRetryPolicy(RetryPolicy): 熔断器重试策略防止级联故障 def __init__(self, failure_threshold5, reset_timeout60, **kwargs): super().__init__(**kwargs) self.failure_count 0 self.circuit_open False self.last_failure_time None self.failure_threshold failure_threshold self.reset_timeout reset_timeout def should_retry(self, exc: Exception, attempt: int) - bool: 检查熔断器状态 current_time time.time() # 检查是否需要重置熔断器 if (self.circuit_open and self.last_failure_time and current_time - self.last_failure_time self.reset_timeout): self.circuit_open False self.failure_count 0 # 如果熔断器打开直接返回失败 if self.circuit_open: return False # 检查重试条件 if not super().should_retry(exc, attempt): return False # 更新失败计数 self.failure_count 1 self.last_failure_time current_time # 检查是否触发熔断 if self.failure_count self.failure_threshold: self.circuit_open True return False return True重试策略配置最佳实践不同场景的推荐配置场景类型最大重试次数初始延迟退避因子适用场景网络API调用3-4次1.0秒2.0HTTP API、REST服务数据库操作2-3次0.5秒1.5数据库连接、查询文件IO操作1-2次2.0秒1.0文件读写、存储操作第三方服务4-5次2.0秒2.5外部API、云服务LLM服务调用2-3次3.0秒2.0AI模型API调用性能优化建议# 优化建议1合理设置重试次数 optimized_policy RetryPolicy( max_attempts3, # 平衡成功率和响应时间 initial_interval1.0, backoff_factor2.0, max_interval30.0, # 避免无限等待 jitterTrue # 避免重试风暴 ) # 优化建议2精细化异常处理 def smart_retry_condition(exc: Exception) - bool: 智能重试条件判断 # 网络错误总是重试 if isinstance(exc, ConnectionError): return True # 超时错误重试 if isinstance(exc, TimeoutError): return True # HTTP 5xx错误重试 if hasattr(exc, status_code): status getattr(exc, status_code, None) if status and 500 status 600: return True # 特定业务异常不重试 business_errors (ValueError, TypeError, PermissionError) if isinstance(exc, business_errors): return False # 其他情况根据具体业务决定 return False监控与调试技巧重试事件追踪from dataclasses import dataclass from datetime import datetime from typing import Dict, Any import logging dataclass class RetryEvent: 重试事件记录 timestamp: datetime node_name: str attempt: int exception: str delay: float success: bool class MonitoredRetryPolicy(RetryPolicy): 带监控的重试策略 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.events: List[RetryEvent] [] self.logger logging.getLogger(__name__) def before_retry(self, exc: Exception, attempt: int, delay: float): 重试前记录 event RetryEvent( timestampdatetime.now(), node_namegetattr(self, node_name, unknown), attemptattempt, exceptionf{type(exc).__name__}: {str(exc)}, delaydelay, successFalse ) self.events.append(event) # 记录到日志 self.logger.warning( f重试事件: 节点{event.node_name}, f尝试{attempt}/{self.max_attempts}, f延迟{delay:.2f}s, 错误{event.exception} ) def on_success(self, attempt: int): 成功记录 event RetryEvent( timestampdatetime.now(), node_namegetattr(self, node_name, unknown), attemptattempt, exception, delay0, successTrue ) self.events.append(event) self.logger.info(f节点 {event.node_name} 在第 {attempt} 次尝试成功)调试配置示例# 启用详细调试日志 import logging logging.basicConfig( levellogging.DEBUG, format%(asctime)s - %(name)s - %(levelname)s - %(message)s ) # 创建调试用重试策略 debug_policy RetryPolicy( max_attempts2, initial_interval1.0, retry_on(Exception,), # 重试所有异常用于调试 jitterFalse # 禁用抖动以便调试 ) # 在开发环境中使用 if DEBUG_MODE: policy debug_policy else: policy production_policy故障排除指南常见问题解决方案问题现象可能原因解决方案重试不生效异常类型不在retry_on列表中检查异常类型使用更宽泛的匹配条件重试过于频繁退避因子设置过小增加backoff_factor到2.0或更高重试延迟过长max_interval设置过大根据业务需求调整最大间隔重试风暴jitterFalse且并发量高启用jitterTrue添加随机抖动熔断器误触发failure_threshold设置过低根据实际失败率调整阈值调试检查清单# 重试策略调试检查清单 def validate_retry_policy(policy: RetryPolicy): 验证重试策略配置 checks [] # 检查1: 最大重试次数 if policy.max_attempts 1: checks.append(❌ max_attempts必须大于0) elif policy.max_attempts 10: checks.append(⚠️ max_attempts过大可能影响用户体验) else: checks.append(✅ max_attempts配置合理) # 检查2: 重试间隔 if policy.initial_interval 0: checks.append(❌ initial_interval必须大于0) else: checks.append(✅ initial_interval配置合理) # 检查3: 退避因子 if policy.backoff_factor 1.0: checks.append(❌ backoff_factor必须大于等于1.0) else: checks.append(✅ backoff_factor配置合理) # 检查4: 最大间隔 if policy.max_interval policy.initial_interval: checks.append(❌ max_interval必须大于等于initial_interval) else: checks.append(✅ max_interval配置合理) return checks高级应用组合重试策略分层重试策略from typing import List class CompositeRetryPolicy: 组合重试策略支持多种策略组合 def __init__(self, policies: List[RetryPolicy]): self.policies policies def should_retry(self, exc: Exception, attempt: int) - bool: 检查所有策略是否允许重试 for policy in self.policies: if not policy.should_retry(exc, attempt): return False return True def get_delay(self, attempt: int) - float: 获取最大延迟时间 delays [policy.get_delay(attempt) for policy in self.policies] return max(delays) # 使用组合策略 network_policy RetryPolicy(max_attempts3, retry_on(ConnectionError,)) timeout_policy RetryPolicy(max_attempts2, retry_on(TimeoutError,)) composite_policy CompositeRetryPolicy([network_policy, timeout_policy])自适应重试策略class AdaptiveRetryPolicy(RetryPolicy): 自适应重试策略根据历史成功率调整 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.success_history [] self.window_size 100 def update_success_rate(self, success: bool): 更新成功率历史 self.success_history.append(success) if len(self.success_history) self.window_size: self.success_history.pop(0) def get_success_rate(self) - float: 计算最近的成功率 if not self.success_history: return 1.0 return sum(self.success_history) / len(self.success_history) def should_retry(self, exc: Exception, attempt: int) - bool: 根据成功率动态调整重试策略 success_rate self.get_success_rate() # 成功率低时减少重试次数 if success_rate 0.7 and attempt 1: return False return super().should_retry(exc, attempt)性能调优最佳实践重试策略性能指标class PerformanceMetrics: 重试性能指标监控 def __init__(self): self.total_attempts 0 self.successful_attempts 0 self.failed_attempts 0 self.total_delay 0.0 def record_attempt(self, success: bool, delay: float): 记录重试尝试 self.total_attempts 1 if success: self.successful_attempts 1 else: self.failed_attempts 1 self.total_delay delay def get_success_rate(self) - float: 计算成功率 if self.total_attempts 0: return 0.0 return self.successful_attempts / self.total_attempts def get_average_delay(self) - float: 计算平均延迟 if self.total_attempts 0: return 0.0 return self.total_delay / self.total_attempts def get_metrics(self) - Dict[str, Any]: 获取所有指标 return { total_attempts: self.total_attempts, successful_attempts: self.successful_attempts, failed_attempts: self.failed_attempts, success_rate: self.get_success_rate(), average_delay: self.get_average_delay(), total_delay: self.total_delay }优化建议总结合理设置重试次数根据服务SLA和用户体验平衡设置启用指数退避避免重试风暴减轻服务压力添加随机抖动分散重试时间防止同步重试精细化异常处理只为可恢复错误重试监控重试率及时发现系统问题实施熔断机制防止级联故障总结构建稳定AI工作流的关键LangGraph的重试策略为构建可靠的AI应用提供了坚实基础。通过灵活的配置选项、智能的异常处理和丰富的监控能力开发者可以✅实现自动错误恢复处理暂时性故障提高系统可用性✅优化用户体验减少失败感知提供更流畅的服务✅保护后端服务避免重试风暴防止级联故障✅全面监控运维实时跟踪重试行为及时发现系统问题掌握LangGraph的重试机制您将能够构建出真正稳定可靠的AI工作流在复杂的生产环境中保持高可用性为用户提供卓越的服务体验。无论是简单的API调用还是复杂的多节点工作流合理的重试策略都是确保系统稳定性的关键一环。【免费下载链接】langgraphBuild resilient agents.项目地址: https://gitcode.com/GitHub_Trending/la/langgraph创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
LangGraph重试策略：构建稳定AI工作流的终极指南
LangGraph重试策略构建稳定AI工作流的终极指南【免费下载链接】langgraphBuild resilient agents.项目地址: https://gitcode.com/GitHub_Trending/la/langgraph在当今复杂多变的AI应用环境中网络抖动、API限流、服务暂时不可用等问题已成为常态。LangGraph作为构建智能工作流的强大框架其重试策略机制为开发者提供了构建稳定可靠AI系统的关键能力。本文将深入探讨LangGraph的重试机制帮助您掌握如何在实际项目中实现自动恢复和容错处理。为什么AI工作流需要智能重试现代AI应用通常涉及多个外部服务调用每个环节都可能面临失败风险。LangGraph重试策略正是为解决这些挑战而设计网络波动API调用超时、连接中断服务限流第三方API的速率限制和配额管理资源竞争数据库连接池耗尽、内存不足暂时性错误服务重启、负载均衡切换LangGraph UI界面展示工作流可视化重试策略确保节点执行可靠性LangGraph重试策略核心架构RetryPolicy类详解LangGraph通过RetryPolicy类提供灵活的重试配置这是构建稳定工作流的基础from langgraph.types import RetryPolicy # 基础重试策略配置 basic_retry RetryPolicy( max_attempts3, # 最大重试次数包含首次尝试 initial_interval0.5, # 初始重试间隔秒 backoff_factor2.0, # 退避因子 max_interval128.0, # 最大重试间隔 jitterTrue, # 是否添加随机抖动 retry_ondefault_retry_on # 默认重试条件 ) # 自定义异常处理策略 custom_retry RetryPolicy( max_attempts5, initial_interval1.0, retry_onlambda exc: ( isinstance(exc, ConnectionError) or (hasattr(exc, status_code) and exc.status_code 500) ) )内置智能异常分类LangGraph内置了智能的异常处理逻辑自动识别可恢复错误# 默认重试条件实现 def default_retry_on(exc: Exception) - bool: import httpx import requests # 网络连接错误总是重试 if isinstance(exc, ConnectionError): return True # HTTP 5xx服务器错误重试 if isinstance(exc, httpx.HTTPStatusError): return 500 exc.response.status_code 600 # Requests库的HTTP错误 if isinstance(exc, requests.HTTPError): return 500 exc.response.status_code 600 if exc.response else True # 以下错误类型不重试 non_retryable ( ValueError, TypeError, ArithmeticError, ImportError, LookupError, NameError, SyntaxError, RuntimeError, ReferenceError, StopIteration, StopAsyncIteration, OSError ) if isinstance(exc, non_retryable): return False # 其他异常默认重试 return True实战配置节点级重试策略基础工作流重试配置from langgraph.graph import StateGraph from langgraph.prebuilt import ToolNode def unreliable_api_call(input_data): 模拟可能失败的API调用 import random if random.random() 0.4: # 40%失败率 raise ConnectionError(API服务暂时不可用) return {result: success, data: input_data} # 创建带重试策略的工作流 builder StateGraph(dict) # 配置带重试的节点 api_node ToolNode( tools[unreliable_api_call], retry_policyRetryPolicy( max_attempts4, initial_interval1.0, backoff_factor1.5, jitterTrue ) ) builder.add_node(api_processor, api_node) builder.set_entry_point(api_processor) builder.set_finish_point(api_processor) workflow builder.compile() # 执行工作流 result workflow.invoke({input: test_data})多节点差异化重试策略from langgraph.types import RetryPolicy # 为不同节点配置不同的重试策略 database_retry RetryPolicy( max_attempts3, initial_interval0.5, retry_on(ConnectionError, TimeoutError) ) external_api_retry RetryPolicy( max_attempts5, initial_interval2.0, backoff_factor2.0, retry_onlambda exc: isinstance(exc, (ConnectionError, TimeoutError)) ) llm_service_retry RetryPolicy( max_attempts2, initial_interval3.0, retry_on(ConnectionError,) ) # 构建复杂工作流 builder StateGraph(dict) builder.add_node(db_query, database_query_node, retry_policydatabase_retry) builder.add_node(api_call, external_api_node, retry_policyexternal_api_retry) builder.add_node(llm_process, llm_service_node, retry_policyllm_service_retry)高级重试模式实现指数退避与随机抖动import random import time class SmartRetryPolicy(RetryPolicy): 智能重试策略指数退避 随机抖动 def calculate_delay(self, attempt: int) - float: 计算重试延迟时间 # 指数退避delay initial * (backoff_factor ^ (attempt-1)) delay self.initial_interval * (self.backoff_factor ** (attempt - 1)) # 应用最大间隔限制 delay min(delay, self.max_interval) # 添加随机抖动避免重试风暴 if self.jitter: jitter_factor random.uniform(0.8, 1.2) delay * jitter_factor return delay def should_retry(self, exc: Exception, attempt: int) - bool: 判断是否应该重试 if attempt self.max_attempts: return False # 自定义重试条件判断 if callable(self.retry_on): return self.retry_on(exc) elif isinstance(self.retry_on, (list, tuple)): return any(isinstance(exc, exc_type) for exc_type in self.retry_on) else: return isinstance(exc, self.retry_on)熔断器模式集成class CircuitBreakerRetryPolicy(RetryPolicy): 熔断器重试策略防止级联故障 def __init__(self, failure_threshold5, reset_timeout60, **kwargs): super().__init__(**kwargs) self.failure_count 0 self.circuit_open False self.last_failure_time None self.failure_threshold failure_threshold self.reset_timeout reset_timeout def should_retry(self, exc: Exception, attempt: int) - bool: 检查熔断器状态 current_time time.time() # 检查是否需要重置熔断器 if (self.circuit_open and self.last_failure_time and current_time - self.last_failure_time self.reset_timeout): self.circuit_open False self.failure_count 0 # 如果熔断器打开直接返回失败 if self.circuit_open: return False # 检查重试条件 if not super().should_retry(exc, attempt): return False # 更新失败计数 self.failure_count 1 self.last_failure_time current_time # 检查是否触发熔断 if self.failure_count self.failure_threshold: self.circuit_open True return False return True重试策略配置最佳实践不同场景的推荐配置场景类型最大重试次数初始延迟退避因子适用场景网络API调用3-4次1.0秒2.0HTTP API、REST服务数据库操作2-3次0.5秒1.5数据库连接、查询文件IO操作1-2次2.0秒1.0文件读写、存储操作第三方服务4-5次2.0秒2.5外部API、云服务LLM服务调用2-3次3.0秒2.0AI模型API调用性能优化建议# 优化建议1合理设置重试次数 optimized_policy RetryPolicy( max_attempts3, # 平衡成功率和响应时间 initial_interval1.0, backoff_factor2.0, max_interval30.0, # 避免无限等待 jitterTrue # 避免重试风暴 ) # 优化建议2精细化异常处理 def smart_retry_condition(exc: Exception) - bool: 智能重试条件判断 # 网络错误总是重试 if isinstance(exc, ConnectionError): return True # 超时错误重试 if isinstance(exc, TimeoutError): return True # HTTP 5xx错误重试 if hasattr(exc, status_code): status getattr(exc, status_code, None) if status and 500 status 600: return True # 特定业务异常不重试 business_errors (ValueError, TypeError, PermissionError) if isinstance(exc, business_errors): return False # 其他情况根据具体业务决定 return False监控与调试技巧重试事件追踪from dataclasses import dataclass from datetime import datetime from typing import Dict, Any import logging dataclass class RetryEvent: 重试事件记录 timestamp: datetime node_name: str attempt: int exception: str delay: float success: bool class MonitoredRetryPolicy(RetryPolicy): 带监控的重试策略 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.events: List[RetryEvent] [] self.logger logging.getLogger(__name__) def before_retry(self, exc: Exception, attempt: int, delay: float): 重试前记录 event RetryEvent( timestampdatetime.now(), node_namegetattr(self, node_name, unknown), attemptattempt, exceptionf{type(exc).__name__}: {str(exc)}, delaydelay, successFalse ) self.events.append(event) # 记录到日志 self.logger.warning( f重试事件: 节点{event.node_name}, f尝试{attempt}/{self.max_attempts}, f延迟{delay:.2f}s, 错误{event.exception} ) def on_success(self, attempt: int): 成功记录 event RetryEvent( timestampdatetime.now(), node_namegetattr(self, node_name, unknown), attemptattempt, exception, delay0, successTrue ) self.events.append(event) self.logger.info(f节点 {event.node_name} 在第 {attempt} 次尝试成功)调试配置示例# 启用详细调试日志 import logging logging.basicConfig( levellogging.DEBUG, format%(asctime)s - %(name)s - %(levelname)s - %(message)s ) # 创建调试用重试策略 debug_policy RetryPolicy( max_attempts2, initial_interval1.0, retry_on(Exception,), # 重试所有异常用于调试 jitterFalse # 禁用抖动以便调试 ) # 在开发环境中使用 if DEBUG_MODE: policy debug_policy else: policy production_policy故障排除指南常见问题解决方案问题现象可能原因解决方案重试不生效异常类型不在retry_on列表中检查异常类型使用更宽泛的匹配条件重试过于频繁退避因子设置过小增加backoff_factor到2.0或更高重试延迟过长max_interval设置过大根据业务需求调整最大间隔重试风暴jitterFalse且并发量高启用jitterTrue添加随机抖动熔断器误触发failure_threshold设置过低根据实际失败率调整阈值调试检查清单# 重试策略调试检查清单 def validate_retry_policy(policy: RetryPolicy): 验证重试策略配置 checks [] # 检查1: 最大重试次数 if policy.max_attempts 1: checks.append(❌ max_attempts必须大于0) elif policy.max_attempts 10: checks.append(⚠️ max_attempts过大可能影响用户体验) else: checks.append(✅ max_attempts配置合理) # 检查2: 重试间隔 if policy.initial_interval 0: checks.append(❌ initial_interval必须大于0) else: checks.append(✅ initial_interval配置合理) # 检查3: 退避因子 if policy.backoff_factor 1.0: checks.append(❌ backoff_factor必须大于等于1.0) else: checks.append(✅ backoff_factor配置合理) # 检查4: 最大间隔 if policy.max_interval policy.initial_interval: checks.append(❌ max_interval必须大于等于initial_interval) else: checks.append(✅ max_interval配置合理) return checks高级应用组合重试策略分层重试策略from typing import List class CompositeRetryPolicy: 组合重试策略支持多种策略组合 def __init__(self, policies: List[RetryPolicy]): self.policies policies def should_retry(self, exc: Exception, attempt: int) - bool: 检查所有策略是否允许重试 for policy in self.policies: if not policy.should_retry(exc, attempt): return False return True def get_delay(self, attempt: int) - float: 获取最大延迟时间 delays [policy.get_delay(attempt) for policy in self.policies] return max(delays) # 使用组合策略 network_policy RetryPolicy(max_attempts3, retry_on(ConnectionError,)) timeout_policy RetryPolicy(max_attempts2, retry_on(TimeoutError,)) composite_policy CompositeRetryPolicy([network_policy, timeout_policy])自适应重试策略class AdaptiveRetryPolicy(RetryPolicy): 自适应重试策略根据历史成功率调整 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.success_history [] self.window_size 100 def update_success_rate(self, success: bool): 更新成功率历史 self.success_history.append(success) if len(self.success_history) self.window_size: self.success_history.pop(0) def get_success_rate(self) - float: 计算最近的成功率 if not self.success_history: return 1.0 return sum(self.success_history) / len(self.success_history) def should_retry(self, exc: Exception, attempt: int) - bool: 根据成功率动态调整重试策略 success_rate self.get_success_rate() # 成功率低时减少重试次数 if success_rate 0.7 and attempt 1: return False return super().should_retry(exc, attempt)性能调优最佳实践重试策略性能指标class PerformanceMetrics: 重试性能指标监控 def __init__(self): self.total_attempts 0 self.successful_attempts 0 self.failed_attempts 0 self.total_delay 0.0 def record_attempt(self, success: bool, delay: float): 记录重试尝试 self.total_attempts 1 if success: self.successful_attempts 1 else: self.failed_attempts 1 self.total_delay delay def get_success_rate(self) - float: 计算成功率 if self.total_attempts 0: return 0.0 return self.successful_attempts / self.total_attempts def get_average_delay(self) - float: 计算平均延迟 if self.total_attempts 0: return 0.0 return self.total_delay / self.total_attempts def get_metrics(self) - Dict[str, Any]: 获取所有指标 return { total_attempts: self.total_attempts, successful_attempts: self.successful_attempts, failed_attempts: self.failed_attempts, success_rate: self.get_success_rate(), average_delay: self.get_average_delay(), total_delay: self.total_delay }优化建议总结合理设置重试次数根据服务SLA和用户体验平衡设置启用指数退避避免重试风暴减轻服务压力添加随机抖动分散重试时间防止同步重试精细化异常处理只为可恢复错误重试监控重试率及时发现系统问题实施熔断机制防止级联故障总结构建稳定AI工作流的关键LangGraph的重试策略为构建可靠的AI应用提供了坚实基础。通过灵活的配置选项、智能的异常处理和丰富的监控能力开发者可以✅实现自动错误恢复处理暂时性故障提高系统可用性✅优化用户体验减少失败感知提供更流畅的服务✅保护后端服务避免重试风暴防止级联故障✅全面监控运维实时跟踪重试行为及时发现系统问题掌握LangGraph的重试机制您将能够构建出真正稳定可靠的AI工作流在复杂的生产环境中保持高可用性为用户提供卓越的服务体验。无论是简单的API调用还是复杂的多节点工作流合理的重试策略都是确保系统稳定性的关键一环。【免费下载链接】langgraphBuild resilient agents.项目地址: https://gitcode.com/GitHub_Trending/la/langgraph创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
