网站建设要哪些工作,两个网站 一个域名,广州建工集团有限公司官网,企业解决方案除了思爱普day33 Agent 错误恢复与回退策略
xsun_ai_study
错误类型分类与处理策略
错误分类体系 核心恢复策略矩阵
策略类型适用场景实现复杂度恢复成功率重试机制临时性错误#xff08;网络、API限流#xff09;低60-80%降级处理工具不可用、功能缺失中70-90%熔断机制服务持续故障…day33 Agent 错误恢复与回退策略xsun_ai_study错误类型分类与处理策略错误分类体系核心恢复策略矩阵策略类型适用场景实现复杂度恢复成功率重试机制临时性错误网络、API限流低60-80%降级处理工具不可用、功能缺失中70-90%熔断机制服务持续故障中90-95%旁路策略主路径失败高80-95%人工干预复杂逻辑错误低95-100%多层级防御架构第1层预防层PreventionclassPreventiveMeasures:预防性措施staticmethoddefvalidate_input(user_input:str,max_length:int1000)-ValidationResult:输入验证checks[(长度检查,len(user_input)max_length),(恶意代码检查,notany(keywordinuser_input.lower()forkeywordin[system(,exec(,eval(])),(敏感信息检查,notany(patterninuser_inputforpatternin[密码,token:,apikey])),(编码检查,user_input.isprintable())]failures[nameforname,passedinchecksifnotpassed]returnValidationResult(validlen(failures)0,failuresfailures)staticmethoddefsanitize_tool_parameters(params:Dict)-Dict:参数消毒sanitized{}forkey,valueinparams.items():ifisinstance(value,str):# 移除潜在的注入代码sanitized[key]value.replace(;,).replace(,).replace($(,)else:sanitized[key]valuereturnsanitized第2层检测层DetectionclassErrorDetector:错误检测器def__init__(self):self.error_patterns{timeout:[timeout,timed out,请求超时,operation timeout,连接超时],rate_limit:[rate limit,quota,limit exceeded,API调用次数超限,429],authentication:[unauthorized,forbidden,invalid token,authentication failed,401,403],validation:[invalid parameter,bad request,validation failed,参数错误,400],server_error:[internal server error,server unavailable,服务器错误,500,503],llm_error:[content policy,cannot fulfill,refused,抱歉我无法,根据我的使用条款]}defclassify_error(self,error_message:str)-ErrorType:错误分类error_message_lowererror_message.lower()forerror_type,patternsinself.error_patterns.items():forpatterninpatterns:ifpattern.lower()inerror_message_lower:returnErrorType(typeerror_type,patternpattern,confidence0.9)returnErrorType(typeunknown,pattern,confidence0.0)defdetect_infinite_loop(self,execution_history:List[Dict])-bool:检测无限循环iflen(execution_history)3:returnFalse# 检查最近三次操作是否相同recent_ops[step.get(tool_name,)forstepinexecution_history[-3:]]iflen(set(recent_ops))1andrecent_ops[0]:returnTrue# 检查状态是否重复recent_states[hash(str(step.get(parameters,{})))forstepinexecution_history[-5:]]returnlen(set(recent_states))3第3层恢复层RecoveryclassRecoveryStrategies:恢复策略集合def__init__(self,llm_client,fallback_tools:Dict):self.llmllm_client self.fallback_toolsfallback_tools self.circuit_breakers{}defretry_with_backoff(self,func:Callable,max_retries:int3,initial_delay:float1.0)-Any:指数退避重试delayinitial_delayforattemptinrange(max_retries):try:returnfunc()exceptExceptionase:ifattemptmax_retries-1:raiseerror_typeself.detector.classify_error(str(e))# 对于某些错误不重试iferror_type.typein[authentication,validation]:raiselogger.warning(f重试{attempt1}/{max_retries}:{str(e)})time.sleep(delay)delay*2# 指数退避deffallback_to_simpler_tool(self,failed_tool:str,original_params:Dict,context:Dict)-Any:降级到更简单的工具fallback_chain{web_search:[(local_knowledge_base,0.8),(cached_search_results,0.6),(llm_general_knowledge,0.4)],calculator:[(simple_math_parser,0.9),(llm_calculation,0.7),(approximate_estimation,0.5)],weather_api:[(historical_weather,0.8),(seasonal_average,0.6),(manual_input,0.3)]}iffailed_toolnotinfallback_chain:returnNoneforfallback_tool,confidenceinfallback_chain[failed_tool]:iffallback_toolinself.fallback_tools:try:resultself.fallback_tools[fallback_tool](original_params)logger.info(f使用降级工具{fallback_tool}(置信度:{confidence}))return{result:result,source:fallback_tool,confidence:confidence,is_fallback:True}except:continuereturnNonedefcircuit_breaker(self,tool_name:str,failure_threshold:int5)-bool:熔断器模式iftool_namenotinself.circuit_breakers:self.circuit_breakers[tool_name]{failures:0,last_failure:None,state:closed}cbself.circuit_breakers[tool_name]ifcb[state]open:# 检查是否应该进入半开状态if(cb[last_failure]andtime.time()-cb[last_failure]60):# 60秒后重试cb[state]half-openreturnTruereturnFalseifcb[state]half-open:# 半开状态只允许一次尝试cb[state]open# 假设这次会失败returnTrue# closed状态检查失败次数ifcb[failures]failure_threshold:cb[state]opencb[last_failure]time.time()logger.warning(f熔断器触发:{tool_name})returnFalsereturnTruedefupdate_circuit_state(self,tool_name:str,success:bool):更新熔断器状态iftool_namenotinself.circuit_breakers:returncbself.circuit_breakers[tool_name]ifsuccess:cb[failures]0ifcb[state]half-open:cb[state]closed# 成功关闭熔断器else:cb[failures]1cb[last_failure]time.time()ifcb[state]half-open:cb[state]open# 失败保持打开第4层旁路层BypassclassBypassStrategies:旁路策略staticmethoddefsemantic_approximation(query:str,available_data:List)-str:语义近似当无法获取精确数据时提供近似答案approximation_rules{r.*多少.*钱.*:[根据市场行情类似产品价格在XXX-XXX元之间,价格因地区和时间而异通常范围是...,我无法获取实时价格但可以参考历史数据...],r.*天气.*:[当前季节该地区通常天气是...,根据天气预报模型预计...,可以参考邻近城市的天气情况...],r.*时间.*:[通常需要XXX小时具体取决于...,历史平均时间是...,根据类似情况估计...]}forpattern,responsesinapproximation_rules.items():ifre.match(pattern,query):returnrandom.choice(responses)return虽然无法提供精确答案但根据一般情况...staticmethoddefstepwise_refinement(problem:str,max_steps:int3)-List[str]:逐步细化将复杂问题分解为简单问题refinement_promptf 将以下复杂问题分解为不超过{max_steps}个简单问题 原问题{problem}分解步骤每个步骤应该是独立可回答的问题 1. # 调用LLM进行分解decomposedllm_call(refinement_prompt)returndecomposed.split(\n)staticmethoddefalternative_paths(main_path:List[str],available_tools:List[str])-List[List[str]]:生成替代执行路径alternatives[]# 1. 工具替换路径tool_mapping{web_search:[local_search,knowledge_base_query],calculator:[llm_calculation,rule_based_estimation],weather_api:[historical_data,seasonal_pattern]}fortoolinmain_path:iftoolintool_mapping:foraltintool_mapping[tool]:ifaltinavailable_tools:alt_pathmain_path.copy()alt_path[alt_path.index(tool)]alt alternatives.append(alt_path)# 2. 顺序调整路径如果顺序不重要iflen(main_path)1:forperminitertools.permutations(main_path):iflist(perm)!main_path:alternatives.append(list(perm))returnalternatives[:5]# 返回前5个替代路径第5层修复层RepairclassAutoRepairMechanisms:自动修复机制def__init__(self,llm_client):self.llmllm_client self.repair_history[]defrepair_invalid_response(self,invalid_response:str,expected_format:str)-str:修复无效的LLM响应repair_promptf 以下LLM响应不符合预期格式。请修复它。 预期格式{expected_format}无效响应{invalid_response}问题分析 1. 格式错误如缺少字段、错误分隔符 2. 内容错误如逻辑矛盾、事实错误 3. 结构错误如嵌套错误、类型错误 修复后的响应 try:repairedself.llm.call(repair_prompt)self.repair_history.append({original:invalid_response,repaired:repaired,timestamp:datetime.now()})returnrepairedexcept:# 如果修复失败返回默认结构returnself._create_default_response(expected_format)defrecover_from_deadlock(self,agent_state:Dict,execution_history:List)-Dict:从死锁状态恢复# 策略1回退到最后一个稳定状态stable_states[stateforstateinexecution_historyifstate.get(status)success]ifstable_states:last_stablestable_states[-1]logger.info(f回退到稳定状态:{last_stable.get(step_id)})return{**agent_state,current_step:last_stable.get(step_id),context:last_stable.get(context,{}),recovery_action:rollback_to_stable}# 策略2重置并重新开始logger.warning(无稳定状态可用执行软重置)return{**agent_state,current_step:0,context:{},execution_path:self._find_simpler_path(agent_state[goal]),recovery_action:soft_reset}deffix_data_inconsistency(self,data_sources:List[Dict])-Dict:修复数据不一致问题# 策略1多数投票values[source.get(value)forsourceindata_sources]ifvalues:value_countsCounter(values)most_commonvalue_counts.most_common(1)ifmost_common[0][1]len(values)/2:return{value:most_common[0][0],confidence:0.8}# 策略2加权平均对于数值numeric_values[]weights[]forsourceindata_sources:try:valfloat(source.get(value,0))numeric_values.append(val)weights.append(source.get(confidence,0.5))except:continueifnumeric_values:weighted_avgnp.average(numeric_values,weightsweights)return{value:weighted_avg,confidence:0.7}# 策略3让LLM仲裁arbitration_promptf 以下数据源提供的信息不一致请分析并给出最可能正确的值 数据源{json.dumps(data_sources,indent2,ensure_asciiFalse)}请综合考虑数据源的可信度、时间戳和内在逻辑。 输出格式{{value: 最可能的值, reasoning: 推理过程}} returnself.llm.call(arbitration_prompt)