1. 項(xiàng)目概述當(dāng)AI智能體學(xué)會(huì)“補(bǔ)償”最近在搗鼓LangChain和AI智能體Agent開發(fā)的朋友可能都遇到過一種讓人頭疼的情況你精心設(shè)計(jì)的智能體在一條復(fù)雜的任務(wù)鏈上執(zhí)行時(shí)某個(gè)環(huán)節(jié)突然“卡殼”了?？赡苁且?yàn)橥獠緼PI調(diào)用超時(shí)也可能是解析工具返回的結(jié)果格式意外或者僅僅是遇到了一個(gè)訓(xùn)練數(shù)據(jù)里沒見過的用戶提問。傳統(tǒng)的處理方式往往是直接拋出錯(cuò)誤任務(wù)鏈就此中斷用戶體驗(yàn)戛然而止。這就像組建了一個(gè)團(tuán)隊(duì)其中一個(gè)成員遇到困難就立刻擺爛導(dǎo)致整個(gè)項(xiàng)目停滯顯然不是我們想要的協(xié)作方式。“Robust Agent Compensation (RAC)”這個(gè)概念正是在這種背景下被提出和探討的。它的核心思想直白而有力教會(huì)AI智能體具備“補(bǔ)償”能力。這不是指給AI發(fā)工資而是指當(dāng)智能體在執(zhí)行任務(wù)過程中感知到某些子任務(wù)失敗或結(jié)果不理想時(shí)能夠主動(dòng)采取補(bǔ)救措施嘗試?yán)@過障礙、修復(fù)錯(cuò)誤或?qū)ふ姨娲桨笍亩ＷC整體任務(wù)的推進(jìn)甚至完成。RAC追求的不是單個(gè)步驟的百分百完美而是整個(gè)任務(wù)流的最終魯棒性Robustness。想象一下你讓一個(gè)旅行規(guī)劃智能體幫你訂機(jī)票、酒店和租車。如果租車服務(wù)暫時(shí)不可用一個(gè)具備RAC能力的智能體不會(huì)簡(jiǎn)單地回復(fù)“租車失敗”而可能會(huì)嘗試1查找其他租車平臺(tái)2建議改用網(wǎng)約車或公共交通作為替代方案3甚至調(diào)整整個(gè)行程規(guī)劃以適應(yīng)交通方式的改變。這種“出了問題自己想辦法兜底”的思維正是智能體從機(jī)械執(zhí)行走向自主協(xié)作的關(guān)鍵一步。當(dāng)前隨著LangChain、LangGraph等框架的普及多智能體協(xié)作系統(tǒng)的構(gòu)建門檻大大降低。但如何讓這些智能體在動(dòng)態(tài)、不確定的真實(shí)環(huán)境中可靠工作RAC提供了一個(gè)至關(guān)重要的設(shè)計(jì)范式。它涉及智能體的感知判斷何時(shí)需要補(bǔ)償、決策選擇何種補(bǔ)償策略與執(zhí)行實(shí)施補(bǔ)償動(dòng)作的全過程。對(duì)于開發(fā)者而言理解并實(shí)現(xiàn)RAC意味著你構(gòu)建的AI應(yīng)用將告別“脆弱”變得更加健壯和用戶友好。無論是開發(fā)客服機(jī)器人、自動(dòng)化工作流還是復(fù)雜的決策支持系統(tǒng)RAC都是提升智能體實(shí)用價(jià)值必須啃下的硬骨頭。2. RAC的核心設(shè)計(jì)思路與原理拆解實(shí)現(xiàn)一個(gè)具備補(bǔ)償能力的智能體遠(yuǎn)非簡(jiǎn)單的“try-catch”錯(cuò)誤處理那么簡(jiǎn)單。它需要一套系統(tǒng)的設(shè)計(jì)思路將補(bǔ)償邏輯深度嵌入到智能體的認(rèn)知和行動(dòng)循環(huán)中。下面我們來拆解其核心原理。2.1 從被動(dòng)容錯(cuò)到主動(dòng)補(bǔ)償?shù)姆妒睫D(zhuǎn)變傳統(tǒng)的程序或簡(jiǎn)單智能體處理異常屬于被動(dòng)容錯(cuò)。其邏輯是執(zhí)行動(dòng)作 - 監(jiān)測(cè)異常 - 捕獲異常 - 執(zhí)行預(yù)設(shè)的異常處理程序如重試、回滾、返回錯(cuò)誤信息。這個(gè)過程中異常處理路徑是固定的且往往在任務(wù)流設(shè)計(jì)之初就被限定死缺乏對(duì)當(dāng)前任務(wù)上下文和最終目標(biāo)的考量。而RAC倡導(dǎo)的主動(dòng)補(bǔ)償則是一種更高階的模式。它的流程更接近于執(zhí)行動(dòng)作 - 評(píng)估結(jié)果不僅看成功/失敗更看結(jié)果質(zhì)量與目標(biāo)契合度- 若未達(dá)預(yù)期則基于當(dāng)前任務(wù)狀態(tài)、歷史經(jīng)驗(yàn)和可用資源動(dòng)態(tài)生成一個(gè)或多個(gè)補(bǔ)償計(jì)劃 - 執(zhí)行補(bǔ)償計(jì)劃 - 重新評(píng)估。這里的“補(bǔ)償”是一個(gè)更廣義的概念包括但不限于重試與調(diào)整更換參數(shù)、增加等待時(shí)間后重試同一操作。替代方案執(zhí)行當(dāng)首選工具或API失敗時(shí)自動(dòng)切換到功能近似的備選方案。目標(biāo)降級(jí)與協(xié)商當(dāng)原定目標(biāo)無法完全達(dá)成時(shí)與用戶或其他智能體協(xié)商一個(gè)可接受的、降級(jí)后的目標(biāo)。信息修補(bǔ)與推理當(dāng)獲取的信息不完整或有噪聲時(shí)利用已有知識(shí)進(jìn)行推理和修補(bǔ)以繼續(xù)任務(wù)。任務(wù)分解與重組將失敗的任務(wù)分解成更小的子任務(wù)或與其他成功任務(wù)的結(jié)果重新組合尋找新的完成路徑。這種轉(zhuǎn)變的關(guān)鍵在于智能體需要有一個(gè)持續(xù)的“目標(biāo)感”和“狀態(tài)感知”。它不僅僅關(guān)注當(dāng)前步驟的輸入輸出更要時(shí)刻牢記終極任務(wù)是什么當(dāng)前進(jìn)展到了哪一步以及手頭有哪些資源可用。這通常需要借助智能體狀態(tài)管理如通過LangGraph的持久化狀態(tài)和規(guī)劃與推理模塊如利用大語言模型的規(guī)劃能力來實(shí)現(xiàn)。2.2 RAC實(shí)現(xiàn)的三大技術(shù)支柱要讓智能體學(xué)會(huì)補(bǔ)償我們需要在架構(gòu)上為其提供三方面的支持1. 狀態(tài)感知與異常評(píng)估模塊這是補(bǔ)償?shù)挠|發(fā)點(diǎn)。智能體需要有能力判斷“何時(shí)需要補(bǔ)償”。這不僅僅是捕獲一個(gè)異常錯(cuò)誤碼更需要語義層面的評(píng)估。工具執(zhí)行結(jié)果驗(yàn)證調(diào)用一個(gè)工具后除了檢查HTTP狀態(tài)碼還要解析返回內(nèi)容判斷其是否有效、是否完整、是否符合預(yù)期格式。例如調(diào)用天氣API返回了數(shù)據(jù)但關(guān)鍵的溫度字段為null這應(yīng)被視為需要補(bǔ)償?shù)摹败浭　?。目?biāo)達(dá)成度評(píng)估設(shè)立一些可量化的指標(biāo)或規(guī)則用于評(píng)估當(dāng)前結(jié)果距離最終目標(biāo)還有多遠(yuǎn)。例如在信息搜集任務(wù)中可以評(píng)估信息的覆蓋率、關(guān)鍵實(shí)體的出現(xiàn)次數(shù)等。上下文一致性檢查檢查當(dāng)前步驟的結(jié)果是否與之前步驟的結(jié)果或全局任務(wù)描述存在邏輯矛盾。在LangChain中我們可以利用Tool類的handle_tool_error進(jìn)行基礎(chǔ)錯(cuò)誤捕獲但更高級(jí)的評(píng)估通常需要自定義回調(diào)函數(shù)Callbacks或在智能體執(zhí)行循環(huán)中插入評(píng)估節(jié)點(diǎn)。2. 補(bǔ)償策略庫與策略選擇器這是補(bǔ)償?shù)摹按竽X”。我們需要為智能體預(yù)先定義或讓其動(dòng)態(tài)生成一系列補(bǔ)償策略Compensation Strategies。固定策略庫針對(duì)常見的失敗模式預(yù)先編寫好處理策略。例如“網(wǎng)絡(luò)超時(shí) - 等待2秒后重試最多3次”、“API返回格式錯(cuò)誤 - 嘗試用不同的解析器解析”、“A服務(wù)失敗 - 調(diào)用等價(jià)的B服務(wù)”。動(dòng)態(tài)策略生成對(duì)于未預(yù)見的失敗利用大語言模型LLM根據(jù)錯(cuò)誤信息、任務(wù)上下文和可用工具列表實(shí)時(shí)生成一個(gè)可能的補(bǔ)償計(jì)劃。例如LLM可以分析“用戶想查明天北京的天氣但天氣API故障。我可以嘗試從新聞網(wǎng)站抓取天氣相關(guān)的頭條新聞從中提取天氣信息作為近似參考?！辈呗赃x擇器則負(fù)責(zé)根據(jù)當(dāng)前的異常類型、任務(wù)緊急程度、可用資源如API調(diào)用次數(shù)、時(shí)間預(yù)算等因素從策略庫中選擇最合適的一個(gè)或多個(gè)策略。這可以是一個(gè)簡(jiǎn)單的規(guī)則引擎if-else也可以是一個(gè)訓(xùn)練過的分類模型。3. 補(bǔ)償執(zhí)行與狀態(tài)回滾管理這是補(bǔ)償?shù)摹笆帜_”。執(zhí)行補(bǔ)償策略時(shí)可能會(huì)改變智能體的狀態(tài)如變量、記憶。我們需要謹(jǐn)慎管理這些變更。原子性與事務(wù)對(duì)于復(fù)雜的補(bǔ)償操作可能需要確保一系列動(dòng)作要么全部成功要么全部失敗并回滾到補(bǔ)償前的狀態(tài)避免狀態(tài)不一致。這在多步驟數(shù)據(jù)操作中尤為重要。副作用管理有些操作具有副作用如發(fā)送了郵件、創(chuàng)建了訂單。補(bǔ)償操作可能需要撤銷這些副作用如發(fā)送更正郵件、取消訂單這需要工具本身提供逆向操作或在設(shè)計(jì)時(shí)就考慮“可補(bǔ)償性”。執(zhí)行追蹤詳細(xì)記錄補(bǔ)償事件的發(fā)生時(shí)間、原因、采取的策略及結(jié)果。這對(duì)于后續(xù)調(diào)試、優(yōu)化策略庫以及向用戶提供透明解釋都至關(guān)重要。在LangGraph這類基于狀態(tài)圖的框架中補(bǔ)償可以很好地建模為圖中的特殊邊或節(jié)點(diǎn)。當(dāng)某個(gè)主任務(wù)節(jié)點(diǎn)失敗或輸出不達(dá)標(biāo)時(shí)流程可以自動(dòng)路由到對(duì)應(yīng)的“補(bǔ)償節(jié)點(diǎn)”執(zhí)行完后再?zèng)Q定是返回主流程、嘗試另一條路徑還是最終失敗。3. 基于LangChain/LangGraph的RAC實(shí)戰(zhàn)實(shí)現(xiàn)理論講了不少現(xiàn)在我們進(jìn)入實(shí)戰(zhàn)環(huán)節(jié)。我將以一個(gè)具體的場(chǎng)景為例展示如何使用LangChain和LangGraph構(gòu)建一個(gè)具備基礎(chǔ)RAC能力的智能體。我們的場(chǎng)景是一個(gè)旅行規(guī)劃智能體它能根據(jù)用戶需求查詢航班、酒店并在某項(xiàng)服務(wù)失敗時(shí)嘗試補(bǔ)償。3.1 智能體基礎(chǔ)架構(gòu)與工具定義首先我們定義智能體所需的核心工具。為了模擬真實(shí)環(huán)境我們會(huì)創(chuàng)建一些可能失敗的工具。from langchain.tools import Tool, tool from langchain.agents import AgentExecutor, create_react_agent from langchain_core.prompts import PromptTemplate from langchain_openai import ChatOpenAI import random import time # 模擬一個(gè)不穩(wěn)定的航班查詢工具 tool def search_flights(destination: str, date: str) - str: 查詢指定目的地和日期的航班信息。 這是一個(gè)模擬工具有30%的概率模擬失敗返回錯(cuò)誤或空結(jié)果。 time.sleep(1) # 模擬網(wǎng)絡(luò)延遲 if random.random() 0.3: # 30%失敗率 # 模擬不同類型的失敗 fail_type random.choice([timeout, no_result, format_error]) if fail_type timeout: raise TimeoutError(Flight search API timed out.) elif fail_type no_result: return No flights found for the given criteria. else: return {\error\: \Invalid response format\} # 返回格式錯(cuò)誤的數(shù)據(jù) # 模擬成功返回 flights [ {airline: AirFast, price: 1200, departure: 08:00}, {airline: SkyHigh, price: 950, departure: 14:30} ] return fFound flights to {destination} on {date}: {flights} # 模擬一個(gè)備選的航班查詢工具補(bǔ)償用 tool def search_flights_backup(destination: str, date: str) - str: 補(bǔ)償工具使用備用數(shù)據(jù)源查詢航班信息。速度較慢但更穩(wěn)定。 time.sleep(2) # 備份服務(wù)更慢 flights [ {airline: GlobalAir (Backup), price: 1100, departure: 10:15}, ] return fFound flights from backup source to {destination} on {date}: {flights} # 酒店查詢工具相對(duì)穩(wěn)定 tool def search_hotels(destination: str, date: str) - str: 查詢指定目的地和日期的酒店信息。 time.sleep(0.5) hotels [{name: Grand Plaza, price: 200}, {name: Cozy Inn, price: 120}] return fFound hotels in {destination} for {date}: {hotels} # 定義一個(gè)補(bǔ)償建議工具由LLM驅(qū)動(dòng) tool def suggest_compensation(failed_task: str, error_info: str, context: str) - str: 根據(jù)失敗的任務(wù)、錯(cuò)誤信息和上下文建議補(bǔ)償方案。 這是一個(gè)元工具它本身會(huì)調(diào)用LLM進(jìn)行分析。 llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt PromptTemplate.from_template( 當(dāng)前任務(wù)上下文{context} 失敗的任務(wù)描述{failed_task} 遇到的錯(cuò)誤或問題{error_info} 請(qǐng)分析情況并建議一個(gè)具體的、可操作的補(bǔ)償方案。方案應(yīng)基于以下可用工具 - search_flights_backup: 備用航班查詢 - search_hotels: 酒店查詢 - 或者建議調(diào)整用戶需求例如更改日期、選擇附近目的地。 只輸出補(bǔ)償方案描述不要輸出其他內(nèi)容。 ) chain prompt | llm suggestion chain.invoke({ context: context, failed_task: failed_task, error_info: error_info }) return suggestion.content # 將工具裝入列表 tools [search_flights, search_flights_backup, search_hotels, suggest_compensation]注意在實(shí)際項(xiàng)目中suggest_compensation工具的實(shí)現(xiàn)需要仔細(xì)設(shè)計(jì)提示詞Prompt確保LLM給出的建議是具體、安全且可執(zhí)行的。避免讓LLM建議調(diào)用不存在的工具或執(zhí)行危險(xiǎn)操作。3.2 構(gòu)建具備補(bǔ)償邏輯的智能體執(zhí)行循環(huán)接下來我們不使用最簡(jiǎn)單的AgentExecutor而是構(gòu)建一個(gè)更定制的執(zhí)行循環(huán)以便在每一步插入補(bǔ)償邏輯。這里我們用LangGraph來更直觀地控制流程。from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List import operator from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langgraph.prebuilt import ToolExecutor, ToolInvocation import json # 定義智能體的狀態(tài)結(jié)構(gòu) class AgentState(TypedDict): messages: Annotated[List, operator.add] # 消息歷史 context: str # 任務(wù)上下文摘要 last_task_failed: bool # 上一步是否失敗 failure_info: str # 失敗信息 compensation_attempted: bool # 是否已嘗試補(bǔ)償 # 初始化工具執(zhí)行器 tool_executor ToolExecutor(tools) # 1. 定義“規(guī)劃節(jié)點(diǎn)”決定下一步做什么 def plan_node(state: AgentState) - AgentState: 根據(jù)當(dāng)前狀態(tài)和對(duì)話歷史決定下一步行動(dòng)調(diào)用工具或結(jié)束。 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 構(gòu)建系統(tǒng)提示包含工具描述和補(bǔ)償引導(dǎo) system_prompt 你是一個(gè)旅行規(guī)劃助手并且具備問題補(bǔ)償能力。你的目標(biāo)是為用戶完成旅行規(guī)劃。 你可以使用以下工具 - search_flights: 查詢航班。注意此工具可能不穩(wěn)定。 - search_hotels: 查詢酒店。 - search_flights_backup: 備用查詢航班更穩(wěn)定但較慢。 - suggest_compensation: 當(dāng)任務(wù)遇到問題時(shí)此工具可以分析情況并建議如何補(bǔ)償。 特殊指導(dǎo)如果你使用search_flights工具后得到的結(jié)果是錯(cuò)誤、超時(shí)或“No flights found”這表示任務(wù)遇到了障礙。 此時(shí)你不應(yīng)立即告訴用戶失敗而應(yīng)嘗試以下補(bǔ)償策略之一 策略A立即使用search_flights_backup工具再試一次。 策略B調(diào)用suggest_compensation工具獲取建議然后根據(jù)建議行動(dòng)。 策略C如果航班確實(shí)無法解決轉(zhuǎn)向先完成酒店查詢等其他可完成的部分。 請(qǐng)根據(jù)對(duì)話歷史決定下一步。如果用戶需求已滿足或無法進(jìn)一步滿足則回復(fù)最終答案。 messages [{role: system, content: system_prompt}] state[messages] response llm.invoke(messages) state[messages].append(AIMessage(contentresponse.content)) return state # 2. 定義“執(zhí)行節(jié)點(diǎn)”執(zhí)行工具調(diào)用并判斷結(jié)果是否需要補(bǔ)償 def execute_node(state: AgentState) - AgentState: 執(zhí)行AI消息中的工具調(diào)用并處理結(jié)果。 last_message state[messages][-1] if not isinstance(last_message, AIMessage) or not last_message.tool_calls: # 如果不是工具調(diào)用直接返回 return state tool_calls last_message.tool_calls tool_invocations [ToolInvocation(idtc[id], tooltc[name], argstc[args]) for tc in tool_calls] # 執(zhí)行工具 try: tool_outputs tool_executor.batch(tool_invocations) except Exception as e: # 捕獲執(zhí)行期異常 tool_outputs [fTool execution error: {str(e)} for _ in tool_invocations] # 將結(jié)果封裝為ToolMessage tool_messages [] needs_compensation False failure_info for invocation, output in zip(tool_invocations, tool_outputs): tool_messages.append(ToolMessage(contentstr(output), tool_call_idinvocation.id)) # 判斷輸出是否需要觸發(fā)補(bǔ)償 # 規(guī)則1工具執(zhí)行拋出異常 # 規(guī)則2輸出包含特定的錯(cuò)誤關(guān)鍵詞根據(jù)工具特性定制 if invocation.tool search_flights: if error in str(output).lower() or timeout in str(output).lower() or No flights found in str(output): needs_compensation True failure_info fFlight search failed with output: {output} print(f[RAC Triggered] Flight search failed. Info: {failure_info}) state[messages].extend(tool_messages) state[last_task_failed] needs_compensation state[failure_info] failure_info if needs_compensation else # 如果失敗且尚未嘗試補(bǔ)償則設(shè)置標(biāo)志后續(xù)路由到補(bǔ)償節(jié)點(diǎn) if needs_compensation and not state.get(compensation_attempted, False): state[compensation_attempted] True # 標(biāo)記已嘗試補(bǔ)償防止無限循環(huán) print([RAC] Routing to compensation node.) else: state[compensation_attempted] False # 重置標(biāo)志 return state # 3. 定義“補(bǔ)償決策節(jié)點(diǎn)” def compensation_node(state: AgentState) - AgentState: 根據(jù)失敗信息決定并執(zhí)行補(bǔ)償動(dòng)作。 if not state[last_task_failed]: return state print(f[Compensation Node] Handling failure: {state[failure_info]}) # 這里實(shí)現(xiàn)一個(gè)簡(jiǎn)單的補(bǔ)償策略直接調(diào)用備用航班查詢 # 在實(shí)際中這里可以調(diào)用suggest_compensation工具或者有一個(gè)更復(fù)雜的策略選擇器 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 構(gòu)建一個(gè)強(qiáng)制調(diào)用備用工具的AI消息 compensation_message AIMessage( content, tool_calls[{ id: comp_1, name: search_flights_backup, args: {destination: Beijing, date: 2024-10-01} # 參數(shù)應(yīng)從上下文中解析此處簡(jiǎn)化 }] ) state[messages].append(compensation_message) # 注意這里添加消息后需要下一個(gè)execute_node來實(shí)際執(zhí)行它。 # 在圖中我們會(huì)讓補(bǔ)償節(jié)點(diǎn)后連接回執(zhí)行節(jié)點(diǎn)。 return state # 4. 構(gòu)建LangGraph workflow StateGraph(AgentState) # 添加節(jié)點(diǎn) workflow.add_node(plan, plan_node) # 規(guī)劃下一步 workflow.add_node(execute, execute_node) # 執(zhí)行工具 workflow.add_node(compensate, compensation_node) # 執(zhí)行補(bǔ)償決策 # 設(shè)置邊和路由邏輯 workflow.set_entry_point(plan) # 從plan節(jié)點(diǎn)出來總是去execute節(jié)點(diǎn)執(zhí)行工具 workflow.add_edge(plan, execute) # 從execute節(jié)點(diǎn)出來根據(jù)狀態(tài)決定下一步 def decide_after_execute(state: AgentState) - str: 決定執(zhí)行后的路由是否需要補(bǔ)償還是繼續(xù)規(guī)劃或結(jié)束。 # 如果上一步失敗了且尚未嘗試補(bǔ)償則去補(bǔ)償節(jié)點(diǎn) if state.get(last_task_failed, False) and state.get(compensation_attempted, False): return compensate # 否則繼續(xù)下一輪規(guī)劃除非有結(jié)束條件比如用戶說謝謝 else: # 這里可以添加更復(fù)雜的結(jié)束判斷例如檢測(cè)到最終答案 last_msg_content state[messages][-1].content if state[messages] else if final in last_msg_content.lower() or thank you in last_msg_content.lower(): return END return plan workflow.add_conditional_edges( execute, decide_after_execute, { compensate: compensate, plan: plan, END: END } ) # 補(bǔ)償節(jié)點(diǎn)執(zhí)行完后應(yīng)回到執(zhí)行節(jié)點(diǎn)去運(yùn)行它產(chǎn)生的工具調(diào)用 workflow.add_edge(compensate, execute) # 編譯圖 app workflow.compile()這個(gè)圖結(jié)構(gòu)形成了一個(gè)核心循環(huán)Plan - Execute - (如果需要補(bǔ)償) - Compensate - Execute - Plan ...。補(bǔ)償節(jié)點(diǎn)compensate被設(shè)計(jì)為一個(gè)獨(dú)立的決策點(diǎn)它可以基于更復(fù)雜的邏輯如調(diào)用LLM分析來生成具體的補(bǔ)償動(dòng)作在這里是直接調(diào)用備用工具。3.3 運(yùn)行與測(cè)試現(xiàn)在讓我們運(yùn)行這個(gè)智能體看看RAC是如何工作的。# 初始化狀態(tài) initial_state AgentState( messages[HumanMessage(contentI want to plan a trip to Beijing on October 1st. Find me a flight and a hotel.)], contextUser wants flight and hotel for Beijing on 2024-10-01., last_task_failedFalse, failure_info, compensation_attemptedFalse ) # 運(yùn)行圖 final_state None for step, s in app.stream(initial_state, stream_modevalues, subgraphsTrue): node_name list(s.keys())[0] print(f\n--- Step: {node_name} ---) last_msg s[node_name][messages][-1] if s[node_name][messages] else None if last_msg: if isinstance(last_msg, AIMessage) and last_msg.tool_calls: print(fAI decided to use tool(s): {[tc[name] for tc in last_msg.tool_calls]}) elif isinstance(last_msg, ToolMessage): print(fTool returned: {last_msg.content[:100]}...) # 打印前100字符 else: print(fMessage: {last_msg.content[:150]}...) if s[node_name].get(last_task_failed): print(f*** Last task failed: {s[node_name].get(failure_info)} ***) final_state s[node_name] if s in locals() else initial_state print(\n Final Conversation ) for msg in final_state[messages]: if isinstance(msg, HumanMessage): print(fUser: {msg.content}) elif isinstance(msg, AIMessage) and not msg.tool_calls: print(fAssistant: {msg.content}) elif isinstance(msg, ToolMessage): print(f[Tool Result]: {msg.content[:80]}...)在一次模擬運(yùn)行中你可能會(huì)看到如下輸出--- Step: plan --- AI decided to use tool(s): [search_flights] --- Step: execute --- Tool returned: Tool execution error: Flight search API timed out. *** Last task failed: Flight search failed with output: Tool execution error: Flight searc... [RAC Triggered] Flight search failed. Info: Flight search failed with output: Tool execution error: Flight search API timed out. [RAC] Routing to compensation node. --- Step: compensate --- [Compensation Node] Handling failure: Flight search failed with output: Tool execution error: Flight search API timed out. --- Step: execute --- AI decided to use tool(s): [search_flights_backup] Tool returned: Found flights from backup source to Beijing on 2024-10-01: [{airline: GlobalAir (Backup), price: 1100, departure: 10:15}]... --- Step: plan --- AI decided to use tool(s): [search_hotels] ...從輸出可以看到當(dāng)search_flights工具模擬超時(shí)失敗后狀態(tài)被標(biāo)記為失敗流程被路由到compensate節(jié)點(diǎn)。該節(jié)點(diǎn)決策后生成了一個(gè)調(diào)用search_flights_backup工具的新指令隨后流程回到execute節(jié)點(diǎn)成功執(zhí)行了備用查詢。最終智能體成功完成了酒店查詢并向用戶提供了包含備用航班信息的完整旅行方案。整個(gè)過程中用戶感知到的可能只是一次稍慢的查詢而非任務(wù)失敗。4. 高級(jí)補(bǔ)償策略與模式探討上面我們實(shí)現(xiàn)了一個(gè)基礎(chǔ)的、規(guī)則驅(qū)動(dòng)的補(bǔ)償流程。但在復(fù)雜的生產(chǎn)環(huán)境中我們需要更智能、更靈活的補(bǔ)償策略。下面探討幾種高級(jí)模式。4.1 基于LLM的動(dòng)態(tài)補(bǔ)償策略生成在前面的例子中補(bǔ)償節(jié)點(diǎn)是直接調(diào)用備用工具。更高級(jí)的做法是讓LLM擔(dān)任“補(bǔ)償策略師”。我們可以修改compensation_node使其調(diào)用suggest_compensation工具然后解析LLM的建議并轉(zhuǎn)化為實(shí)際行動(dòng)。def advanced_compensation_node(state: AgentState) - AgentState: 使用LLM分析失敗并生成動(dòng)態(tài)補(bǔ)償策略。 if not state[last_task_failed]: return state print(f[Advanced Compensation] Analyzing failure with LLM...) # 1. 調(diào)用建議工具 suggestion suggest_compensation.invoke({ failed_task: Search for flights to Beijing on 2024-10-01, error_info: state[failure_info], context: state[context] }) print(f[LLM Suggestion]: {suggestion}) # 2. 解析LLM的建議并轉(zhuǎn)化為具體的工具調(diào)用這里需要更復(fù)雜的解析邏輯 # 例如LLM可能返回“建議使用備用航班查詢工具(search_flights_backup)再試一次?！?# 或者“建議先查詢酒店并告知用戶航班查詢遇到技術(shù)問題稍后再試。” # 這里是一個(gè)簡(jiǎn)化的示例假設(shè)LLM建議調(diào)用備用工具。 # 在實(shí)際中你需要一個(gè)更魯棒的解析器可能還需要另一個(gè)LLM調(diào)用來將自然語言建議轉(zhuǎn)化為工具調(diào)用。 # 簡(jiǎn)化處理如果建議中提到“backup”則調(diào)用備用工具 if backup in suggestion.lower(): compensation_action AIMessage( content, tool_calls[{ id: comp_llm_1, name: search_flights_backup, args: {destination: Beijing, date: 2024-10-01} }] ) state[messages].append(compensation_action) else: # 如果LLM建議其他操作如直接回復(fù)用戶則添加一個(gè)普通AIMessage state[messages].append(AIMessage(contentf[Compensation Plan] {suggestion})) return state注意讓LLM直接生成工具調(diào)用存在安全風(fēng)險(xiǎn)如工具注入攻擊。更安全的做法是讓LLM從一個(gè)預(yù)定義的、安全的補(bǔ)償策略列表中選擇或者使用嚴(yán)格的輸出解析如Pydantic來確保生成的指令是合法的。4.2 多智能體協(xié)作中的補(bǔ)償協(xié)商在由多個(gè)智能體組成的系統(tǒng)中一個(gè)智能體的失敗可能需要其他智能體協(xié)助補(bǔ)償。這引入了“協(xié)商”機(jī)制。場(chǎng)景智能體A負(fù)責(zé)航班預(yù)訂智能體B負(fù)責(zé)酒店預(yù)訂。A發(fā)現(xiàn)航班售罄。補(bǔ)償流程A將失敗事件和上下文廣播給系統(tǒng)內(nèi)的其他智能體或一個(gè)專用的“協(xié)調(diào)者”智能體。智能體B接收到信息后評(píng)估自身能力“我無法解決航班問題但我可以優(yōu)先確保酒店預(yù)訂成功并建議用戶調(diào)整日期。”協(xié)調(diào)者收集所有反饋可能決策“鑒于航班問題建議將旅行日期推遲一天。請(qǐng)智能體A查詢新日期的航班智能體B重新查詢酒店?！敝悄荏wA和B執(zhí)行新的子任務(wù)。在LangGraph中這可以通過多個(gè)并行的智能體子圖加上一個(gè)中央?yún)f(xié)調(diào)狀態(tài)來實(shí)現(xiàn)。補(bǔ)償事件會(huì)觸發(fā)狀態(tài)更新協(xié)調(diào)邏輯根據(jù)狀態(tài)決定如何重新路由任務(wù)或修改任務(wù)目標(biāo)。4.3 補(bǔ)償策略的評(píng)估與學(xué)習(xí)一個(gè)成熟的RAC系統(tǒng)應(yīng)該能從歷史補(bǔ)償案例中學(xué)習(xí)。我們可以記錄每次補(bǔ)償事件元數(shù)據(jù)失敗任務(wù)、錯(cuò)誤類型、觸發(fā)的補(bǔ)償策略、補(bǔ)償后的結(jié)果成功/失敗、最終任務(wù)完成度。學(xué)習(xí)機(jī)制策略優(yōu)先級(jí)調(diào)整如果某個(gè)補(bǔ)償策略如“重試3次”在特定錯(cuò)誤類型如“網(wǎng)絡(luò)超時(shí)”上成功率很高則提高其優(yōu)先級(jí)。策略庫擴(kuò)充對(duì)于反復(fù)出現(xiàn)且現(xiàn)有策略無法很好處理的新失敗模式可以人工或通過LLM總結(jié)生成新的補(bǔ)償策略加入策略庫。參數(shù)調(diào)優(yōu)例如自動(dòng)調(diào)整“重試”策略的等待間隔和次數(shù)上限。這可以通過一個(gè)簡(jiǎn)單的獎(jiǎng)勵(lì)機(jī)制來實(shí)現(xiàn)補(bǔ)償后任務(wù)最終成功則給該次補(bǔ)償策略“加分”反之則“扣分”。長(zhǎng)期來看系統(tǒng)會(huì)傾向于選擇成功率更高的策略。5. 常見問題、挑戰(zhàn)與避坑指南在實(shí)際開發(fā)中實(shí)現(xiàn)有效的RAC會(huì)遇到不少挑戰(zhàn)。以下是一些常見問題及解決思路。5.1 補(bǔ)償循環(huán)與無限遞歸問題智能體陷入“失敗-補(bǔ)償-再失敗-再補(bǔ)償”的死循環(huán)。例如備用工具也失敗補(bǔ)償策略又選擇重試主工具如此反復(fù)。解決方案設(shè)置補(bǔ)償預(yù)算為整個(gè)任務(wù)或單個(gè)子任務(wù)設(shè)置最大的補(bǔ)償嘗試次數(shù)如最多3次或總時(shí)間預(yù)算。多樣化策略確保補(bǔ)償策略庫中有不同的策略如重試、替換、降級(jí)目標(biāo)避免每次都用同一招。狀態(tài)記錄在智能體狀態(tài)中清晰記錄已嘗試過的補(bǔ)償動(dòng)作避免重復(fù)執(zhí)行無效操作。在我們的示例中compensation_attempted標(biāo)志就是一個(gè)簡(jiǎn)單的防循環(huán)機(jī)制。引入隨機(jī)退避對(duì)于重試類策略采用指數(shù)退避算法增加重試間隔避免加重服務(wù)負(fù)擔(dān)。5.2 補(bǔ)償動(dòng)作的副作用管理問題某些工具調(diào)用具有不可逆的副作用如發(fā)送郵件、支付扣款。如果補(bǔ)償涉及“撤銷”操作但撤銷本身也可能失敗。解決方案設(shè)計(jì)冪等且可逆的工具在工具設(shè)計(jì)階段就考慮補(bǔ)償需求。例如支付工具提供“預(yù)授權(quán)”和“確認(rèn)”兩步在確認(rèn)前可以安全取消。使用Saga模式對(duì)于跨多個(gè)服務(wù)的分布式事務(wù)為每個(gè)服務(wù)設(shè)計(jì)補(bǔ)償事務(wù)Compensating Transaction。如果后續(xù)步驟失敗則按相反順序執(zhí)行補(bǔ)償事務(wù)來回滾。這在復(fù)雜的業(yè)務(wù)工作流中很常見。人工審核介入點(diǎn)對(duì)于高風(fēng)險(xiǎn)操作補(bǔ)償策略不是自動(dòng)執(zhí)行而是生成需要人工審核的建議如“建議人工聯(lián)系客服取消訂單”。5.3 補(bǔ)償策略的可靠性與安全性問題由LLM動(dòng)態(tài)生成的補(bǔ)償策略可能不可靠或不安全例如建議調(diào)用一個(gè)不存在的工具或執(zhí)行一個(gè)成本極高的操作。解決方案沙箱與驗(yàn)證在安全沙箱中模擬執(zhí)行LLM生成的補(bǔ)償計(jì)劃評(píng)估其資源消耗和潛在影響再?zèng)Q定是否在真實(shí)環(huán)境執(zhí)行。策略白名單只允許LLM從一組經(jīng)過嚴(yán)格審查和測(cè)試的預(yù)定義策略中選擇而不是自由生成。成本限制為智能體設(shè)置明確的資源限制如API調(diào)用次數(shù)上限、最大財(cái)務(wù)成本任何補(bǔ)償策略都不能突破這些限制。5.4 用戶體驗(yàn)與透明度問題智能體在后臺(tái)進(jìn)行了多次補(bǔ)償嘗試用戶卻長(zhǎng)時(shí)間得不到反饋體驗(yàn)不佳。解決方案漸進(jìn)式披露對(duì)于短暫的、預(yù)期內(nèi)的失敗如網(wǎng)絡(luò)抖動(dòng)重試可以不打擾用戶。對(duì)于需要更長(zhǎng)時(shí)間或更換方案的補(bǔ)償應(yīng)及時(shí)通知用戶。例如“正在查詢航班當(dāng)前線路繁忙正在嘗試備用方案...”解釋性輸出任務(wù)完成后可以簡(jiǎn)要告知用戶遇到的挑戰(zhàn)和采取的解決方案。例如“為您找到了航班。最初查詢時(shí)遇到臨時(shí)問題已通過備用渠道成功獲取信息?！边@能建立信任感。提供選擇權(quán)對(duì)于重大的補(bǔ)償方案如更改旅行日期應(yīng)將多個(gè)選項(xiàng)呈現(xiàn)給用戶讓其做出最終決定而不是完全自主決策。5.5 性能開銷問題補(bǔ)償邏輯增加了判斷、決策和執(zhí)行的開銷可能影響智能體的響應(yīng)速度。解決方案異步補(bǔ)償對(duì)于非關(guān)鍵路徑或耗時(shí)長(zhǎng)的補(bǔ)償操作可以將其放入后臺(tái)異步執(zhí)行主流程先返回一個(gè)中間狀態(tài)給用戶。熱點(diǎn)策略緩存將高頻使用的、成功的補(bǔ)償策略及其結(jié)果緩存起來當(dāng)下次遇到相同的失敗模式時(shí)可以直接使用緩存結(jié)果跳過LLM分析和策略選擇。監(jiān)控與優(yōu)化密切監(jiān)控補(bǔ)償觸發(fā)的頻率和耗時(shí)持續(xù)優(yōu)化策略選擇算法和工具性能從根源上減少失敗的發(fā)生。實(shí)現(xiàn)Robust Agent Compensation是一個(gè)持續(xù)迭代的過程。它沒有銀彈需要開發(fā)者深入理解自己的業(yè)務(wù)場(chǎng)景、工具特性和失敗模式。從簡(jiǎn)單的重試機(jī)制開始逐步引入更智能的策略和協(xié)作機(jī)制是構(gòu)建真正健壯、可信賴的AI智能體應(yīng)用的必經(jīng)之路。