建復(fù)雜任務(wù)分解與模塊化執(zhí)行架構(gòu))
1. 項(xiàng)目概述什么是“分層組合性”智能體最近在搗鼓AI智能體AI Agent項(xiàng)目時(shí)我一直在思考一個(gè)問(wèn)題如何讓一個(gè)智能體真正變得“有用”而不僅僅是能回答幾個(gè)問(wèn)題我們經(jīng)??吹揭恍┭菔局悄荏w可以調(diào)用工具、執(zhí)行簡(jiǎn)單任務(wù)比如查天氣、發(fā)郵件。但一旦任務(wù)稍微復(fù)雜一點(diǎn)比如“幫我規(guī)劃一個(gè)周末的短途旅行預(yù)算控制在2000元以?xún)?nèi)要包含交通、住宿、景點(diǎn)和美食推薦并生成一份可分享的行程單”很多智能體就“卡殼”了。它們要么只能完成其中一步比如只查了天氣要么生成一個(gè)籠統(tǒng)、無(wú)法執(zhí)行的計(jì)劃。這背后的核心瓶頸往往不是大模型本身的能力不足而是智能體的“思考”和“行動(dòng)”方式缺乏有效的結(jié)構(gòu)。這正是“分層組合性”Hierarchical Compositionality要解決的問(wèn)題。簡(jiǎn)單來(lái)說(shuō)它就像樂(lè)高積木。你有一堆基礎(chǔ)積木塊原子技能比如“查詢(xún)交通信息”、“篩選酒店”、“總結(jié)景點(diǎn)介紹”。分層組合性就是一套設(shè)計(jì)圖紙和組裝邏輯它告訴智能體面對(duì)一個(gè)宏大目標(biāo)比如“規(guī)劃旅行”你應(yīng)該如何將它逐層分解成子目標(biāo)“規(guī)劃交通”、“預(yù)訂住宿”、“安排行程”每個(gè)子目標(biāo)又如何由更小的任務(wù)“查詢(xún)北京到上海的航班”、“對(duì)比三家酒店的價(jià)格和評(píng)分”組合而成最終調(diào)用那些基礎(chǔ)的積木塊來(lái)具體執(zhí)行。所以“分層組合性輔助AI智能體”這個(gè)項(xiàng)目本質(zhì)上是在構(gòu)建一個(gè)具備“任務(wù)分解”與“模塊化執(zhí)行”能力的高級(jí)助手。它不再是一個(gè)簡(jiǎn)單的“問(wèn)答機(jī)”或“單步工具調(diào)用器”而是一個(gè)能理解復(fù)雜用戶(hù)意圖、自主制定分層計(jì)劃、并協(xié)調(diào)各種技能可靠執(zhí)行的“大腦”。這對(duì)于實(shí)現(xiàn)真正意義上的個(gè)人助理、自動(dòng)化工作流、甚至垂直領(lǐng)域的專(zhuān)業(yè)顧問(wèn)都至關(guān)重要。無(wú)論你是開(kāi)發(fā)者想構(gòu)建更強(qiáng)大的Agent還是技術(shù)愛(ài)好者想理解下一代AI應(yīng)用的走向搞懂分層組合性都是關(guān)鍵一步。2. 核心設(shè)計(jì)思路為何“分層”與“組合”是智能體的靈魂要理解這個(gè)設(shè)計(jì)我們可以把它和傳統(tǒng)的智能體架構(gòu)做個(gè)對(duì)比。目前最常見(jiàn)的智能體框架比如基于ReActReasoning Acting模式或類(lèi)似AutoGPT的架構(gòu)其工作流通常是線性的感知用戶(hù)輸入→ 思考生成一個(gè)動(dòng)作或回復(fù)→ 行動(dòng)執(zhí)行該動(dòng)作→ 觀察獲取結(jié)果→ 循環(huán)。這種模式在處理明確、單步的任務(wù)時(shí)很有效但面對(duì)復(fù)雜任務(wù)時(shí)問(wèn)題就暴露了。2.1 傳統(tǒng)智能體的局限陷入“思維迷宮”想象一下你讓一個(gè)傳統(tǒng)架構(gòu)的智能體去完成那個(gè)旅行規(guī)劃任務(wù)。它可能會(huì)這樣“思考”用戶(hù)要旅行規(guī)劃。我需要先查天氣。行動(dòng)調(diào)用天氣API天氣查完了。接下來(lái)需要交通信息。行動(dòng)調(diào)用交通查詢(xún)API但用戶(hù)沒(méi)說(shuō)出發(fā)地和目的地它可能卡住或者胡亂猜一個(gè)如果幸運(yùn)地查到了交通接下來(lái)需要酒店。預(yù)算2000元怎么分配它沒(méi)有這個(gè)概念。你會(huì)發(fā)現(xiàn)它的“思考”是短視的、缺乏全局規(guī)劃的。它像一個(gè)在迷宮里亂撞的人走一步看一步很容易陷入死胡同比如缺少關(guān)鍵參數(shù)或者在一個(gè)細(xì)節(jié)上無(wú)限循環(huán)比如反復(fù)比較兩個(gè)無(wú)關(guān)緊要的酒店。這是因?yàn)槿狈σ粋€(gè)頂層的、指導(dǎo)性的“計(jì)劃”。2.2 分層組合性的破局之道自上而下的藍(lán)圖繪制分層組合性智能體則采用了完全不同的思路。它的核心流程可以概括為目標(biāo)分解 → 技能組合 → 分層執(zhí)行 → 統(tǒng)一協(xié)調(diào)。目標(biāo)分解Decomposition智能體接收到復(fù)雜指令后第一件事不是急于行動(dòng)而是進(jìn)行“戰(zhàn)略規(guī)劃”。利用大模型的推理能力將頂層目標(biāo)Goal分解為一個(gè)樹(shù)狀結(jié)構(gòu)的任務(wù)層級(jí)Task Hierarchy。例如根任務(wù)規(guī)劃一個(gè)預(yù)算2000元的周末短途旅行。一級(jí)子任務(wù)交通規(guī)劃、住宿安排、景點(diǎn)與美食規(guī)劃、行程單整合。二級(jí)子任務(wù)以交通規(guī)劃為例確認(rèn)出發(fā)地與目的地、查詢(xún)交通方式與價(jià)格、根據(jù)時(shí)間和預(yù)算選擇最優(yōu)方案。三級(jí)子任務(wù)以查詢(xún)交通為例調(diào)用航班查詢(xún)工具、調(diào)用火車(chē)票查詢(xún)工具。技能組合Composition這個(gè)任務(wù)樹(shù)上的每一個(gè)葉子節(jié)點(diǎn)不可再分的最小任務(wù)都需要對(duì)應(yīng)一個(gè)或多個(gè)可執(zhí)行的“技能”Skill。技能是智能體的原子能力通常對(duì)應(yīng)一個(gè)工具函數(shù)Tool Function、一段代碼或一個(gè)API調(diào)用。設(shè)計(jì)的關(guān)鍵在于技能要足夠“原子化”和“可復(fù)用”。例如“查詢(xún)航班信息”是一個(gè)技能“查詢(xún)酒店價(jià)格”是另一個(gè)技能。而“規(guī)劃交通”這個(gè)高層任務(wù)則是通過(guò)組合“確認(rèn)地點(diǎn)”、“查詢(xún)航班”、“查詢(xún)火車(chē)”、“比價(jià)決策”等多個(gè)技能的邏輯來(lái)實(shí)現(xiàn)的。分層執(zhí)行Hierarchical Execution智能體從任務(wù)樹(shù)的根部或某個(gè)高層節(jié)點(diǎn)開(kāi)始但它并不直接執(zhí)行所有葉子任務(wù)。它采用一種“管理者”模式高層任務(wù)負(fù)責(zé)協(xié)調(diào)和決策將具體工作委派給低層任務(wù)。例如“交通規(guī)劃”這個(gè)節(jié)點(diǎn)會(huì)先調(diào)用“確認(rèn)地點(diǎn)”技能獲得“北京”到“上海”的信息后再并發(fā)調(diào)用“查詢(xún)航班”和“查詢(xún)火車(chē)”技能最后根據(jù)結(jié)果和預(yù)算調(diào)用“決策”邏輯可能也是一個(gè)簡(jiǎn)單的規(guī)則或模型來(lái)選擇方案。這個(gè)過(guò)程中每一層都只關(guān)心自己的輸入、輸出和子任務(wù)的調(diào)度。統(tǒng)一協(xié)調(diào)Orchestration一個(gè)中央調(diào)度器或稱(chēng)為“執(zhí)行引擎”負(fù)責(zé)監(jiān)控整個(gè)任務(wù)樹(shù)的執(zhí)行狀態(tài)。它處理子任務(wù)之間的依賴(lài)關(guān)系例如必須先有地點(diǎn)信息才能查交通、處理執(zhí)行失敗的重試或回退、管理上下文信息將子任務(wù)的結(jié)果傳遞給父任務(wù)并最終將各個(gè)部分的結(jié)果整合成完整的輸出。這種設(shè)計(jì)的巨大優(yōu)勢(shì)在于可解釋性強(qiáng)整個(gè)解決問(wèn)題的過(guò)程像一份清晰的項(xiàng)目計(jì)劃書(shū)你可以看到任務(wù)是如何被拆解和完成的哪里出了問(wèn)題一目了然。魯棒性高某個(gè)子任務(wù)失敗比如某個(gè)API暫時(shí)不可用智能體可以在該層級(jí)嘗試備用方案而不至于導(dǎo)致整個(gè)任務(wù)崩潰。可擴(kuò)展性好要增加智能體的能力只需要開(kāi)發(fā)新的原子技能并將其注冊(cè)到技能庫(kù)中。高層任務(wù)可以像搭積木一樣組合這些新技能無(wú)需重寫(xiě)核心邏輯。效率提升通過(guò)識(shí)別可以并行執(zhí)行的獨(dú)立子任務(wù)如同時(shí)查詢(xún)航班和火車(chē)可以顯著縮短整體執(zhí)行時(shí)間。實(shí)操心得分解的粒度是關(guān)鍵在目標(biāo)分解時(shí)最難把握的是分解的“粒度”。分得太粗如只分解到“安排行程”和沒(méi)分解一樣分得太細(xì)如把“點(diǎn)擊查詢(xún)按鈕”也作為一個(gè)任務(wù)會(huì)導(dǎo)致任務(wù)樹(shù)過(guò)于龐大調(diào)度開(kāi)銷(xiāo)激增且容易陷入細(xì)節(jié)。我的經(jīng)驗(yàn)是一個(gè)葉子任務(wù)應(yīng)該對(duì)應(yīng)一個(gè)有明確輸入輸出、能通過(guò)一次或一組確定的工具調(diào)用完成的單元。例如“獲取未來(lái)三天上海的天氣”是一個(gè)好的葉子任務(wù)“分析天氣”就不是因?yàn)樗幻鞔_具體要做什么。3. 架構(gòu)拆解如何構(gòu)建一個(gè)分層組合性智能體理論講完了我們來(lái)看看具體怎么搭。一個(gè)典型的分層組合性智能體系統(tǒng)通常包含以下幾個(gè)核心模塊我會(huì)結(jié)合一個(gè)簡(jiǎn)單的代碼框架思路來(lái)解釋。3.1 核心模塊構(gòu)成模塊名稱(chēng)職責(zé)關(guān)鍵技術(shù)點(diǎn)/實(shí)現(xiàn)選擇任務(wù)規(guī)劃器 (Task Planner)將用戶(hù)目標(biāo)解析并分解為層次化任務(wù)樹(shù)。這是智能體的“大腦皮層”。依賴(lài)大語(yǔ)言模型LLM。通過(guò)精心設(shè)計(jì)的Prompt讓LLM按照特定格式如JSON、YAML輸出結(jié)構(gòu)化的任務(wù)樹(shù)。通常需要提供技能庫(kù)的描述作為上下文。技能庫(kù) (Skill Library)存儲(chǔ)所有可用的原子技能及其元數(shù)據(jù)描述、輸入輸出格式、調(diào)用方式??梢杂靡粋€(gè)Python字典或?qū)iT(mén)的注冊(cè)表管理。每個(gè)技能對(duì)應(yīng)一個(gè)函數(shù)函數(shù)需要有清晰的文檔字符串docstring來(lái)描述其功能這會(huì)被用于Prompt工程。技能執(zhí)行器 (Skill Executor)負(fù)責(zé)調(diào)用技能庫(kù)中的具體函數(shù)并處理輸入輸出。通常利用Python的反射機(jī)制根據(jù)技能名動(dòng)態(tài)調(diào)用對(duì)應(yīng)的函數(shù)。需要做好錯(cuò)誤處理和類(lèi)型轉(zhuǎn)換。工作流引擎 (Workflow Engine)核心調(diào)度系統(tǒng)。負(fù)責(zé)遍歷任務(wù)樹(shù)管理任務(wù)狀態(tài)待執(zhí)行、執(zhí)行中、成功、失敗解析任務(wù)間的依賴(lài)決定執(zhí)行順序串行、并行??梢宰约簩?shí)現(xiàn)一個(gè)簡(jiǎn)單的狀態(tài)機(jī)也可以利用現(xiàn)成的輕量級(jí)工作流引擎如Prefect、Airflow的核心概念。對(duì)于復(fù)雜依賴(lài)需要引入有向無(wú)環(huán)圖DAG來(lái)管理。上下文管理器 (Context Manager)在任務(wù)執(zhí)行過(guò)程中傳遞和共享數(shù)據(jù)。例如子任務(wù)A輸出的“目的地城市”需要傳遞給子任務(wù)B作為輸入。維護(hù)一個(gè)全局或會(huì)話(huà)級(jí)的上下文字典。關(guān)鍵設(shè)計(jì)是定義清晰的數(shù)據(jù)槽Slot和填充規(guī)則。父任務(wù)的結(jié)果如何映射到子任務(wù)的輸入?yún)?shù)需要明確的約定。3.2 一個(gè)簡(jiǎn)化的實(shí)現(xiàn)流程讓我們用偽代碼勾勒一下從用戶(hù)輸入到最終輸出的核心循環(huán)class HierarchicalAgent: def __init__(self, llm_client, skill_lib): self.llm llm_client self.skills skill_lib self.context {} def run(self, user_input: str): # 步驟1規(guī)劃 - 生成任務(wù)樹(shù) task_tree self._plan(user_input) # 步驟2執(zhí)行 - 工作流引擎調(diào)度 final_result self._execute_task(task_tree.root) return final_result def _plan(self, goal: str) - TaskTree: # 構(gòu)建Prompt讓LLM基于目標(biāo)和個(gè)人技能進(jìn)行分解 prompt f 你是一個(gè)任務(wù)規(guī)劃專(zhuān)家。你的技能庫(kù)如下 {self._format_skills_for_prompt()} 請(qǐng)將用戶(hù)目標(biāo)分解為層次化的任務(wù)樹(shù)。目標(biāo){goal} 輸出格式必須是嚴(yán)格的JSON包含id, name, description, skills_needed所需的技能名列表 sub_tasks子任務(wù)列表等字段。 llm_response self.llm.generate(prompt) task_tree_json parse_json(llm_response) return TaskTree.from_json(task_tree_json) def _execute_task(self, task_node: TaskNode): # 如果這是一個(gè)葉子任務(wù)有具體技能需要執(zhí)行 if task_node.skills_needed: results [] for skill_name in task_node.skills_needed: # 從上下文和任務(wù)描述中提取該技能所需的參數(shù) skill_params self._extract_params_for_skill(skill_name, task_node, self.context) # 調(diào)用技能執(zhí)行器 result self.skill_executor.execute(skill_name, skill_params) results.append(result) # 將結(jié)果更新到上下文中供后續(xù)任務(wù)使用 self._update_context(skill_name, result) # 合并葉子任務(wù)的結(jié)果 return self._merge_results(results, task_node) else: # 如果這是一個(gè)復(fù)合任務(wù)有子任務(wù)則遞歸執(zhí)行子任務(wù)并整合結(jié)果 child_results [] for child_task in task_node.sub_tasks: # 這里可以加入依賴(lài)判斷和并行執(zhí)行優(yōu)化 result self._execute_task(child_task) child_results.append(result) return self._merge_child_results(child_results, task_node)3.3 關(guān)鍵技術(shù)選型與考量大模型選擇任務(wù)規(guī)劃器的質(zhì)量直接取決于LLM的推理和遵循指令能力。GPT-4、Claude 3等頂級(jí)閉源模型效果最好但成本高。開(kāi)源模型如Qwen、DeepSeek、Llama 3的指令微調(diào)版本也是不錯(cuò)的選擇但需要在Prompt工程上投入更多精力確保其輸出結(jié)構(gòu)化。技能描述如何向LLM清晰描述你的技能庫(kù)至關(guān)重要。描述應(yīng)包括1) 功能簡(jiǎn)述2) 必需的輸入?yún)?shù)及其類(lèi)型和含義3) 輸出格式。清晰的描述能極大提升規(guī)劃器匹配技能的準(zhǔn)確率。錯(cuò)誤處理與重試在分層架構(gòu)中錯(cuò)誤處理也需要分層。葉子技能執(zhí)行失敗如API超時(shí)可以在該層級(jí)設(shè)置重試機(jī)制或啟用備用技能。如果整個(gè)子任務(wù)失敗工作流引擎應(yīng)能向上匯報(bào)由父任務(wù)決定是重試整個(gè)子任務(wù)、選擇替代方案還是整體失敗。上下文管理策略簡(jiǎn)單的鍵值對(duì)存儲(chǔ)可能不夠。需要考慮上下文的作用域全局、任務(wù)鏈、會(huì)話(huà)、生命周期以及沖突解決當(dāng)多個(gè)任務(wù)試圖寫(xiě)入同一個(gè)上下文鍵時(shí)。一種常見(jiàn)模式是使用“黑板”模式所有任務(wù)都可以讀寫(xiě)但需要版本或來(lái)源標(biāo)記。注意事項(xiàng)Prompt工程是成敗關(guān)鍵讓LLM穩(wěn)定輸出結(jié)構(gòu)化的任務(wù)樹(shù)是整個(gè)系統(tǒng)的基石。這需要極其精細(xì)的Prompt設(shè)計(jì)。除了提供清晰的技能描述和輸出格式要求外最好在Prompt中加入幾個(gè)高質(zhì)量的示例Few-shot Learning。例如給出一個(gè)“訂餐”目標(biāo)及其對(duì)應(yīng)的標(biāo)準(zhǔn)任務(wù)樹(shù)JSON。這能顯著減少LLM的“胡思亂想”提高輸出結(jié)構(gòu)的穩(wěn)定性和準(zhǔn)確性。同時(shí)必須在代碼中對(duì)LLM的輸出進(jìn)行嚴(yán)格的校驗(yàn)和解析對(duì)不符合格式的響應(yīng)要有降級(jí)或重試策略。4. 實(shí)戰(zhàn)演練從零搭建一個(gè)旅行規(guī)劃助手光說(shuō)不練假把式。我們以“周末短途旅行規(guī)劃”為例手把手走一遍構(gòu)建核心組件的流程。假設(shè)我們已經(jīng)有了一個(gè)能調(diào)用外部API的大模型客戶(hù)端比如OpenAI或國(guó)內(nèi)平臺(tái)的SDK。4.1 第一步定義原子技能庫(kù)首先我們需要定義智能體能用的“積木塊”。這里我們創(chuàng)建幾個(gè)最基礎(chǔ)的技能。# skill_library.py class SkillLibrary: def __init__(self): self.skills {} self._register_skills() def _register_skills(self): # 技能1確認(rèn)地點(diǎn)信息從模糊描述中提取具體城市 self.skills[clarify_location] { function: self._clarify_location, description: 從用戶(hù)模糊的地點(diǎn)描述中提取出明確的出發(fā)城市和目的城市。例如‘從北京去上?!?- (‘北京’ ‘上?!?, input_schema: {description: str}, output_schema: {departure: str, destination: str} } # 技能2查詢(xún)航班信息模擬 self.skills[search_flights] { function: self._search_flights, description: 根據(jù)出發(fā)地、目的地、日期查詢(xún)航班信息和價(jià)格。, input_schema: {departure: str, destination: str, date: str}, output_schema: {flights: list} # 列表里是航班詳情字典 } # 技能3查詢(xún)酒店信息模擬 self.skills[search_hotels] { function: self._search_hotels, description: 根據(jù)城市、入住日期、離店日期和價(jià)格范圍查詢(xún)酒店信息。, input_schema: {city: str, check_in: str, check_out: str, max_price: int}, output_schema: {hotels: list} } # 技能4生成行程摘要 self.skills[generate_itinerary] { function: self._generate_itinerary, description: 將交通、住宿、景點(diǎn)等零散信息整合成一份格式優(yōu)美的Markdown行程單。, input_schema: {transport: dict, accommodation: dict, activities: list}, output_schema: {itinerary_md: str} } # 下面是具體的技能函數(shù)實(shí)現(xiàn)模擬 def _clarify_location(self, description): # 這里應(yīng)該調(diào)用一個(gè)NER模型或LLM來(lái)解析。為簡(jiǎn)化我們模擬一下。 if 北京 in description and 上海 in description: return {departure: 北京, destination: 上海} # ... 其他邏輯 return {departure: 未知, destination: 未知} def _search_flights(self, departure, destination, date): # 模擬API調(diào)用返回假數(shù)據(jù) return { flights: [ {airline: 航司A, dep_time: 08:00, arr_time: 10:00, price: 1200}, {airline: 航司B, dep_time: 14:00, arr_time: 16:00, price: 900}, ] } # ... 其他技能函數(shù)類(lèi)似4.2 第二步實(shí)現(xiàn)任務(wù)規(guī)劃器規(guī)劃器的核心是構(gòu)造一個(gè)能讓LLM“乖乖聽(tīng)話(huà)”的Prompt。# planner.py class TaskPlanner: def __init__(self, llm_client, skill_lib): self.llm llm_client self.skill_lib skill_lib def plan(self, goal): skills_prompt self._build_skills_prompt() system_prompt f你是一個(gè)高級(jí)任務(wù)規(guī)劃AI。請(qǐng)將用戶(hù)目標(biāo)分解為可執(zhí)行的任務(wù)層次結(jié)構(gòu)。 你可以使用的技能如下 {skills_prompt} 請(qǐng)以JSON格式輸出任務(wù)樹(shù)。每個(gè)任務(wù)節(jié)點(diǎn)必須包含以下字段 - id: 唯一標(biāo)識(shí)符。 - name: 任務(wù)名稱(chēng)。 - description: 任務(wù)詳細(xì)描述。 - skills_needed: 執(zhí)行此任務(wù)所需的具體技能名稱(chēng)列表必須是上面提供的技能名。如果是需要進(jìn)一步分解的復(fù)合任務(wù)此列表為空。 - sub_tasks: 子任務(wù)列表格式同父任務(wù)。如果沒(méi)有子任務(wù)則為空列表。 輸出示例 {{ id: root, name: 規(guī)劃周末旅行, description: 總體協(xié)調(diào)旅行規(guī)劃, skills_needed: [], sub_tasks: [ {{ id: t1, name: 確認(rèn)旅行細(xì)節(jié), description: 明確出發(fā)地、目的地、時(shí)間和預(yù)算, skills_needed: [clarify_location, extract_budget_and_date], sub_tasks: [] }} // ... 其他子任務(wù) ] }} user_prompt f用戶(hù)目標(biāo){goal} # 調(diào)用LLM response self.llm.chat_completion( systemsystem_prompt, messages[{role: user, content: user_prompt}], temperature0.1 # 低溫度保證輸出穩(wěn)定 ) # 解析并返回任務(wù)樹(shù)對(duì)象 return self._parse_response(response)4.3 第三步構(gòu)建工作流引擎與執(zhí)行循環(huán)這是最復(fù)雜的部分我們需要一個(gè)能遍歷樹(shù)、管理狀態(tài)和上下文的引擎。# workflow_engine.py class WorkflowEngine: def __init__(self, skill_executor): self.skill_executor skill_executor self.context {} # 全局上下文黑板 def execute_tree(self, task_tree): return self._execute_node(task_tree.root) def _execute_node(self, node): print(f[執(zhí)行] 開(kāi)始任務(wù): {node.name}) # 情況1葉子任務(wù)有具體技能要執(zhí)行 if node.skills_needed: results {} for skill_name in node.skills_needed: # 從節(jié)點(diǎn)描述和全局上下文中提取該技能需要的參數(shù) # 這里需要一個(gè)參數(shù)提取器可能也需要LLM輔助或基于規(guī)則 params self._extract_parameters(skill_name, node, self.context) print(f 調(diào)用技能: {skill_name}, 參數(shù): {params}) try: skill_result self.skill_executor.execute(skill_name, params) # 將技能結(jié)果按預(yù)定規(guī)則存入上下文 self._update_context_with_result(skill_name, skill_result) results[skill_name] skill_result except Exception as e: print(f 技能 {skill_name} 執(zhí)行失敗: {e}) # 這里可以實(shí)現(xiàn)失敗重試、備用技能等邏輯 return {status: failed, error: str(e)} # 所有技能執(zhí)行成功合并結(jié)果返回給父任務(wù) merged_result self._merge_skill_results(results, node) node.result merged_result return merged_result # 情況2復(fù)合任務(wù)需要執(zhí)行子任務(wù) else: child_results [] # 這里可以?xún)?yōu)化分析子任務(wù)間依賴(lài)決定串行還是并行執(zhí)行 for child in node.sub_tasks: child_result self._execute_node(child) if child_result.get(status) failed: # 子任務(wù)失敗可以在這里決定是否繼續(xù)或向上傳遞失敗 pass child_results.append(child_result) # 所有子任務(wù)完成整合結(jié)果 final_result self._synthesize_results(child_results, node) node.result final_result return final_result4.4 第四步組裝并運(yùn)行智能體最后我們把所有模塊像拼圖一樣組裝起來(lái)。# main.py from llm_client import get_llm_client from skill_library import SkillLibrary from planner import TaskPlanner from workflow_engine import WorkflowEngine from skill_executor import SkillExecutor def main(): # 1. 初始化所有組件 llm get_llm_client(api_keyyour_key) # 你的LLM客戶(hù)端 skill_lib SkillLibrary() planner TaskPlanner(llm, skill_lib) executor SkillExecutor(skill_lib) # 技能執(zhí)行器 engine WorkflowEngine(executor) # 2. 創(chuàng)建智能體 agent HierarchicalAgent(planner, engine) # 3. 運(yùn)行 user_goal 幫我規(guī)劃一個(gè)從北京到上海的周末旅行總預(yù)算不超過(guò)2000元包含交通和住宿最后給我一份行程單。 print(f用戶(hù)目標(biāo): {user_goal}) print(*50) final_itinerary agent.run(user_goal) print(*50) print(【最終行程單】) print(final_itinerary) if __name__ __main__: main()運(yùn)行這個(gè)程序你會(huì)看到智能體一步步地規(guī)劃、確認(rèn)地點(diǎn)、查詢(xún)航班酒店、進(jìn)行預(yù)算分配最終生成一份整合好的行程單。雖然這里的技能都是模擬的但整個(gè)分層決策和執(zhí)行的骨架已經(jīng)清晰可見(jiàn)。實(shí)操心得從模擬到真實(shí)的挑戰(zhàn)在原型階段用模擬函數(shù)快速驗(yàn)證流程是完全可行的。但當(dāng)你接入真實(shí)API時(shí)會(huì)立刻遇到幾個(gè)挑戰(zhàn)1)API穩(wěn)定性與錯(cuò)誤處理真實(shí)網(wǎng)絡(luò)請(qǐng)求會(huì)超時(shí)、會(huì)返回異常格式。必須在技能執(zhí)行層做好重試和異常捕獲。2)參數(shù)提取的復(fù)雜性_extract_parameters函數(shù)在實(shí)際中會(huì)非常復(fù)雜。一個(gè)任務(wù)的描述文本中如何準(zhǔn)確找到調(diào)用“search_hotels”技能所需的city、check_in、max_price等參數(shù)這本身可能就需要一個(gè)小的LLM調(diào)用或一套復(fù)雜的規(guī)則引擎。3)成本控制每一次任務(wù)分解、每一次參數(shù)提取可能都需要調(diào)用LLM累積起來(lái)成本不菲。需要對(duì)頻繁、固定的任務(wù)模式進(jìn)行緩存或固化比如將常見(jiàn)的任務(wù)樹(shù)模板存儲(chǔ)起來(lái)直接復(fù)用。5. 進(jìn)階探討架構(gòu)演進(jìn)與核心挑戰(zhàn)實(shí)現(xiàn)了一個(gè)基礎(chǔ)版本后我們會(huì)發(fā)現(xiàn)還有很多可以?xún)?yōu)化和深入的地方。分層組合性智能體的架構(gòu)設(shè)計(jì)本身就是一個(gè)充滿(mǎn)挑戰(zhàn)和趣味的領(lǐng)域。5.1 動(dòng)態(tài)技能發(fā)現(xiàn)與組合我們之前的技能庫(kù)是靜態(tài)注冊(cè)的。但在一個(gè)開(kāi)放環(huán)境中智能體可能需要使用它之前不知道的技能。這就引出了動(dòng)態(tài)技能發(fā)現(xiàn)。例如智能體可以訪問(wèn)一個(gè)技能市場(chǎng)或一個(gè)工具文檔庫(kù)。當(dāng)規(guī)劃器發(fā)現(xiàn)現(xiàn)有技能無(wú)法滿(mǎn)足某個(gè)子任務(wù)時(shí)它可以主動(dòng)查詢(xún)技能庫(kù)尋找描述匹配的新技能并將其動(dòng)態(tài)組合到當(dāng)前的任務(wù)計(jì)劃中。這要求技能必須有機(jī)器可讀的、標(biāo)準(zhǔn)化的描述例如遵循OpenAPI規(guī)范。5.2 分層強(qiáng)化學(xué)習(xí)與反思最初的規(guī)劃可能不是最優(yōu)的。智能體需要在執(zhí)行中學(xué)習(xí)。我們可以引入分層強(qiáng)化學(xué)習(xí)的思想。每個(gè)任務(wù)節(jié)點(diǎn)尤其是高層決策節(jié)點(diǎn)都可以被賦予一個(gè)“價(jià)值”評(píng)估。當(dāng)整個(gè)任務(wù)成功完成并得到用戶(hù)正面反饋時(shí)成功路徑上的所有節(jié)點(diǎn)都會(huì)獲得“獎(jiǎng)勵(lì)”從而強(qiáng)化那些做出正確決策如選擇了正確的子任務(wù)分解方式、選擇了性?xún)r(jià)比高的航班的節(jié)點(diǎn)。反之失敗則會(huì)帶來(lái)“懲罰”。通過(guò)多次運(yùn)行智能體可以學(xué)會(huì)更好的分解和決策策略。另一個(gè)關(guān)鍵機(jī)制是反思。在執(zhí)行完一個(gè)任務(wù)尤其是失敗后智能體可以啟動(dòng)一個(gè)“反思”子任務(wù)讓LLM分析執(zhí)行日志找出問(wèn)題根源是規(guī)劃不合理技能選擇錯(cuò)誤參數(shù)不對(duì)并據(jù)此修正當(dāng)前的任務(wù)樹(shù)或未來(lái)的規(guī)劃策略實(shí)現(xiàn)自我改進(jìn)。5.3 與現(xiàn)有框架的融合你不必完全從零開(kāi)始。許多新興的Agent開(kāi)發(fā)框架已經(jīng)開(kāi)始支持分層或工作流的概念。例如LangChain其LangGraph庫(kù)非常適合構(gòu)建有狀態(tài)、帶循環(huán)的工作流你可以用它來(lái)實(shí)現(xiàn)任務(wù)樹(shù)的調(diào)度和執(zhí)行引擎。Microsoft Autogen支持定義多Agent協(xié)作你可以將不同的“子任務(wù)”分配給不同角色的Agent一個(gè)負(fù)責(zé)規(guī)劃一個(gè)負(fù)責(zé)查詢(xún)一個(gè)負(fù)責(zé)整合天然契合分層思想。CrewAI明確以“角色Agent 任務(wù)Task 流程Process”為核心其流程概念如順序執(zhí)行、分層執(zhí)行可以直接用來(lái)建模我們的任務(wù)樹(shù)。我的建議是在理解核心原理后可以基于這些成熟框架進(jìn)行構(gòu)建能省去大量底層調(diào)度和通信的麻煩。5.4 當(dāng)前面臨的核心挑戰(zhàn)規(guī)劃可靠性LLM生成的任務(wù)樹(shù)可能邏輯混亂、不完整或無(wú)法執(zhí)行。需要設(shè)計(jì)更魯棒的規(guī)劃-驗(yàn)證-重規(guī)劃循環(huán)。上下文管理爆炸隨著任務(wù)樹(shù)變深變寬需要在不同層級(jí)間傳遞的數(shù)據(jù)量巨大如何高效、準(zhǔn)確地進(jìn)行上下文傳遞和版本管理是一大難題。長(zhǎng)程依賴(lài)與回溯有時(shí)執(zhí)行深層的子任務(wù)時(shí)才發(fā)現(xiàn)高層的某個(gè)假設(shè)是錯(cuò)的比如預(yù)算根本不夠。這就需要智能體能夠“回溯”到上層甚至修改整個(gè)計(jì)劃。這要求工作流引擎具備更強(qiáng)的狀態(tài)管理和回滾能力。評(píng)估與調(diào)試如何評(píng)估一個(gè)分層智能體的好壞不像簡(jiǎn)單問(wèn)答有準(zhǔn)確率。需要設(shè)計(jì)一套針對(duì)任務(wù)完成度、步驟合理性、效率等多維度的評(píng)估體系。調(diào)試一個(gè)出錯(cuò)的智能體也像調(diào)試一個(gè)分布式系統(tǒng)非常復(fù)雜。6. 典型問(wèn)題排查與優(yōu)化技巧在實(shí)際開(kāi)發(fā)和運(yùn)行中你會(huì)遇到各種各樣的問(wèn)題。下面是我踩過(guò)的一些坑和總結(jié)的應(yīng)對(duì)方法。6.1 常見(jiàn)問(wèn)題速查表問(wèn)題現(xiàn)象可能原因排查步驟與解決方案LLM規(guī)劃器輸出格式錯(cuò)誤Prompt指令不清晰LLM未遵循格式輸出被截?cái)唷?. 在Prompt中加入更嚴(yán)格的格式描述和示例。2. 降低生成溫度temperature。3. 在代碼中添加輸出格式校驗(yàn)和自動(dòng)修復(fù)邏輯如嘗試用正則表達(dá)式提取JSON。4. 對(duì)于復(fù)雜規(guī)劃采用“分步規(guī)劃”策略先讓LLM輸出大綱再細(xì)化每一層。技能執(zhí)行失敗參數(shù)錯(cuò)誤參數(shù)提取邏輯有誤上下文數(shù)據(jù)未正確傳遞技能描述與規(guī)劃器理解不匹配。1. 打印出技能調(diào)用前的完整上下文和提取的參數(shù)進(jìn)行比對(duì)。2. 強(qiáng)化參數(shù)提取器可考慮用一個(gè)小型LLM專(zhuān)門(mén)做參數(shù)映射。3. 確保技能描述清晰無(wú)歧義輸入輸出格式定義精確。任務(wù)陷入無(wú)限循環(huán)或重復(fù)執(zhí)行任務(wù)樹(shù)中存在循環(huán)依賴(lài)工作流引擎的狀態(tài)管理有BugLLM生成了遞歸性質(zhì)的任務(wù)分解。1. 為任務(wù)樹(shù)中的每個(gè)節(jié)點(diǎn)增加執(zhí)行狀態(tài)未開(kāi)始、執(zhí)行中、完成、失敗和訪問(wèn)記錄。2. 在執(zhí)行前檢查依賴(lài)環(huán)。3. 設(shè)置最大執(zhí)行深度或超時(shí)時(shí)間防止死循環(huán)。整體執(zhí)行時(shí)間過(guò)長(zhǎng)所有任務(wù)串行執(zhí)行部分技能如網(wǎng)絡(luò)請(qǐng)求耗時(shí)久LLM調(diào)用延遲高。1. 分析任務(wù)樹(shù)將沒(méi)有依賴(lài)關(guān)系的子任務(wù)改為并行執(zhí)行。2. 對(duì)耗時(shí)的技能調(diào)用設(shè)置合理的超時(shí)和異步機(jī)制。3. 對(duì)LLM的規(guī)劃結(jié)果進(jìn)行緩存對(duì)于相似的用戶(hù)目標(biāo)直接復(fù)用之前的任務(wù)樹(shù)。結(jié)果質(zhì)量差不符合用戶(hù)意圖任務(wù)分解偏離主題技能組合邏輯不合理最終結(jié)果合成方式不佳。1. 在規(guī)劃階段讓LLM同時(shí)評(píng)估任務(wù)樹(shù)的“可行性”和“與目標(biāo)的相關(guān)性”。2. 在最終結(jié)果合成前增加一個(gè)“質(zhì)量檢查”步驟用LLM判斷結(jié)果是否滿(mǎn)足用戶(hù)核心訴求。3. 引入用戶(hù)反饋機(jī)制在關(guān)鍵步驟請(qǐng)求用戶(hù)確認(rèn)如“我找到了A和B兩個(gè)航班您更看重時(shí)間還是價(jià)格”。6.2 性能與成本優(yōu)化技巧規(guī)劃結(jié)果緩存對(duì)于常見(jiàn)、通用的用戶(hù)目標(biāo)如“查天氣”、“寫(xiě)周報(bào)”其任務(wù)樹(shù)結(jié)構(gòu)往往是相似的。可以將(用戶(hù)目標(biāo), 技能庫(kù)版本)作為鍵將規(guī)劃好的任務(wù)樹(shù)緩存起來(lái)。下次遇到類(lèi)似請(qǐng)求直接使用緩存省去一次LLM調(diào)用。技能調(diào)用批處理如果多個(gè)葉子任務(wù)需要調(diào)用同一個(gè)外部API比如都需要查詢(xún)不同城市的天氣可以將這些請(qǐng)求合并成一個(gè)批量請(qǐng)求減少網(wǎng)絡(luò)開(kāi)銷(xiāo)。輕量級(jí)模型分工不是所有步驟都需要最強(qiáng)的GPT-4??梢杂眯∧P腿缧⌒烷_(kāi)源模型處理簡(jiǎn)單的參數(shù)提取、格式轉(zhuǎn)換、結(jié)果初步過(guò)濾等任務(wù)用大模型專(zhuān)注于核心的復(fù)雜規(guī)劃和決策。這種“大小模型混用”的策略能有效控制成本。異步與非阻塞執(zhí)行工作流引擎應(yīng)采用異步設(shè)計(jì)。當(dāng)某個(gè)技能在等待網(wǎng)絡(luò)響應(yīng)時(shí)引擎可以去調(diào)度其他獨(dú)立的子任務(wù)充分利用等待時(shí)間。6.3 提升智能體“智能”感的技巧漸進(jìn)式呈現(xiàn)不要讓用戶(hù)長(zhǎng)時(shí)間等待最終結(jié)果。智能體可以在執(zhí)行過(guò)程中階段性輸出一些中間成果。例如“已為您找到3個(gè)符合預(yù)算的航班選項(xiàng)正在比對(duì)時(shí)間...”、“酒店已篩選完畢接下來(lái)為您規(guī)劃景點(diǎn)路線”。這能極大提升用戶(hù)體驗(yàn)。提供選擇與解釋在做出關(guān)鍵決策如選擇航班時(shí)智能體可以將其推理過(guò)程簡(jiǎn)要呈現(xiàn)給用戶(hù)“選擇了航班A因?yàn)樗鼤r(shí)間最優(yōu)且價(jià)格在預(yù)算內(nèi)”甚至提供幾個(gè)選項(xiàng)讓用戶(hù)定奪。這增加了透明度和可控性。記憶與個(gè)性化為智能體引入長(zhǎng)期記憶。記住用戶(hù)的歷史偏好如“用戶(hù)通常選擇靠過(guò)道的座位”、“喜歡中式早餐”并在后續(xù)規(guī)劃中主動(dòng)應(yīng)用這些偏好會(huì)讓它感覺(jué)更貼心、更智能。構(gòu)建一個(gè)成熟可用的分層組合性智能體是一個(gè)系統(tǒng)工程涉及Prompt工程、軟件架構(gòu)、算法設(shè)計(jì)等多個(gè)方面。它沒(méi)有銀彈需要根據(jù)具體的應(yīng)用場(chǎng)景進(jìn)行大量的迭代和調(diào)優(yōu)。但毫無(wú)疑問(wèn)這是通向更強(qiáng)大、更可靠AI助手的一條必經(jīng)之路。從今天開(kāi)始嘗試為你自己的智能體引入“分層”和“組合”的思想你會(huì)發(fā)現(xiàn)它的能力邊界將被大大拓展。