化：從串行瓶頸到3倍性能提升實(shí)戰(zhàn))
1. 項(xiàng)目緣起從單線程Agent到并行工具鏈的必然演進(jìn)最近在重構(gòu)一個(gè)智能客服的Agent項(xiàng)目遇到了一個(gè)典型的性能瓶頸。這個(gè)Agent需要同時(shí)調(diào)用多個(gè)外部API來回答用戶的一個(gè)復(fù)雜問題比如用戶問“幫我查一下明天上海的天氣并推薦一個(gè)適合這種天氣的戶外活動(dòng)再找一家附近評(píng)分高的餐廳”。在最初的版本里我讓Agent按順序執(zhí)行先調(diào)用天氣API等結(jié)果返回后再調(diào)用活動(dòng)推薦API最后再調(diào)用餐廳搜索API。邏輯上沒問題但實(shí)測下來一個(gè)請(qǐng)求的平均響應(yīng)時(shí)間竟然超過了8秒。用戶等8秒才看到第一個(gè)字這體驗(yàn)直接不及格。問題的核心就在于工具Tools的調(diào)用是串行的。在LangChain的默認(rèn)Agent執(zhí)行器AgentExecutor里它拿到一個(gè)需要調(diào)用多個(gè)工具的計(jì)劃Plan后會(huì)像一個(gè)嚴(yán)格的管家不緊不慢地、一個(gè)接一個(gè)地執(zhí)行。工具A不返回結(jié)果絕不開始工具B。這在很多需要信息聚合的場景下造成了大量不必要的等待時(shí)間。這讓我開始思考Agent的“思考”LLM推理可以是單線程的但“執(zhí)行”工具調(diào)用為什么不能是并發(fā)的呢尤其是在工具之間沒有強(qiáng)依賴關(guān)系的時(shí)候。于是“并行優(yōu)化”就成了這個(gè)階段必須啃下的硬骨頭。我調(diào)研了社區(qū)方案發(fā)現(xiàn)大家普遍會(huì)提到Promise.all這個(gè)思路——沒錯(cuò)就是前端開發(fā)里用來處理多個(gè)異步任務(wù)的那個(gè)Promise.all。它的理念完美契合這里的需求同時(shí)發(fā)起多個(gè)獨(dú)立的網(wǎng)絡(luò)I/O請(qǐng)求然后等待所有結(jié)果返回最后再統(tǒng)一處理。這不正是解決我串行調(diào)用痛點(diǎn)的良藥嗎但將Promise.all的思想移植到LangChain的Agent開發(fā)中遠(yuǎn)不是加幾行async/await那么簡單。它涉及到對(duì)LangChain工具調(diào)用機(jī)制的理解、執(zhí)行流程的重構(gòu)、以及錯(cuò)誤處理與降級(jí)策略的設(shè)計(jì)。接下來我就結(jié)合這次實(shí)戰(zhàn)拆解如何構(gòu)建一個(gè)支持并行工具鏈的Agent并分享其中幾個(gè)關(guān)鍵的“踩坑”與“填坑”過程。2. 理解LangChain Agent的執(zhí)行機(jī)制與工具鏈瓶頸在動(dòng)手優(yōu)化之前我們必須先弄清楚LangChain的Agent到底是怎么工作的。很多人把LangChain當(dāng)作一個(gè)“黑盒”只知道它能連接大模型和工具但如果不理解其內(nèi)部執(zhí)行流優(yōu)化就無從談起。2.1 Agent的核心循環(huán)計(jì)劃、執(zhí)行、觀察一個(gè)典型的LangChain Agent以ReAct模式為例工作流程可以簡化為一個(gè)循環(huán)計(jì)劃PlanLLM根據(jù)用戶輸入和當(dāng)前上下文包括歷史對(duì)話和上一步的工具觀察結(jié)果決定下一步該做什么。輸出通常是一個(gè)結(jié)構(gòu)化的動(dòng)作Action包含要調(diào)用的工具名tool和輸入?yún)?shù)tool_input。執(zhí)行ExecuteAgent執(zhí)行器AgentExecutor解析這個(gè)動(dòng)作找到對(duì)應(yīng)的工具函數(shù)傳入?yún)?shù)并執(zhí)行它。這通常是一個(gè)異步的I/O操作比如調(diào)用一個(gè)HTTP API、查詢數(shù)據(jù)庫或執(zhí)行一段計(jì)算。觀察Observe工具執(zhí)行完成后將結(jié)果Observation返回給Agent執(zhí)行器。循環(huán)判斷執(zhí)行器將工具觀察結(jié)果連同歷史信息再次喂給LLM讓LLM判斷是否已經(jīng)獲得了足夠信息來生成最終答案Final Answer還是需要繼續(xù)調(diào)用下一個(gè)工具。這個(gè)循環(huán)會(huì)一直持續(xù)直到LLM決定輸出最終答案。瓶頸就出現(xiàn)在第2步“執(zhí)行”階段。在默認(rèn)實(shí)現(xiàn)中AgentExecutor是順序處理每一個(gè)“動(dòng)作”的。即使LLM在一次推理中規(guī)劃了多個(gè)可以并行執(zhí)行的動(dòng)作例如“同時(shí)查詢天氣和新聞”執(zhí)行器也會(huì)把它們當(dāng)作一個(gè)動(dòng)作序列來處理先執(zhí)行A等A的結(jié)果返回后再讓LLM基于A的結(jié)果思考決定是否執(zhí)行B。這并沒有充分利用多個(gè)工具之間可能存在的獨(dú)立性。2.2 工具Tool的本質(zhì)與封裝在LangChain中一個(gè)Tool本質(zhì)上是一個(gè)可調(diào)用的函數(shù)它有著明確的名稱name、描述description和參數(shù)模式args_schema。當(dāng)LLM決定使用某個(gè)工具時(shí)它是在根據(jù)工具的描述來匹配用戶意圖。from langchain.tools import tool tool def get_weather(city: str) - str: 根據(jù)城市名查詢天氣情況。 # 模擬一個(gè)耗時(shí)的網(wǎng)絡(luò)請(qǐng)求 import asyncio await asyncio.sleep(2) # 假設(shè)請(qǐng)求耗時(shí)2秒 return f{city}的天氣是晴天25攝氏度。 # 工具會(huì)被封裝成具有標(biāo)準(zhǔn)化接口的對(duì)象供Agent調(diào)用。當(dāng)我們談?wù)摗肮ぞ哝湣睍r(shí)通常指的是Agent在一次任務(wù)中可能按順序或按條件調(diào)用的多個(gè)工具的集合。而“并行優(yōu)化”目標(biāo)就是打破這個(gè)鏈?zhǔn)街胁槐匾捻樞蚣s束讓其中可以并發(fā)的部分同時(shí)飛起來。2.3 默認(rèn)執(zhí)行器的性能分析為了量化瓶頸我寫了一個(gè)簡單的測試。創(chuàng)建三個(gè)工具每個(gè)工具模擬一個(gè)耗時(shí)2秒的I/O操作。讓Agent執(zhí)行一個(gè)需要調(diào)用這三個(gè)工具的任務(wù)。import asyncio import time async def run_sequential_agent(): # 假設(shè)這是順序執(zhí)行的Agent start time.time() result_a await tool_a() result_b await tool_b() result_c await tool_c() end time.time() print(f順序執(zhí)行總耗時(shí): {end - start:.2f}秒) # 輸出約 6.00秒 async def hypothetical_parallel_agent(): # 這是我們期望的并行執(zhí)行 start time.time() # 同時(shí)發(fā)起所有任務(wù) task_a asyncio.create_task(tool_a()) task_b asyncio.create_task(tool_b()) task_c asyncio.create_task(tool_c()) # 等待所有任務(wù)完成 results await asyncio.gather(task_a, task_b, task_c) end time.time() print(f并行執(zhí)行總耗時(shí): {end - start:.2f}秒) # 輸出約 2.01秒測試結(jié)果一目了然順序執(zhí)行耗時(shí)是各工具耗時(shí)的線性累加~6秒而理想中的并行執(zhí)行耗時(shí)約等于最慢的那個(gè)工具的耗時(shí)~2秒。對(duì)于用戶體驗(yàn)來說這3倍的差距就是“卡頓”和“流暢”的天壤之別。因此改造勢在必行。3. 設(shè)計(jì)并行化工具鏈的核心架構(gòu)并行化不是簡單地把所有工具調(diào)用都扔進(jìn)asyncio.gather。一個(gè)健壯的并行Agent架構(gòu)需要考慮任務(wù)規(guī)劃、依賴關(guān)系、執(zhí)行調(diào)度和結(jié)果整合。我的設(shè)計(jì)思路主要分為以下幾個(gè)層次。3.1 識(shí)別可并行任務(wù)LLM的“并行思維”提示首先我們需要讓LLM具備“并行規(guī)劃”的能力。在默認(rèn)的ReAct提示詞中LLM習(xí)慣于思考“下一步”做什么。我們需要引導(dǎo)它去思考“哪些步驟可以同時(shí)做”。這主要通過修改系統(tǒng)的提示詞System Prompt來實(shí)現(xiàn)。在給Agent的指令中需要明確加入并行化的引導(dǎo)“你是一個(gè)高效的任務(wù)規(guī)劃者。在規(guī)劃步驟時(shí)請(qǐng)仔細(xì)分析任務(wù)。如果多個(gè)子任務(wù)之間沒有依賴關(guān)系即任務(wù)B不需要任務(wù)A的結(jié)果作為輸入那么你應(yīng)該將這些子任務(wù)規(guī)劃為可以同時(shí)執(zhí)行。在你的輸出中對(duì)于可以并行的任務(wù)請(qǐng)用特定的格式例如PARALLEL: [動(dòng)作1, 動(dòng)作2, ...]來標(biāo)識(shí)?！崩鐚?duì)于查詢天氣和新聞的任務(wù)LLM應(yīng)該輸出思考用戶需要天氣和新聞這兩者沒有依賴關(guān)系。 行動(dòng)PARALLEL: [{tool: get_weather, tool_input: {city: 上海}}, {tool: get_news, tool_input: {topic: 科技}}]這樣我們后續(xù)的并行執(zhí)行器就可以解析這個(gè)PARALLEL標(biāo)記知道里面的動(dòng)作是可以并發(fā)執(zhí)行的。這一步是關(guān)鍵的前提它決定了并行化的潛力和正確性。3.2 構(gòu)建并行執(zhí)行器ParallelExecutor這是架構(gòu)的核心。我們需要?jiǎng)?chuàng)建一個(gè)新的執(zhí)行器繼承或重寫LangChain的AgentExecutor主要覆蓋_call或_atake_next_step方法中關(guān)于工具執(zhí)行的部分。其核心邏輯如下解析動(dòng)作檢查LLM輸出的動(dòng)作。如果是單個(gè)動(dòng)作按原流程處理。如果檢測到PARALLEL標(biāo)記則提取出里面的動(dòng)作列表。依賴檢查可選但建議對(duì)于標(biāo)記為并行的動(dòng)作列表執(zhí)行器可以做一個(gè)簡單的靜態(tài)檢查確保這些工具之間沒有直接的輸入輸出依賴這需要預(yù)先定義工具的資源依賴關(guān)系初期可以靠LLM和提示詞保證后期可以加入更嚴(yán)格的檢查。并發(fā)執(zhí)行使用asyncio.gather或anyio庫如果使用LangChain的異步生態(tài)并發(fā)執(zhí)行這些工具。為每個(gè)工具調(diào)用創(chuàng)建一個(gè)異步任務(wù)。結(jié)果收集與聚合等待所有并發(fā)任務(wù)完成。收集每個(gè)任務(wù)的結(jié)果或可能發(fā)生的異常。結(jié)果格式化與傳遞將并行執(zhí)行的結(jié)果整合成一個(gè)格式化的觀察字符串傳遞給LLM進(jìn)行下一步的“思考”。例如可以將結(jié)果合并為“天氣結(jié)果xxx新聞結(jié)果yyy”。import asyncio from typing import List, Dict, Any from langchain.agents import AgentExecutor from langchain.tools import BaseTool class ParallelAgentExecutor(AgentExecutor): 支持并行工具調(diào)用的Agent執(zhí)行器。 async def _execute_parallel_actions(self, actions: List[Dict]) - str: 并發(fā)執(zhí)行一組動(dòng)作。 async def run_tool(tool: BaseTool, tool_input: Dict): try: result await tool.arun(tool_input) # 使用工具的異步接口 return {tool: tool.name, status: success, result: result} except Exception as e: return {tool: tool.name, status: error, result: str(e)} # 創(chuàng)建所有工具任務(wù)的協(xié)程 tasks [] for action in actions: tool self.tools_by_name[action[tool]] tasks.append(run_tool(tool, action[tool_input])) # 并發(fā)執(zhí)行并等待所有結(jié)果 tool_results await asyncio.gather(*tasks, return_exceptionsFalse) # 格式化結(jié)果供LLM觀察 observations [] for res in tool_results: if res[status] success: observations.append(f{res[tool]}的結(jié)果是{res[result]}) else: observations.append(f調(diào)用{res[tool]}時(shí)出錯(cuò){res[result]}) return \n.join(observations) async def _atake_next_step(self, ...): # ... 原有的邏輯解析出next_action ... # 判斷是否為并行動(dòng)作 if isinstance(next_action, dict) and next_action.get(type) parallel: parallel_actions next_action[actions] observation await self._execute_parallel_actions(parallel_actions) # 將并行執(zhí)行的結(jié)果作為單次觀察返回 return AgentFinish(return_values{output: observation}, log...) else: # 走原有的單動(dòng)作執(zhí)行邏輯 return await super()._atake_next_step(...)3.3 處理依賴與錯(cuò)誤并行模式下的新挑戰(zhàn)并行化引入了新的復(fù)雜性主要集中在兩點(diǎn)依賴關(guān)系這是并行正確性的生命線。如果兩個(gè)本應(yīng)有依賴關(guān)系的任務(wù)被錯(cuò)誤地并行執(zhí)行就會(huì)得到錯(cuò)誤的結(jié)果。例如“計(jì)算公司總營收”這個(gè)任務(wù)依賴于“獲取各部門營收”這個(gè)子任務(wù)的結(jié)果。在提示詞工程階段我們必須通過大量示例Few-shot和清晰的指令訓(xùn)練LLM準(zhǔn)確識(shí)別依賴。此外可以在執(zhí)行器層面實(shí)現(xiàn)一個(gè)輕量級(jí)的依賴圖檢查如果工具注冊(cè)時(shí)聲明了其產(chǎn)出outputs和所需輸入inputs系統(tǒng)可以自動(dòng)判斷一組動(dòng)作是否可并行。錯(cuò)誤處理與降級(jí)在串行模型中一個(gè)工具失敗整個(gè)鏈就中斷了邏輯簡單。但在并行模型中情況更復(fù)雜部分失敗三個(gè)并行任務(wù)一個(gè)失敗兩個(gè)成功。我們是直接整體失敗還是繼續(xù)使用成功的結(jié)果通常為了更好的用戶體驗(yàn)我們選擇“部分降級(jí)”。執(zhí)行器需要捕獲每個(gè)任務(wù)的異常將成功的結(jié)果和友好的錯(cuò)誤信息如“暫時(shí)無法獲取新聞但已為您查詢到天氣”一起整合返回給LLM。LLM需要具備根據(jù)不完整信息繼續(xù)推理或給出提示的能力。超時(shí)控制并行執(zhí)行中最慢的任務(wù)決定整體耗時(shí)。我們需要為整個(gè)并行任務(wù)組設(shè)置一個(gè)總超時(shí)也要為每個(gè)獨(dú)立工具設(shè)置單獨(dú)的超時(shí)。防止一個(gè)“慢”工具拖死整個(gè)請(qǐng)求。可以使用asyncio.wait_for來實(shí)現(xiàn)。async def run_tool_with_timeout(tool, tool_input, timeout5): try: # 為單個(gè)工具調(diào)用設(shè)置超時(shí) result await asyncio.wait_for(tool.arun(tool_input), timeouttimeout) return {status: success, result: result} except asyncio.TimeoutError: return {status: error, result: f工具執(zhí)行超時(shí){timeout}秒} except Exception as e: return {status: error, result: f工具執(zhí)行錯(cuò)誤{e}}4. 實(shí)戰(zhàn)將串行Agent改造為并行Agent的完整步驟理論說完了我們來看一個(gè)具體的改造案例。假設(shè)我們有一個(gè)基于OpenAI Function Calling的Agent它原本串行調(diào)用get_weather、search_web和calculate三個(gè)工具。4.1 第一步定義支持并行規(guī)劃的提示詞這是最核心的一步。我們需要精心設(shè)計(jì)System Message和Few-shot Examples。from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder system_message 你是一個(gè)高級(jí)任務(wù)規(guī)劃與執(zhí)行Agent。你的目標(biāo)是高效、準(zhǔn)確地完成用戶請(qǐng)求。 **并行規(guī)劃規(guī)則** 1. 仔細(xì)分析用戶請(qǐng)求將其拆解為多個(gè)子任務(wù)。 2. 判斷子任務(wù)間的依賴關(guān)系。如果任務(wù)B**不需要**任務(wù)A的輸出作為輸入那么任務(wù)A和B就可以并行執(zhí)行。 3. 對(duì)于可以并行執(zhí)行的任務(wù)組請(qǐng)?jiān)谀愕妮敵鲋惺褂靡韵翵SON格式明確標(biāo)識(shí) json { type: parallel, actions: [ {tool: 工具A名稱, tool_input: {參數(shù): 值}}, {tool: 工具B名稱, tool_input: {參數(shù): 值}} ] }對(duì)于必須串行的任務(wù)或者單個(gè)任務(wù)請(qǐng)按正常格式輸出。示例1并行場景 用戶告訴我今天北京的天氣和最新的科技頭條新聞。 思考查詢天氣和搜索新聞是獨(dú)立任務(wù)無依賴。 輸出{ type: parallel, actions: [ {tool: get_weather, tool_input: {city: 北京}}, {tool: search_web, tool_input: {query: 科技 頭條 新聞}} ] }示例2串行場景 用戶先查一下上海的溫度如果高于30度就推薦一個(gè)室內(nèi)活動(dòng)。 思考推薦活動(dòng)依賴于天氣查詢的結(jié)果必須串行。 輸出 {tool: get_weather, tool_input: {city: 上海}} prompt ChatPromptTemplate.from_messages([ (system, system_message), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ])### 4.2 第二步創(chuàng)建自定義的Output Parser LangChain的Agent需要Output Parser來將LLM的文本輸出解析成結(jié)構(gòu)化的動(dòng)作。我們需要自定義一個(gè)解析器使其能識(shí)別我們定義的并行格式。 python from langchain.schema import AgentAction, AgentFinish import json import re class ParallelOutputParser: 解析LLM輸出識(shí)別并行動(dòng)作或單個(gè)動(dòng)作。 def parse(self, llm_output: str) - dict: # 首先嘗試匹配JSON格式的并行指令 json_match re.search(rjson\n(.*?)\n, llm_output, re.DOTALL) if json_match: try: data json.loads(json_match.group(1)) if data.get(type) parallel and actions in data: return {type: parallel, actions: data[actions]} except json.JSONDecodeError: pass # 如果不是并行格式嘗試按原有的單動(dòng)作格式解析這里簡化處理 # 假設(shè)原有解析邏輯能解析出 tool 和 tool_input # ... 原有的解析代碼 ... # 如果解析成功返回 {type: single, action: agent_action} # 如果解析為最終答案返回 AgentFinish # 兜底如果無法解析返回需要繼續(xù)思考或報(bào)錯(cuò) return {type: error, message: f無法解析LLM輸出: {llm_output[:100]}...}4.3 第三步實(shí)現(xiàn)自定義的并行Agent執(zhí)行器這里我們基于LangChain的AgentExecutor進(jìn)行擴(kuò)展。重點(diǎn)重寫_take_next_step方法或它的異步版本_atake_next_step。from langchain.agents import AgentExecutor from langchain.schema import AgentFinish import asyncio class CustomParallelExecutor(AgentExecutor): async def _execute_parallel_tools(self, actions: list) - str: 并發(fā)執(zhí)行工具并返回整合的觀察字符串。 tasks [] tool_name_to_func {tool.name: tool for tool in self.tools} for action_spec in actions: tool_name action_spec.get(tool) tool_input action_spec.get(tool_input, {}) tool tool_name_to_func.get(tool_name) if not tool: tasks.append(asyncio.sleep(0)) # 占位結(jié)果會(huì)是錯(cuò)誤 continue async def run_single(tool_obj, input_dict): try: # 使用異步調(diào)用并設(shè)置單個(gè)工具超時(shí) result await asyncio.wait_for(tool_obj.arun(input_dict), timeout10.0) return {tool: tool_obj.name, success: True, output: result} except asyncio.TimeoutError: return {tool: tool_obj.name, success: False, output: 請(qǐng)求超時(shí)} except Exception as e: return {tool: tool_obj.name, success: False, output: f錯(cuò)誤: {str(e)}} tasks.append(run_single(tool, tool_input)) # 并發(fā)執(zhí)行所有任務(wù) results await asyncio.gather(*tasks, return_exceptionsFalse) # 構(gòu)建觀察文本 observation_parts [] for res in results: if res[success]: observation_parts.append(f[{res[tool]}]: {res[output]}) else: observation_parts.append(f[{res[tool]} 失敗]: {res[output]}) return \n.join(observation_parts) async def _atake_next_step(self, name_to_tool_map, color_mapping, inputs, intermediate_steps): # 調(diào)用LLM獲取下一步?jīng)Q策 output await self.agent.aplan(intermediate_steps, **inputs) # 使用自定義解析器解析輸出 parsed self.output_parser.parse(output) if parsed[type] parallel: # 執(zhí)行并行工具組 observation await self._execute_parallel_tools(parsed[actions]) # 將并行執(zhí)行的結(jié)果作為一步觀察加入到歷史中 new_intermediate_steps intermediate_steps [(output, observation)] # 這里直接返回一個(gè)AgentFinish或者可以繼續(xù)循環(huán)。簡單起見我們讓LLM基于并行結(jié)果再思考一次。 # 更復(fù)雜的實(shí)現(xiàn)可以在這里不直接結(jié)束而是繼續(xù)循環(huán)。 return AgentFinish(return_values{output: f并行任務(wù)執(zhí)行完畢。結(jié)果如下\n{observation}}, logoutput) elif parsed[type] single: # 原有的單動(dòng)作執(zhí)行邏輯這里需要調(diào)用父類方法或類似邏輯 action parsed[action] tool name_to_tool_map[action.tool] observation await tool.arun(action.tool_input) return (action, observation) elif isinstance(parsed, AgentFinish): return parsed else: raise ValueError(f無法處理的解析結(jié)果: {parsed})4.4 第四步組裝并測試Agent將以上所有部件組裝起來創(chuàng)建一個(gè)完整的并行Agent。from langchain.chat_models import ChatOpenAI from langchain.agents import create_openai_functions_agent # 1. 定義工具 tools [get_weather_tool, search_web_tool, calculate_tool] # 2. 創(chuàng)建支持并行規(guī)劃的LLM鏈?zhǔn)褂米远x提示詞 llm ChatOpenAI(modelgpt-4, temperature0) agent create_openai_functions_agent(llm, tools, prompt) # 注意這里需要適配openai函數(shù)調(diào)用本身不支持我們自定義的格式。本例更適用于ReAct或自定義格式的Agent。 # 因此更常見的做法是使用initialize_agent并指定agent_type為zero-shot-react-description然后替換其executor和prompt。 # 為簡化示例我們假設(shè)已經(jīng)創(chuàng)建了一個(gè)能輸出我們所需格式的agent對(duì)象。 # 3. 創(chuàng)建自定義執(zhí)行器 agent_executor CustomParallelExecutor.from_agent_and_tools( agentagent, toolstools, max_iterations5, early_stopping_methodgenerate, ) # 4. 測試 async def test_agent(): result await agent_executor.arun({ input: 查詢今天上海和北京的天氣并搜索人工智能領(lǐng)域的最新突破。, chat_history: [] }) print(result) # 運(yùn)行測試 import asyncio asyncio.run(test_agent())5. 性能對(duì)比、踩坑記錄與進(jìn)階優(yōu)化完成基礎(chǔ)改造后進(jìn)行性能對(duì)比測試是必不可少的。同時(shí)在實(shí)際開發(fā)中我遇到了幾個(gè)典型的“坑”。5.1 性能對(duì)比數(shù)據(jù)我使用相同的三個(gè)模擬工具每個(gè)耗時(shí)2秒和相同的復(fù)雜查詢?nèi)蝿?wù)對(duì)比了改造前后的Agent。執(zhí)行模式平均響應(yīng)時(shí)間 (10次)資源占用 (峰值內(nèi)存)代碼復(fù)雜度原生串行Agent~6.05 秒較低低使用LangChain標(biāo)準(zhǔn)組件自定義并行Agent~2.10 秒略高并發(fā)協(xié)程高需自定義解析器、執(zhí)行器理想并行無開銷~2.00 秒--可以看到性能提升接近3倍與理論預(yù)期相符。微小的額外開銷主要來自并行任務(wù)調(diào)度和結(jié)果聚合的邏輯。對(duì)于I/O密集型任務(wù)網(wǎng)絡(luò)請(qǐng)求、數(shù)據(jù)庫查詢這種提升是決定性的。5.2 實(shí)戰(zhàn)中遇到的“坑”與解決方案坑1LLM的“并行規(guī)劃”不穩(wěn)定最初LLM并不總是按照我們期望的格式輸出。有時(shí)會(huì)忘記PARALLEL標(biāo)記有時(shí)會(huì)把有依賴的任務(wù)也并行。解決方案提示詞工程是關(guān)鍵。除了清晰的系統(tǒng)指令提供更多、更高質(zhì)量的Few-shot示例至關(guān)重要。示例需要覆蓋正例正確并行、負(fù)例錯(cuò)誤并行導(dǎo)致問題和邊界案例。此外可以在Output Parser中加入后處理邏輯如果LLM輸出了多個(gè)獨(dú)立動(dòng)作但沒有用并行標(biāo)記可以嘗試自動(dòng)將其“升級(jí)”為并行任務(wù)需謹(jǐn)慎確保依賴判斷準(zhǔn)確?？?工具副作用與資源競爭如果兩個(gè)并行工具都要寫入同一個(gè)文件或者修改同一個(gè)數(shù)據(jù)庫記錄就會(huì)引發(fā)競態(tài)條件。解決方案在工具設(shè)計(jì)層面要明確工具的“純度”和副作用。對(duì)于有副作用的工具需要在提示詞中明確告知LLM它們不能并行或者在執(zhí)行器層面通過“資源鎖”機(jī)制來序列化對(duì)特定資源的訪問。一個(gè)簡單的做法是為工具添加標(biāo)簽如{side_effect: high}執(zhí)行器在規(guī)劃時(shí)避開同時(shí)執(zhí)行高副作用工具?？?錯(cuò)誤處理的用戶體驗(yàn)最初并行任務(wù)中一個(gè)失敗我就讓整個(gè)Agent返回錯(cuò)誤。用戶反饋很糟糕“明明查到了天氣卻因?yàn)樾侣劷涌趻斓舳裁炊伎床坏健薄＝鉀Q方案實(shí)現(xiàn)優(yōu)雅降級(jí)。執(zhí)行器需要捕獲每個(gè)任務(wù)的異常并將部分成功的結(jié)果和友好的錯(cuò)誤提示一起返回給LLM。同時(shí)提示詞需要訓(xùn)練LLM學(xué)會(huì)處理這種“部分成功”的觀察讓它能夠基于已有信息繼續(xù)回答。例如觀察是“[get_weather]: 上海晴25度。[search_web]: 請(qǐng)求超時(shí)”LLM應(yīng)該能回答“查到上海天氣是晴天25度但暫時(shí)無法獲取最新新聞?！笨?上下文長度與并行結(jié)果整合并行執(zhí)行可能返回大量文本結(jié)果全部拼接到上下文中可能會(huì)超出LLM的令牌限制。解決方案引入結(jié)果摘要Summarization步驟。在執(zhí)行器聚合結(jié)果后可以先調(diào)用一個(gè)快速的“摘要LLM”如GPT-3.5-turbo將冗長的并行結(jié)果提煉成關(guān)鍵要點(diǎn)再將摘要放入主Agent的上下文中。這相當(dāng)于增加了一個(gè)“中間件”用很小的成本保護(hù)了核心上下文的長度。5.3 進(jìn)階優(yōu)化思路動(dòng)態(tài)并行度控制不是所有任務(wù)都適合無限并行。可以根據(jù)系統(tǒng)負(fù)載、工具類型I/O密集型 vs CPU密集型動(dòng)態(tài)調(diào)整并發(fā)任務(wù)的數(shù)量。例如使用信號(hào)量Semaphore來限制同時(shí)運(yùn)行的網(wǎng)絡(luò)請(qǐng)求數(shù)。優(yōu)先級(jí)調(diào)度為工具設(shè)置優(yōu)先級(jí)。在并行組中高優(yōu)先級(jí)的任務(wù)可以先獲取資源或者其超時(shí)時(shí)間更短確保關(guān)鍵路徑的響應(yīng)速度。與LangGraph結(jié)合LangGraph是LangChain中用于構(gòu)建有狀態(tài)、循環(huán)工作流的庫。我們可以將并行執(zhí)行器作為LangGraph中的一個(gè)節(jié)點(diǎn)Node。這個(gè)節(jié)點(diǎn)接收一個(gè)“并行動(dòng)作列表”并發(fā)執(zhí)行后將結(jié)果輸出到下一個(gè)節(jié)點(diǎn)。這樣能更靈活地編排包含并行、條件判斷、循環(huán)的復(fù)雜Agent流程?？梢暬c調(diào)試并行流程比串行更難調(diào)試?？梢蚤_發(fā)一個(gè)簡單的可視化面板記錄每個(gè)工具的啟動(dòng)時(shí)間、結(jié)束時(shí)間、狀態(tài)成功/失敗和耗時(shí)幫助開發(fā)者直觀定位性能瓶頸或錯(cuò)誤點(diǎn)。這次從串行到并行的Agent改造實(shí)戰(zhàn)讓我深刻體會(huì)到提升AI應(yīng)用性能不僅在于選用更快的模型或硬件更在于對(duì)應(yīng)用架構(gòu)和流程的精細(xì)設(shè)計(jì)。將Promise.all的并發(fā)思想融入Agent工具鏈?zhǔn)菍?duì)LangChain等框架能力邊界的一次有效拓展。它要求開發(fā)者不僅會(huì)調(diào)用API更要理解框架的執(zhí)行原理并敢于為特定場景定制解決方案。這個(gè)過程雖然充滿了挑戰(zhàn)但看到響應(yīng)時(shí)間從8秒降到2秒用戶體驗(yàn)獲得質(zhì)的飛躍時(shí)所有的努力都變得無比值得。 LangChain Agent并行工具鏈優(yōu)化：從串行瓶頸到3倍性能提升實(shí)戰(zhàn)
1. 項(xiàng)目緣起從單線程Agent到并行工具鏈的必然演進(jìn)最近在重構(gòu)一個(gè)智能客服的Agent項(xiàng)目遇到了一個(gè)典型的性能瓶頸。這個(gè)Agent需要同時(shí)調(diào)用多個(gè)外部API來回答用戶的一個(gè)復(fù)雜問題比如用戶問“幫我查一下明天上海的天氣并推薦一個(gè)適合這種天氣的戶外活動(dòng)再找一家附近評(píng)分高的餐廳”。在最初的版本里我讓Agent按順序執(zhí)行先調(diào)用天氣API等結(jié)果返回后再調(diào)用活動(dòng)推薦API最后再調(diào)用餐廳搜索API。邏輯上沒問題但實(shí)測下來一個(gè)請(qǐng)求的平均響應(yīng)時(shí)間竟然超過了8秒。用戶等8秒才看到第一個(gè)字這體驗(yàn)直接不及格。問題的核心就在于工具Tools的調(diào)用是串行的。在LangChain的默認(rèn)Agent執(zhí)行器AgentExecutor里它拿到一個(gè)需要調(diào)用多個(gè)工具的計(jì)劃Plan后會(huì)像一個(gè)嚴(yán)格的管家不緊不慢地、一個(gè)接一個(gè)地執(zhí)行。工具A不返回結(jié)果絕不開始工具B。這在很多需要信息聚合的場景下造成了大量不必要的等待時(shí)間。這讓我開始思考Agent的“思考”LLM推理可以是單線程的但“執(zhí)行”工具調(diào)用為什么不能是并發(fā)的呢尤其是在工具之間沒有強(qiáng)依賴關(guān)系的時(shí)候。于是“并行優(yōu)化”就成了這個(gè)階段必須啃下的硬骨頭。我調(diào)研了社區(qū)方案發(fā)現(xiàn)大家普遍會(huì)提到Promise.all這個(gè)思路——沒錯(cuò)就是前端開發(fā)里用來處理多個(gè)異步任務(wù)的那個(gè)Promise.all。它的理念完美契合這里的需求同時(shí)發(fā)起多個(gè)獨(dú)立的網(wǎng)絡(luò)I/O請(qǐng)求然后等待所有結(jié)果返回最后再統(tǒng)一處理。這不正是解決我串行調(diào)用痛點(diǎn)的良藥嗎但將Promise.all的思想移植到LangChain的Agent開發(fā)中遠(yuǎn)不是加幾行async/await那么簡單。它涉及到對(duì)LangChain工具調(diào)用機(jī)制的理解、執(zhí)行流程的重構(gòu)、以及錯(cuò)誤處理與降級(jí)策略的設(shè)計(jì)。接下來我就結(jié)合這次實(shí)戰(zhàn)拆解如何構(gòu)建一個(gè)支持并行工具鏈的Agent并分享其中幾個(gè)關(guān)鍵的“踩坑”與“填坑”過程。2. 理解LangChain Agent的執(zhí)行機(jī)制與工具鏈瓶頸在動(dòng)手優(yōu)化之前我們必須先弄清楚LangChain的Agent到底是怎么工作的。很多人把LangChain當(dāng)作一個(gè)“黑盒”只知道它能連接大模型和工具但如果不理解其內(nèi)部執(zhí)行流優(yōu)化就無從談起。2.1 Agent的核心循環(huán)計(jì)劃、執(zhí)行、觀察一個(gè)典型的LangChain Agent以ReAct模式為例工作流程可以簡化為一個(gè)循環(huán)計(jì)劃PlanLLM根據(jù)用戶輸入和當(dāng)前上下文包括歷史對(duì)話和上一步的工具觀察結(jié)果決定下一步該做什么。輸出通常是一個(gè)結(jié)構(gòu)化的動(dòng)作Action包含要調(diào)用的工具名tool和輸入?yún)?shù)tool_input。執(zhí)行ExecuteAgent執(zhí)行器AgentExecutor解析這個(gè)動(dòng)作找到對(duì)應(yīng)的工具函數(shù)傳入?yún)?shù)并執(zhí)行它。這通常是一個(gè)異步的I/O操作比如調(diào)用一個(gè)HTTP API、查詢數(shù)據(jù)庫或執(zhí)行一段計(jì)算。觀察Observe工具執(zhí)行完成后將結(jié)果Observation返回給Agent執(zhí)行器。循環(huán)判斷執(zhí)行器將工具觀察結(jié)果連同歷史信息再次喂給LLM讓LLM判斷是否已經(jīng)獲得了足夠信息來生成最終答案Final Answer還是需要繼續(xù)調(diào)用下一個(gè)工具。這個(gè)循環(huán)會(huì)一直持續(xù)直到LLM決定輸出最終答案。瓶頸就出現(xiàn)在第2步“執(zhí)行”階段。在默認(rèn)實(shí)現(xiàn)中AgentExecutor是順序處理每一個(gè)“動(dòng)作”的。即使LLM在一次推理中規(guī)劃了多個(gè)可以并行執(zhí)行的動(dòng)作例如“同時(shí)查詢天氣和新聞”執(zhí)行器也會(huì)把它們當(dāng)作一個(gè)動(dòng)作序列來處理先執(zhí)行A等A的結(jié)果返回后再讓LLM基于A的結(jié)果思考決定是否執(zhí)行B。這并沒有充分利用多個(gè)工具之間可能存在的獨(dú)立性。2.2 工具Tool的本質(zhì)與封裝在LangChain中一個(gè)Tool本質(zhì)上是一個(gè)可調(diào)用的函數(shù)它有著明確的名稱name、描述description和參數(shù)模式args_schema。當(dāng)LLM決定使用某個(gè)工具時(shí)它是在根據(jù)工具的描述來匹配用戶意圖。from langchain.tools import tool tool def get_weather(city: str) - str: 根據(jù)城市名查詢天氣情況。 # 模擬一個(gè)耗時(shí)的網(wǎng)絡(luò)請(qǐng)求 import asyncio await asyncio.sleep(2) # 假設(shè)請(qǐng)求耗時(shí)2秒 return f{city}的天氣是晴天25攝氏度。 # 工具會(huì)被封裝成具有標(biāo)準(zhǔn)化接口的對(duì)象供Agent調(diào)用。當(dāng)我們談?wù)摗肮ぞ哝湣睍r(shí)通常指的是Agent在一次任務(wù)中可能按順序或按條件調(diào)用的多個(gè)工具的集合。而“并行優(yōu)化”目標(biāo)就是打破這個(gè)鏈?zhǔn)街胁槐匾捻樞蚣s束讓其中可以并發(fā)的部分同時(shí)飛起來。2.3 默認(rèn)執(zhí)行器的性能分析為了量化瓶頸我寫了一個(gè)簡單的測試。創(chuàng)建三個(gè)工具每個(gè)工具模擬一個(gè)耗時(shí)2秒的I/O操作。讓Agent執(zhí)行一個(gè)需要調(diào)用這三個(gè)工具的任務(wù)。import asyncio import time async def run_sequential_agent(): # 假設(shè)這是順序執(zhí)行的Agent start time.time() result_a await tool_a() result_b await tool_b() result_c await tool_c() end time.time() print(f順序執(zhí)行總耗時(shí): {end - start:.2f}秒) # 輸出約 6.00秒 async def hypothetical_parallel_agent(): # 這是我們期望的并行執(zhí)行 start time.time() # 同時(shí)發(fā)起所有任務(wù) task_a asyncio.create_task(tool_a()) task_b asyncio.create_task(tool_b()) task_c asyncio.create_task(tool_c()) # 等待所有任務(wù)完成 results await asyncio.gather(task_a, task_b, task_c) end time.time() print(f并行執(zhí)行總耗時(shí): {end - start:.2f}秒) # 輸出約 2.01秒測試結(jié)果一目了然順序執(zhí)行耗時(shí)是各工具耗時(shí)的線性累加~6秒而理想中的并行執(zhí)行耗時(shí)約等于最慢的那個(gè)工具的耗時(shí)~2秒。對(duì)于用戶體驗(yàn)來說這3倍的差距就是“卡頓”和“流暢”的天壤之別。因此改造勢在必行。3. 設(shè)計(jì)并行化工具鏈的核心架構(gòu)并行化不是簡單地把所有工具調(diào)用都扔進(jìn)asyncio.gather。一個(gè)健壯的并行Agent架構(gòu)需要考慮任務(wù)規(guī)劃、依賴關(guān)系、執(zhí)行調(diào)度和結(jié)果整合。我的設(shè)計(jì)思路主要分為以下幾個(gè)層次。3.1 識(shí)別可并行任務(wù)LLM的“并行思維”提示首先我們需要讓LLM具備“并行規(guī)劃”的能力。在默認(rèn)的ReAct提示詞中LLM習(xí)慣于思考“下一步”做什么。我們需要引導(dǎo)它去思考“哪些步驟可以同時(shí)做”。這主要通過修改系統(tǒng)的提示詞System Prompt來實(shí)現(xiàn)。在給Agent的指令中需要明確加入并行化的引導(dǎo)“你是一個(gè)高效的任務(wù)規(guī)劃者。在規(guī)劃步驟時(shí)請(qǐng)仔細(xì)分析任務(wù)。如果多個(gè)子任務(wù)之間沒有依賴關(guān)系即任務(wù)B不需要任務(wù)A的結(jié)果作為輸入那么你應(yīng)該將這些子任務(wù)規(guī)劃為可以同時(shí)執(zhí)行。在你的輸出中對(duì)于可以并行的任務(wù)請(qǐng)用特定的格式例如PARALLEL: [動(dòng)作1, 動(dòng)作2, ...]來標(biāo)識(shí)?！崩鐚?duì)于查詢天氣和新聞的任務(wù)LLM應(yīng)該輸出思考用戶需要天氣和新聞這兩者沒有依賴關(guān)系。 行動(dòng)PARALLEL: [{tool: get_weather, tool_input: {city: 上海}}, {tool: get_news, tool_input: {topic: 科技}}]這樣我們后續(xù)的并行執(zhí)行器就可以解析這個(gè)PARALLEL標(biāo)記知道里面的動(dòng)作是可以并發(fā)執(zhí)行的。這一步是關(guān)鍵的前提它決定了并行化的潛力和正確性。3.2 構(gòu)建并行執(zhí)行器ParallelExecutor這是架構(gòu)的核心。我們需要?jiǎng)?chuàng)建一個(gè)新的執(zhí)行器繼承或重寫LangChain的AgentExecutor主要覆蓋_call或_atake_next_step方法中關(guān)于工具執(zhí)行的部分。其核心邏輯如下解析動(dòng)作檢查LLM輸出的動(dòng)作。如果是單個(gè)動(dòng)作按原流程處理。如果檢測到PARALLEL標(biāo)記則提取出里面的動(dòng)作列表。依賴檢查可選但建議對(duì)于標(biāo)記為并行的動(dòng)作列表執(zhí)行器可以做一個(gè)簡單的靜態(tài)檢查確保這些工具之間沒有直接的輸入輸出依賴這需要預(yù)先定義工具的資源依賴關(guān)系初期可以靠LLM和提示詞保證后期可以加入更嚴(yán)格的檢查。并發(fā)執(zhí)行使用asyncio.gather或anyio庫如果使用LangChain的異步生態(tài)并發(fā)執(zhí)行這些工具。為每個(gè)工具調(diào)用創(chuàng)建一個(gè)異步任務(wù)。結(jié)果收集與聚合等待所有并發(fā)任務(wù)完成。收集每個(gè)任務(wù)的結(jié)果或可能發(fā)生的異常。結(jié)果格式化與傳遞將并行執(zhí)行的結(jié)果整合成一個(gè)格式化的觀察字符串傳遞給LLM進(jìn)行下一步的“思考”。例如可以將結(jié)果合并為“天氣結(jié)果xxx新聞結(jié)果yyy”。import asyncio from typing import List, Dict, Any from langchain.agents import AgentExecutor from langchain.tools import BaseTool class ParallelAgentExecutor(AgentExecutor): 支持并行工具調(diào)用的Agent執(zhí)行器。 async def _execute_parallel_actions(self, actions: List[Dict]) - str: 并發(fā)執(zhí)行一組動(dòng)作。 async def run_tool(tool: BaseTool, tool_input: Dict): try: result await tool.arun(tool_input) # 使用工具的異步接口 return {tool: tool.name, status: success, result: result} except Exception as e: return {tool: tool.name, status: error, result: str(e)} # 創(chuàng)建所有工具任務(wù)的協(xié)程 tasks [] for action in actions: tool self.tools_by_name[action[tool]] tasks.append(run_tool(tool, action[tool_input])) # 并發(fā)執(zhí)行并等待所有結(jié)果 tool_results await asyncio.gather(*tasks, return_exceptionsFalse) # 格式化結(jié)果供LLM觀察 observations [] for res in tool_results: if res[status] success: observations.append(f{res[tool]}的結(jié)果是{res[result]}) else: observations.append(f調(diào)用{res[tool]}時(shí)出錯(cuò){res[result]}) return \n.join(observations) async def _atake_next_step(self, ...): # ... 原有的邏輯解析出next_action ... # 判斷是否為并行動(dòng)作 if isinstance(next_action, dict) and next_action.get(type) parallel: parallel_actions next_action[actions] observation await self._execute_parallel_actions(parallel_actions) # 將并行執(zhí)行的結(jié)果作為單次觀察返回 return AgentFinish(return_values{output: observation}, log...) else: # 走原有的單動(dòng)作執(zhí)行邏輯 return await super()._atake_next_step(...)3.3 處理依賴與錯(cuò)誤并行模式下的新挑戰(zhàn)并行化引入了新的復(fù)雜性主要集中在兩點(diǎn)依賴關(guān)系這是并行正確性的生命線。如果兩個(gè)本應(yīng)有依賴關(guān)系的任務(wù)被錯(cuò)誤地并行執(zhí)行就會(huì)得到錯(cuò)誤的結(jié)果。例如“計(jì)算公司總營收”這個(gè)任務(wù)依賴于“獲取各部門營收”這個(gè)子任務(wù)的結(jié)果。在提示詞工程階段我們必須通過大量示例Few-shot和清晰的指令訓(xùn)練LLM準(zhǔn)確識(shí)別依賴。此外可以在執(zhí)行器層面實(shí)現(xiàn)一個(gè)輕量級(jí)的依賴圖檢查如果工具注冊(cè)時(shí)聲明了其產(chǎn)出outputs和所需輸入inputs系統(tǒng)可以自動(dòng)判斷一組動(dòng)作是否可并行。錯(cuò)誤處理與降級(jí)在串行模型中一個(gè)工具失敗整個(gè)鏈就中斷了邏輯簡單。但在并行模型中情況更復(fù)雜部分失敗三個(gè)并行任務(wù)一個(gè)失敗兩個(gè)成功。我們是直接整體失敗還是繼續(xù)使用成功的結(jié)果通常為了更好的用戶體驗(yàn)我們選擇“部分降級(jí)”。執(zhí)行器需要捕獲每個(gè)任務(wù)的異常將成功的結(jié)果和友好的錯(cuò)誤信息如“暫時(shí)無法獲取新聞但已為您查詢到天氣”一起整合返回給LLM。LLM需要具備根據(jù)不完整信息繼續(xù)推理或給出提示的能力。超時(shí)控制并行執(zhí)行中最慢的任務(wù)決定整體耗時(shí)。我們需要為整個(gè)并行任務(wù)組設(shè)置一個(gè)總超時(shí)也要為每個(gè)獨(dú)立工具設(shè)置單獨(dú)的超時(shí)。防止一個(gè)“慢”工具拖死整個(gè)請(qǐng)求。可以使用asyncio.wait_for來實(shí)現(xiàn)。async def run_tool_with_timeout(tool, tool_input, timeout5): try: # 為單個(gè)工具調(diào)用設(shè)置超時(shí) result await asyncio.wait_for(tool.arun(tool_input), timeouttimeout) return {status: success, result: result} except asyncio.TimeoutError: return {status: error, result: f工具執(zhí)行超時(shí){timeout}秒} except Exception as e: return {status: error, result: f工具執(zhí)行錯(cuò)誤{e}}4. 實(shí)戰(zhàn)將串行Agent改造為并行Agent的完整步驟理論說完了我們來看一個(gè)具體的改造案例。假設(shè)我們有一個(gè)基于OpenAI Function Calling的Agent它原本串行調(diào)用get_weather、search_web和calculate三個(gè)工具。4.1 第一步定義支持并行規(guī)劃的提示詞這是最核心的一步。我們需要精心設(shè)計(jì)System Message和Few-shot Examples。from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder system_message 你是一個(gè)高級(jí)任務(wù)規(guī)劃與執(zhí)行Agent。你的目標(biāo)是高效、準(zhǔn)確地完成用戶請(qǐng)求。 **并行規(guī)劃規(guī)則** 1. 仔細(xì)分析用戶請(qǐng)求將其拆解為多個(gè)子任務(wù)。 2. 判斷子任務(wù)間的依賴關(guān)系。如果任務(wù)B**不需要**任務(wù)A的輸出作為輸入那么任務(wù)A和B就可以并行執(zhí)行。 3. 對(duì)于可以并行執(zhí)行的任務(wù)組請(qǐng)?jiān)谀愕妮敵鲋惺褂靡韵翵SON格式明確標(biāo)識(shí) json { type: parallel, actions: [ {tool: 工具A名稱, tool_input: {參數(shù): 值}}, {tool: 工具B名稱, tool_input: {參數(shù): 值}} ] }對(duì)于必須串行的任務(wù)或者單個(gè)任務(wù)請(qǐng)按正常格式輸出。示例1并行場景 用戶告訴我今天北京的天氣和最新的科技頭條新聞。 思考查詢天氣和搜索新聞是獨(dú)立任務(wù)無依賴。 輸出{ type: parallel, actions: [ {tool: get_weather, tool_input: {city: 北京}}, {tool: search_web, tool_input: {query: 科技 頭條 新聞}} ] }示例2串行場景 用戶先查一下上海的溫度如果高于30度就推薦一個(gè)室內(nèi)活動(dòng)。 思考推薦活動(dòng)依賴于天氣查詢的結(jié)果必須串行。 輸出 {tool: get_weather, tool_input: {city: 上海}} prompt ChatPromptTemplate.from_messages([ (system, system_message), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ])### 4.2 第二步創(chuàng)建自定義的Output Parser LangChain的Agent需要Output Parser來將LLM的文本輸出解析成結(jié)構(gòu)化的動(dòng)作。我們需要自定義一個(gè)解析器使其能識(shí)別我們定義的并行格式。 python from langchain.schema import AgentAction, AgentFinish import json import re class ParallelOutputParser: 解析LLM輸出識(shí)別并行動(dòng)作或單個(gè)動(dòng)作。 def parse(self, llm_output: str) - dict: # 首先嘗試匹配JSON格式的并行指令 json_match re.search(rjson\n(.*?)\n, llm_output, re.DOTALL) if json_match: try: data json.loads(json_match.group(1)) if data.get(type) parallel and actions in data: return {type: parallel, actions: data[actions]} except json.JSONDecodeError: pass # 如果不是并行格式嘗試按原有的單動(dòng)作格式解析這里簡化處理 # 假設(shè)原有解析邏輯能解析出 tool 和 tool_input # ... 原有的解析代碼 ... # 如果解析成功返回 {type: single, action: agent_action} # 如果解析為最終答案返回 AgentFinish # 兜底如果無法解析返回需要繼續(xù)思考或報(bào)錯(cuò) return {type: error, message: f無法解析LLM輸出: {llm_output[:100]}...}4.3 第三步實(shí)現(xiàn)自定義的并行Agent執(zhí)行器這里我們基于LangChain的AgentExecutor進(jìn)行擴(kuò)展。重點(diǎn)重寫_take_next_step方法或它的異步版本_atake_next_step。from langchain.agents import AgentExecutor from langchain.schema import AgentFinish import asyncio class CustomParallelExecutor(AgentExecutor): async def _execute_parallel_tools(self, actions: list) - str: 并發(fā)執(zhí)行工具并返回整合的觀察字符串。 tasks [] tool_name_to_func {tool.name: tool for tool in self.tools} for action_spec in actions: tool_name action_spec.get(tool) tool_input action_spec.get(tool_input, {}) tool tool_name_to_func.get(tool_name) if not tool: tasks.append(asyncio.sleep(0)) # 占位結(jié)果會(huì)是錯(cuò)誤 continue async def run_single(tool_obj, input_dict): try: # 使用異步調(diào)用并設(shè)置單個(gè)工具超時(shí) result await asyncio.wait_for(tool_obj.arun(input_dict), timeout10.0) return {tool: tool_obj.name, success: True, output: result} except asyncio.TimeoutError: return {tool: tool_obj.name, success: False, output: 請(qǐng)求超時(shí)} except Exception as e: return {tool: tool_obj.name, success: False, output: f錯(cuò)誤: {str(e)}} tasks.append(run_single(tool, tool_input)) # 并發(fā)執(zhí)行所有任務(wù) results await asyncio.gather(*tasks, return_exceptionsFalse) # 構(gòu)建觀察文本 observation_parts [] for res in results: if res[success]: observation_parts.append(f[{res[tool]}]: {res[output]}) else: observation_parts.append(f[{res[tool]} 失敗]: {res[output]}) return \n.join(observation_parts) async def _atake_next_step(self, name_to_tool_map, color_mapping, inputs, intermediate_steps): # 調(diào)用LLM獲取下一步?jīng)Q策 output await self.agent.aplan(intermediate_steps, **inputs) # 使用自定義解析器解析輸出 parsed self.output_parser.parse(output) if parsed[type] parallel: # 執(zhí)行并行工具組 observation await self._execute_parallel_tools(parsed[actions]) # 將并行執(zhí)行的結(jié)果作為一步觀察加入到歷史中 new_intermediate_steps intermediate_steps [(output, observation)] # 這里直接返回一個(gè)AgentFinish或者可以繼續(xù)循環(huán)。簡單起見我們讓LLM基于并行結(jié)果再思考一次。 # 更復(fù)雜的實(shí)現(xiàn)可以在這里不直接結(jié)束而是繼續(xù)循環(huán)。 return AgentFinish(return_values{output: f并行任務(wù)執(zhí)行完畢。結(jié)果如下\n{observation}}, logoutput) elif parsed[type] single: # 原有的單動(dòng)作執(zhí)行邏輯這里需要調(diào)用父類方法或類似邏輯 action parsed[action] tool name_to_tool_map[action.tool] observation await tool.arun(action.tool_input) return (action, observation) elif isinstance(parsed, AgentFinish): return parsed else: raise ValueError(f無法處理的解析結(jié)果: {parsed})4.4 第四步組裝并測試Agent將以上所有部件組裝起來創(chuàng)建一個(gè)完整的并行Agent。from langchain.chat_models import ChatOpenAI from langchain.agents import create_openai_functions_agent # 1. 定義工具 tools [get_weather_tool, search_web_tool, calculate_tool] # 2. 創(chuàng)建支持并行規(guī)劃的LLM鏈?zhǔn)褂米远x提示詞 llm ChatOpenAI(modelgpt-4, temperature0) agent create_openai_functions_agent(llm, tools, prompt) # 注意這里需要適配openai函數(shù)調(diào)用本身不支持我們自定義的格式。本例更適用于ReAct或自定義格式的Agent。 # 因此更常見的做法是使用initialize_agent并指定agent_type為zero-shot-react-description然后替換其executor和prompt。 # 為簡化示例我們假設(shè)已經(jīng)創(chuàng)建了一個(gè)能輸出我們所需格式的agent對(duì)象。 # 3. 創(chuàng)建自定義執(zhí)行器 agent_executor CustomParallelExecutor.from_agent_and_tools( agentagent, toolstools, max_iterations5, early_stopping_methodgenerate, ) # 4. 測試 async def test_agent(): result await agent_executor.arun({ input: 查詢今天上海和北京的天氣并搜索人工智能領(lǐng)域的最新突破。, chat_history: [] }) print(result) # 運(yùn)行測試 import asyncio asyncio.run(test_agent())5. 性能對(duì)比、踩坑記錄與進(jìn)階優(yōu)化完成基礎(chǔ)改造后進(jìn)行性能對(duì)比測試是必不可少的。同時(shí)在實(shí)際開發(fā)中我遇到了幾個(gè)典型的“坑”。5.1 性能對(duì)比數(shù)據(jù)我使用相同的三個(gè)模擬工具每個(gè)耗時(shí)2秒和相同的復(fù)雜查詢?nèi)蝿?wù)對(duì)比了改造前后的Agent。執(zhí)行模式平均響應(yīng)時(shí)間 (10次)資源占用 (峰值內(nèi)存)代碼復(fù)雜度原生串行Agent~6.05 秒較低低使用LangChain標(biāo)準(zhǔn)組件自定義并行Agent~2.10 秒略高并發(fā)協(xié)程高需自定義解析器、執(zhí)行器理想并行無開銷~2.00 秒--可以看到性能提升接近3倍與理論預(yù)期相符。微小的額外開銷主要來自并行任務(wù)調(diào)度和結(jié)果聚合的邏輯。對(duì)于I/O密集型任務(wù)網(wǎng)絡(luò)請(qǐng)求、數(shù)據(jù)庫查詢這種提升是決定性的。5.2 實(shí)戰(zhàn)中遇到的“坑”與解決方案坑1LLM的“并行規(guī)劃”不穩(wěn)定最初LLM并不總是按照我們期望的格式輸出。有時(shí)會(huì)忘記PARALLEL標(biāo)記有時(shí)會(huì)把有依賴的任務(wù)也并行。解決方案提示詞工程是關(guān)鍵。除了清晰的系統(tǒng)指令提供更多、更高質(zhì)量的Few-shot示例至關(guān)重要。示例需要覆蓋正例正確并行、負(fù)例錯(cuò)誤并行導(dǎo)致問題和邊界案例。此外可以在Output Parser中加入后處理邏輯如果LLM輸出了多個(gè)獨(dú)立動(dòng)作但沒有用并行標(biāo)記可以嘗試自動(dòng)將其“升級(jí)”為并行任務(wù)需謹(jǐn)慎確保依賴判斷準(zhǔn)確?？?工具副作用與資源競爭如果兩個(gè)并行工具都要寫入同一個(gè)文件或者修改同一個(gè)數(shù)據(jù)庫記錄就會(huì)引發(fā)競態(tài)條件。解決方案在工具設(shè)計(jì)層面要明確工具的“純度”和副作用。對(duì)于有副作用的工具需要在提示詞中明確告知LLM它們不能并行或者在執(zhí)行器層面通過“資源鎖”機(jī)制來序列化對(duì)特定資源的訪問。一個(gè)簡單的做法是為工具添加標(biāo)簽如{side_effect: high}執(zhí)行器在規(guī)劃時(shí)避開同時(shí)執(zhí)行高副作用工具?？?錯(cuò)誤處理的用戶體驗(yàn)最初并行任務(wù)中一個(gè)失敗我就讓整個(gè)Agent返回錯(cuò)誤。用戶反饋很糟糕“明明查到了天氣卻因?yàn)樾侣劷涌趻斓舳裁炊伎床坏健薄＝鉀Q方案實(shí)現(xiàn)優(yōu)雅降級(jí)。執(zhí)行器需要捕獲每個(gè)任務(wù)的異常并將部分成功的結(jié)果和友好的錯(cuò)誤提示一起返回給LLM。同時(shí)提示詞需要訓(xùn)練LLM學(xué)會(huì)處理這種“部分成功”的觀察讓它能夠基于已有信息繼續(xù)回答。例如觀察是“[get_weather]: 上海晴25度。[search_web]: 請(qǐng)求超時(shí)”LLM應(yīng)該能回答“查到上海天氣是晴天25度但暫時(shí)無法獲取最新新聞?！笨?上下文長度與并行結(jié)果整合并行執(zhí)行可能返回大量文本結(jié)果全部拼接到上下文中可能會(huì)超出LLM的令牌限制。解決方案引入結(jié)果摘要Summarization步驟。在執(zhí)行器聚合結(jié)果后可以先調(diào)用一個(gè)快速的“摘要LLM”如GPT-3.5-turbo將冗長的并行結(jié)果提煉成關(guān)鍵要點(diǎn)再將摘要放入主Agent的上下文中。這相當(dāng)于增加了一個(gè)“中間件”用很小的成本保護(hù)了核心上下文的長度。5.3 進(jìn)階優(yōu)化思路動(dòng)態(tài)并行度控制不是所有任務(wù)都適合無限并行。可以根據(jù)系統(tǒng)負(fù)載、工具類型I/O密集型 vs CPU密集型動(dòng)態(tài)調(diào)整并發(fā)任務(wù)的數(shù)量。例如使用信號(hào)量Semaphore來限制同時(shí)運(yùn)行的網(wǎng)絡(luò)請(qǐng)求數(shù)。優(yōu)先級(jí)調(diào)度為工具設(shè)置優(yōu)先級(jí)。在并行組中高優(yōu)先級(jí)的任務(wù)可以先獲取資源或者其超時(shí)時(shí)間更短確保關(guān)鍵路徑的響應(yīng)速度。與LangGraph結(jié)合LangGraph是LangChain中用于構(gòu)建有狀態(tài)、循環(huán)工作流的庫。我們可以將并行執(zhí)行器作為LangGraph中的一個(gè)節(jié)點(diǎn)Node。這個(gè)節(jié)點(diǎn)接收一個(gè)“并行動(dòng)作列表”并發(fā)執(zhí)行后將結(jié)果輸出到下一個(gè)節(jié)點(diǎn)。這樣能更靈活地編排包含并行、條件判斷、循環(huán)的復(fù)雜Agent流程?？梢暬c調(diào)試并行流程比串行更難調(diào)試?？梢蚤_發(fā)一個(gè)簡單的可視化面板記錄每個(gè)工具的啟動(dòng)時(shí)間、結(jié)束時(shí)間、狀態(tài)成功/失敗和耗時(shí)幫助開發(fā)者直觀定位性能瓶頸或錯(cuò)誤點(diǎn)。這次從串行到并行的Agent改造實(shí)戰(zhàn)讓我深刻體會(huì)到提升AI應(yīng)用性能不僅在于選用更快的模型或硬件更在于對(duì)應(yīng)用架構(gòu)和流程的精細(xì)設(shè)計(jì)。將Promise.all的并發(fā)思想融入Agent工具鏈?zhǔn)菍?duì)LangChain等框架能力邊界的一次有效拓展。它要求開發(fā)者不僅會(huì)調(diào)用API更要理解框架的執(zhí)行原理并敢于為特定場景定制解決方案。這個(gè)過程雖然充滿了挑戰(zhàn)但看到響應(yīng)時(shí)間從8秒降到2秒用戶體驗(yàn)獲得質(zhì)的飛躍時(shí)所有的努力都變得無比值得。