做 AI Agent 開發(fā)有一段時(shí)間的同學(xué)大概率會遇到這樣一個(gè)尷尬場景單個(gè) Agent 套上一個(gè)精心設(shè)計(jì)的 Prompt確實(shí)能完成一些單點(diǎn)任務(wù)比如改寫文案、抽取數(shù)據(jù)、寫一段代碼?？梢坏┤蝿?wù)變成“先做市場調(diào)研再寫技術(shù)方案然后生成匯報(bào) PPT 大綱最后安排成員跟進(jìn)”單 Agent 就開始頻繁失控——上下文越拖越長中間步驟忘了前面結(jié)論工具調(diào)用來回調(diào)不對最后的輸出質(zhì)量完全不可控。這類問題并不是 Prompt 寫得不夠好而是架構(gòu)選型出了問題。真實(shí)業(yè)務(wù)里的大型任務(wù)天然適合拆成多個(gè)角色、多個(gè)步驟、多種工具協(xié)同完成這背后就是多智能體協(xié)作與異步任務(wù)編排的用武之地。本文會圍繞 DeepAgents 多智能體協(xié)作這條主線從零實(shí)現(xiàn)一個(gè)可運(yùn)行的多智能體系統(tǒng)內(nèi)容包括子智能體設(shè)計(jì)、工具調(diào)用、主管-工人協(xié)作模式、異步任務(wù)編排以及 Harness 在其中的作用。無論你是剛接觸 AI 大模型應(yīng)用開發(fā)還是已經(jīng)在做 Agent 落地這套代碼和思路都可以直接參考。1. 背景與核心概念1.1 單智能體的能力邊界先給“智能體”下一個(gè)通俗定義Agent 大模型 記憶 工具 循環(huán)。大模型提供推理能力記憶保存上下文工具讓模型能操作真實(shí)世界循環(huán)讓模型可以不斷“觀察-思考-行動”直到任務(wù)完成。單 Agent 適合做短鏈路任務(wù)比如“總結(jié)這段文本”“把這份 JSON 轉(zhuǎn)成表格”。一旦任務(wù)鏈路變長它的問題就暴露得很明顯上下文窗口有限前面步驟的內(nèi)容會被截?cái)嗷蛳♂?。無法并行所有步驟都串行效率低。一個(gè) Agent 很難同時(shí)擅長規(guī)劃、檢索、編碼、寫作四種能力。單個(gè)環(huán)節(jié)失敗可能導(dǎo)致整個(gè)任務(wù)失敗沒有降級機(jī)制。這些問題催生了一個(gè)新思路與其訓(xùn)練一個(gè)萬能 Agent不如把任務(wù)拆開交給多個(gè)各司其職的子 Agent 協(xié)作完成。于是多智能體系統(tǒng)成為 2026 年 AI 應(yīng)用開發(fā)中最受關(guān)注的方向之一。1.2 多智能體協(xié)作解決什么問題多智能體協(xié)作的核心思想是把一個(gè)復(fù)雜任務(wù)拆解成多個(gè)子任務(wù)分別交給不同的智能體執(zhí)行再由編排層統(tǒng)一匯總?？梢园阉斫獬梢粋€(gè)項(xiàng)目組產(chǎn)品經(jīng)理負(fù)責(zé)拆需求開發(fā)工程師負(fù)責(zé)寫代碼測試工程師負(fù)責(zé)驗(yàn)證項(xiàng)目經(jīng)理負(fù)責(zé)協(xié)調(diào)進(jìn)度。每個(gè)角色只做自己最擅長的事通過統(tǒng)一的消息機(jī)制溝通最終由負(fù)責(zé)人整合結(jié)果。在工程上這種模式帶來的好處非常直接每個(gè) Agent 的上下文更短專注自身任務(wù)不容易被無關(guān)信息干擾。無依賴的子任務(wù)可以并行執(zhí)行大大縮短整體耗時(shí)。單個(gè) Agent 失敗時(shí)可以重試或重新派發(fā)不影響整個(gè)鏈路。系統(tǒng)更易擴(kuò)展新增一個(gè)能力只需要注冊一個(gè)新的子 Agent。多智能體在 AI 大模型應(yīng)用開發(fā)中的典型場景包括研究報(bào)告自動撰寫、競品分析、代碼倉庫巡檢、客服工單流轉(zhuǎn)、農(nóng)業(yè)大模型中的環(huán)境監(jiān)測與決策聯(lián)動等。1.3 DeepAgents 與 Harness 的關(guān)系“DeepAgents”強(qiáng)調(diào)的不是“很多個(gè) Agent 各說各話”而是帶有深度規(guī)劃、深度執(zhí)行和深度反思能力的智能體系統(tǒng)。一個(gè) DeepAgent 會先拆解目標(biāo)再逐步執(zhí)行過程中不斷檢查結(jié)果是否需要修正。多個(gè) DeepAgent 協(xié)作時(shí)需要一套框架負(fù)責(zé)調(diào)度、狀態(tài)管理、消息路由和錯(cuò)誤恢復(fù)。這個(gè)框架在社區(qū)中被稱作 Harness。你可以把 Agent 理解成員工Harness 理解成項(xiàng)目管理系統(tǒng)員工負(fù)責(zé)干活系統(tǒng)負(fù)責(zé)記錄進(jìn)度、分配任務(wù)、處理異常。2026 年Harness 工程已經(jīng)成為 AI 大模型開發(fā)里的基礎(chǔ)話題。越來越多的模型團(tuán)隊(duì)和應(yīng)用團(tuán)隊(duì)開始強(qiáng)調(diào)模型能力固然重要但控制模型運(yùn)行的 Harness 設(shè)計(jì)往往才是決定一個(gè) Agent 系統(tǒng)能不能穩(wěn)定落地的關(guān)鍵。2. 環(huán)境準(zhǔn)備與版本說明2.1 運(yùn)行環(huán)境本文的示例代碼以 Python 為主建議使用 Python 3.10 或更高版本因?yàn)楹竺娴漠惒骄幣艜玫?asyncio 和 dataclass高版本支持更完善。操作系統(tǒng)不限Windows、macOS、Linux 都可以。項(xiàng)目會調(diào)用大模型接口所以你需要準(zhǔn)備好一個(gè)模型服務(wù)的 API Key。為了示例通用性代碼以“兼容 OpenAI 接口的模型服務(wù)”為例Key 和 Base URL 都通過環(huán)境變量配置。你在實(shí)踐中可以換成任何兼容 OpenAI 協(xié)議的模型服務(wù)包括國內(nèi)外各類大模型平臺。2.2 創(chuàng)建項(xiàng)目并安裝依賴先創(chuàng)建一個(gè)項(xiàng)目目錄并初始化虛擬環(huán)境mkdir deepagents-demo cd deepagents-demo python -m venv .venv source .venv/bin/activate # Windows 下執(zhí)行 .venv\Scripts\activate本項(xiàng)目的依賴非常少核心只有一個(gè) OpenAI SDK。為了環(huán)境變量管理方便再加上 python-dotenv。pip install openai python-dotenv把依賴寫入 requirements.txtopenai1.35.0 python-dotenv1.0.02.3 配置模型服務(wù)在項(xiàng)目根目錄創(chuàng)建.env文件OPENAI_API_KEY你的_API_KEY OPENAI_BASE_URLhttps://你的模型服務(wù)地址 OPENAI_MODELgpt-4o-mini這里的OPENAI_MODEL建議按實(shí)際模型名填寫不同模型能力差異很大小模型適合做簡單抽取強(qiáng)模型適合做規(guī)劃和總結(jié)。在代碼中統(tǒng)一加載配置# config.py import os from dotenv import load_dotenv load_dotenv() API_KEY os.getenv(OPENAI_API_KEY) BASE_URL os.getenv(OPENAI_BASE_URL) MODEL os.getenv(OPENAI_MODEL, gpt-4o-mini)這里有一個(gè)工程建議API Key 千萬不要硬編碼在代碼里也不要提交到 Git 倉庫。用環(huán)境變量管理密鑰配合.env文件并在.gitignore中忽略它是最基本也最有效的安全習(xí)慣。2.4 項(xiàng)目結(jié)構(gòu)規(guī)劃本文最終會實(shí)現(xiàn)三個(gè)層級的完整示例單個(gè)子智能體、多智能體協(xié)作、異步任務(wù)編排。項(xiàng)目結(jié)構(gòu)建議如下deepagents-demo/ ├── .env ├── .gitignore ├── requirements.txt ├── config.py ├── agent.py ├── multi_agent.py ├── orchestrator.py └── demo_async.py后面每一個(gè)文件我都會給出完整代碼和說明。3. 核心概念A(yù)gent、Message、Tool 與 Harness在進(jìn)入代碼之前先把多智能體系統(tǒng)里最核心的四個(gè)概念講透。理解了它們后面看代碼會順暢很多。3.1 Agent有目標(biāo)、有工具、有循環(huán)的執(zhí)行單元一個(gè) Agent 的核心不是“調(diào)用一次大模型”而是“循環(huán)”。標(biāo)準(zhǔn)的 Agent 循環(huán)包含四步模型根據(jù)當(dāng)前消息決定調(diào)用哪個(gè)工具或直接輸出最終答案。系統(tǒng)執(zhí)行工具調(diào)用。把工具結(jié)果作為新消息返回給模型。模型再次判斷任務(wù)是否完成。這個(gè)循環(huán)會一直持續(xù)直到模型不再請求調(diào)用工具或者達(dá)到設(shè)定的最大步數(shù)。沒有這個(gè)循環(huán)大模型只能“說”不能“做”有了這個(gè)循環(huán)大模型才能通過工具影響真實(shí)系統(tǒng)。3.2 Message智能體之間協(xié)作的統(tǒng)一語言無論是單個(gè) Agent 內(nèi)部的多次調(diào)用還是多個(gè) Agent 之間的信息傳遞底層都統(tǒng)一為消息。一條消息最基本的字段是role消息角色常見的有 system、user、assistant、tool。content消息正文。tool_calls模型發(fā)起的工具調(diào)用請求包含函數(shù)名和參數(shù)。多智能體協(xié)作的本質(zhì)就是不同 Agent 之間互相構(gòu)造、交換、消費(fèi) Message。因此在設(shè)計(jì)系統(tǒng)時(shí)消息結(jié)構(gòu)越統(tǒng)一后續(xù)擴(kuò)展越容易。3.3 Tool給智能體裝上“手”工具的本質(zhì)是一個(gè)函數(shù)外加一份 JSON Schema 描述。描述告訴模型“這個(gè)函數(shù)叫什么、有哪些參數(shù)、什么時(shí)候應(yīng)該用”。模型本身不會執(zhí)行代碼它只生成工具調(diào)用的請求真正執(zhí)行的是我們自己的代碼。工具設(shè)計(jì)的好與壞直接決定 Agent 能不能穩(wěn)定完成任務(wù)。一個(gè)常見的坑是工具描述寫得太模糊模型不知道該在什么時(shí)候調(diào)用或者參數(shù)設(shè)計(jì)不合理模型總能生成各種奇怪的參數(shù)。3.4 Harness多智能體系統(tǒng)的執(zhí)行框架Harness 在前文提過這里再展開講。單個(gè) Agent 需要循環(huán)控制多個(gè) Agent 需要任務(wù)調(diào)度整個(gè)系統(tǒng)需要狀態(tài)管理、錯(cuò)誤處理、日志記錄。這一層基礎(chǔ)設(shè)施就是 Harness。可以把 Harness 理解成“智能體操作系統(tǒng)”。它不負(fù)責(zé)具體的業(yè)務(wù)推理而是負(fù)責(zé)決定哪個(gè) Agent 什么時(shí)候執(zhí)行、消息該路由給誰、工具調(diào)用失敗后怎么重試、任務(wù)依賴關(guān)系怎么解析、結(jié)果怎么匯聚。理解了 Harness 的概念也就理解了異步任務(wù)編排在系統(tǒng)中的位置編排器是 Harness 中負(fù)責(zé)“任務(wù)調(diào)度”的模塊而異步能力讓無依賴的任務(wù)真正并行起來。4. 實(shí)戰(zhàn)一實(shí)現(xiàn)一個(gè)可運(yùn)行的子智能體先從最基礎(chǔ)的子智能體開始。這一節(jié)的目標(biāo)是編寫一個(gè)具備工具調(diào)用循環(huán)能力的 Agent 類它能根據(jù)用戶問題自動決定是否調(diào)用工具并最終返回答案。4.1 定義一個(gè)模擬工具為了演示工具調(diào)用機(jī)制這里定義一個(gè)“天氣查詢”工具。為了不依賴外部 API它返回模擬數(shù)據(jù)但工具定義方式與真實(shí)場景完全一致。# tools.py import json def get_weather(city: str, date: str 今天) - str: 查詢指定城市在指定日期的天氣情況 weather_map { 北京: 晴23℃, 上海: 小雨26℃, 廣州: 多云30℃, } weather weather_map.get(city, 晴25℃) return json.dumps({city: city, date: date, weather: weather}, ensure_asciiFalse)這個(gè)函數(shù)接收城市和日期返回 JSON 字符串。實(shí)際項(xiàng)目中工具函數(shù)可能是查數(shù)據(jù)庫、調(diào)內(nèi)部接口、操作文件系統(tǒng)但封裝思路完全一樣輸入?yún)?shù)、返回結(jié)果把外部能力包裝成模型可調(diào)用的函數(shù)。4.2 把函數(shù)轉(zhuǎn)換成模型可識別的工具 SchemaOpenAI 兼容接口要求工具用 JSON Schema 描述。我們可以寫一個(gè)轉(zhuǎn)換函數(shù)也可以直接手寫 Schema# tools.py WEATHER_TOOL { type: function, function: { name: get_weather, description: 查詢指定城市在指定日期的天氣情況。當(dāng)用戶詢問天氣時(shí)使用。, parameters: { type: object, properties: { city: {type: string, description: 城市名稱}, date: {type: string, description: 日期默認(rèn)今天} }, required: [city] } } }注意這里的 description 寫得非常關(guān)鍵。模型靠這段描述來決定什么時(shí)候調(diào)用工具描述越具體行為越穩(wěn)定。4.3 編寫子智能體類現(xiàn)在實(shí)現(xiàn) Agent 類。它的核心是一個(gè)循環(huán)把消息發(fā)給模型如果模型返回工具調(diào)用就執(zhí)行工具、把結(jié)果追加到消息列表然后繼續(xù)循環(huán)如果模型返回普通回復(fù)就結(jié)束循環(huán)。# agent.py import json from openai import OpenAI from config import API_KEY, BASE_URL, MODEL class Agent: def __init__(self, name, system_prompt, toolsNone, tool_mapNone, modelNone, max_steps10): self.name name self.system_prompt system_prompt self.tools tools or [] self.tool_map tool_map or {} self.model model or MODEL self.max_steps max_steps self.messages [{role: system, content: system_prompt}] self.client OpenAI(api_keyAPI_KEY, base_urlBASE_URL) def run(self, user_input: str) - str: self.messages.append({role: user, content: user_input}) for step in range(self.max_steps): response self.client.chat.completions.create( modelself.model, messagesself.messages, toolsself.tools if self.tools else None, ) message response.choices[0].message self.messages.append(message) # 沒有工具調(diào)用說明模型已經(jīng)給出最終答案 if not message.tool_calls: return message.content # 遍歷模型發(fā)起的工具調(diào)用請求 for tool_call in message.tool_calls: fn_name tool_call.function.name fn_args json.loads(tool_call.function.arguments) result self._call_tool(fn_name, fn_args) self.messages.append({ role: tool, tool_call_id: tool_call.id, content: result, }) raise RuntimeError(fAgent {self.name} 達(dá)到最大步數(shù) {self.max_steps}任務(wù)未完成) def _call_tool(self, fn_name, fn_args): if fn_name not in self.tool_map: return json.dumps({error: f未知工具: {fn_name}}, ensure_asciiFalse) try: result self.tool_map[fn_name](**fn_args) return result if isinstance(result, str) else json.dumps(result, ensure_asciiFalse) except Exception as e: return json.dumps({error: str(e)}, ensure_asciiFalse)這里有幾個(gè)細(xì)節(jié)值得注意tools可能為空此時(shí)不傳 tools 參數(shù)避免模型被空工具列表迷惑。message.tool_calls是判斷循環(huán)是否繼續(xù)的關(guān)鍵信號。工具執(zhí)行結(jié)果以 role 為 tool 的消息回傳OpenAI 接口要求必須帶上tool_call_id。工具異常被捕獲后轉(zhuǎn)成 JSON 字符串返回給模型這樣模型能自己根據(jù)錯(cuò)誤調(diào)整參數(shù)而不是整個(gè)流程崩潰。4.4 運(yùn)行與驗(yàn)證寫一個(gè)入口腳本測試# demo_single.py from agent import Agent from tools import get_weather, WEATHER_TOOL system_prompt 你是一個(gè)天氣助手。當(dāng)用戶詢問天氣時(shí)調(diào)用 get_weather 工具查詢?nèi)缓蟾鶕?jù)結(jié)果回答。 agent Agent( nameweather_bot, system_promptsystem_prompt, tools[WEATHER_TOOL], tool_map{get_weather: get_weather}, ) result agent.run(北京今天天氣怎么樣) print(result)運(yùn)行python demo_single.py預(yù)期輸出類似北京今天天氣晴朗氣溫 23℃體感舒適適合外出活動。整個(gè)過程模型經(jīng)歷了“判斷需要調(diào)用工具 - 生成工具調(diào)用參數(shù) - 收到工具結(jié)果 - 生成最終回答 ”四個(gè)階段。雖然功能簡單但它已經(jīng)是完整的 Agent 閉環(huán)了。5. 實(shí)戰(zhàn)二多智能體協(xié)作單獨(dú)的 Agent 能跑通后下一步進(jìn)入多智能體協(xié)作實(shí)戰(zhàn)。5.1 架構(gòu)設(shè)計(jì)主管-工人模式多智能體協(xié)作的模式有很多比如對話模式、管道模式、層級模式。本文選擇最常見、也最容易落地的“主管-工人”模式Supervisor Agent負(fù)責(zé)任務(wù)拆解、結(jié)果質(zhì)量檢查和最終匯總。Worker Agent每個(gè) Worker 負(fù)責(zé)一個(gè)具體子任務(wù)。整個(gè)流程分三步用戶提交總?cè)蝿?wù)。Supervisor 將總?cè)蝿?wù)拆成 N 個(gè)子任務(wù)并指定每個(gè)子任務(wù)的執(zhí)行者。Worker 依次或并行執(zhí)行自己的子任務(wù)結(jié)果回傳給 Supervisor由 Supervisor 合成最終輸出。這種模式的好處是職責(zé)清晰。規(guī)劃、執(zhí)行、匯總分離后續(xù)想替換某個(gè) Worker或者新增一個(gè) Worker都不需要改動其他模塊。5.2 代碼實(shí)現(xiàn)新建multi_agent.py。這里復(fù)用上一節(jié)的Agent類通過不同 system prompt 來區(qū)分 Supervisor 和 Worker 的行為。# multi_agent.py import json from agent import Agent class MultiAgentSystem: def __init__(self, supervisor_prompt, worker_prompts, modelNone): self.supervisor Agent( namesupervisor, system_promptsupervisor_prompt, modelmodel, ) self.workers {} for name, prompt in worker_prompts.items(): self.workers[name] Agent( namename, system_promptprompt, modelmodel, ) def process(self, task: str) - str: # 第一步Supervisor 拆分任務(wù)輸出 JSON 數(shù)組 plan_text self.supervisor.run( f請拆分以下任務(wù)為子任務(wù)每個(gè)子任務(wù)指定執(zhí)行者。\n任務(wù){(diào)task}\n 輸出 JSON 數(shù)組格式為 [{\worker\: \worker名\, \task\: \子任務(wù)描述\}] ) subtasks self._parse_plan(plan_text) # 第二步逐個(gè)子任務(wù)分發(fā)給 Worker 執(zhí)行 worker_results [] for item in subtasks: worker_name item.get(worker) sub_task item.get(task) if worker_name not in self.workers: worker_results.append({worker: worker_name, error: 未知執(zhí)行者}) continue result self.workers[worker_name].run(sub_task) worker_results.append({worker: worker_name, result: result}) # 第三步Supervisor 匯總結(jié)果 summary self.supervisor.run( f以下是子任務(wù)執(zhí)行結(jié)果請綜合成一份完整報(bào)告\n f{json.dumps(worker_results, ensure_asciiFalse, indent2)} ) return summary staticmethod def _parse_plan(text: str): try: return json.loads(text) except json.JSONDecodeError: # 如果模型返回了多余的文字嘗試截取 JSON 部分 start text.find([) end text.rfind(]) 1 if start -1 or end 0: raise ValueError(f無法解析規(guī)劃結(jié)果: {text}) return json.loads(text[start:end]) if start 0 and end start else []回調(diào)解析是這段代碼里最值得注意的部分。大模型并不保證輸出純 JSON經(jīng)常出現(xiàn)“好的以下是拆分結(jié)果[...]”這種帶有前后綴的情況所以這里提供了容錯(cuò)解析。5.3 運(yùn)行與驗(yàn)證模擬一個(gè)實(shí)際場景寫一篇關(guān)于“AI 技術(shù)在農(nóng)業(yè)中的應(yīng)用”的科普文章。三個(gè) Worker 分別負(fù)責(zé)引言、核心應(yīng)用、未來展望。# demo_multi.py from multi_agent import MultiAgentSystem supervisor_prompt ( 你是一個(gè)高級項(xiàng)目經(jīng)理。你會把用戶任務(wù)拆解為多個(gè)子任務(wù) 分派給指定的 Worker 執(zhí)行。Worker 列表writer_intro、writer_core、writer_future。 最后你負(fù)責(zé)匯總所有 Worker 的結(jié)果。 ) worker_prompts { writer_intro: 你是科普文章引言撰寫專家負(fù)責(zé)寫文章引言300字以內(nèi)。, writer_core: 你是農(nóng)業(yè)科技領(lǐng)域?qū)＜邑?fù)責(zé)介紹AI在土壤監(jiān)測、氣象預(yù)警、智能灌溉施肥中的應(yīng)用500字以內(nèi)。, writer_future: 你是行業(yè)趨勢分析師負(fù)責(zé)寫AI農(nóng)業(yè)的未來展望300字以內(nèi)。, } system MultiAgentSystem( supervisor_promptsupervisor_prompt, worker_promptsworker_prompts, ) result system.process(寫一篇關(guān)于AI技術(shù)在農(nóng)業(yè)中的應(yīng)用的科普文章) print(result)python demo_multi.py運(yùn)行完成后輸出會是一篇結(jié)構(gòu)完整的科普文章。通過打印每個(gè) Worker 的返回結(jié)果你可以清楚觀察到多智能體協(xié)作的過程。6. 實(shí)戰(zhàn)三異步任務(wù)編排到了這一步“多智能體協(xié)作”已經(jīng)實(shí)現(xiàn)了但目前的 worker 執(zhí)行是串行的。如果任務(wù)一耗時(shí) 10 秒、任務(wù)二耗時(shí) 10 秒、任務(wù)三耗時(shí) 10 秒整體就要 30 秒。如果這三個(gè)任務(wù)互不依賴完全可以并行執(zhí)行總耗時(shí)只有 10 秒。這就是異步任務(wù)編排的價(jià)值。6.1 任務(wù)類型分析在真實(shí)的智能體系統(tǒng)中任務(wù)之間通常存在兩類關(guān)系無依賴關(guān)系可以并行執(zhí)行。有依賴關(guān)系B 必須等 A 完成才能執(zhí)行。比如“先調(diào)研再寫方案最后生成 PPT”這是一個(gè)鏈?zhǔn)揭蕾嚩巴瑫r(shí)調(diào)研市場、調(diào)研技術(shù)、調(diào)研競品”則是并行關(guān)系。一個(gè)合格的編排器應(yīng)該同時(shí)支持這兩種關(guān)系。6.2 用 asyncio 實(shí)現(xiàn)并行Python 原生的 asyncio 庫已經(jīng)提供了很好的并發(fā)支持。最簡單的并行執(zhí)行可以用asyncio.gather# demo_async.py import asyncio async def worker(task_name, duration): print(f[開始] {task_name}) await asyncio.sleep(duration) print(f[完成] {task_name}) return f{task_name}_result async def main(): results await asyncio.gather( worker(任務(wù)A, 2), worker(任務(wù)B, 3), worker(任務(wù)C, 1), ) print(results) if __name__ __main__: asyncio.run(main())運(yùn)行結(jié)果如下[開始] 任務(wù)A [開始] 任務(wù)B [開始] 任務(wù)C [完成] 任務(wù)C [完成] 任務(wù)A [完成] 任務(wù)B [任務(wù)A_result, 任務(wù)B_result, 任務(wù)C_result]注意這里三個(gè)任務(wù)同時(shí)開始總耗時(shí)只有約 3 秒而不是 6 秒。6.3 帶依賴關(guān)系的異步編排器gather只能處理“一批任務(wù)全部可以并行”的情況。遇到“任務(wù) D 依賴 B 和 C 的結(jié)果”這種場景需要設(shè)計(jì)一個(gè)更通用的編排器。實(shí)現(xiàn)思路每個(gè)任務(wù)聲明自己的依賴任務(wù)名。編排器維護(hù)一個(gè)“待執(zhí)行”集合和“已完成”字典。循環(huán)掃描把“依賴已全部完成”的任務(wù)取出來用 asyncio 并發(fā)執(zhí)行。執(zhí)行完成后更新結(jié)果直到所有任務(wù)完成。# orchestrator.py import asyncio from dataclasses import dataclass, field from typing import Callable, Awaitable dataclass class Task: name: str deps: list field(default_factorylist) coro: Callable[..., Awaitable] None async def execute(self, context: dict): kwargs {dep: context[dep] for dep in self.deps} context[self.name] await self.coro(**kwargs) class AsyncOrchestrator: def __init__(self, tasks: list): self.tasks {task.name: task for task in tasks} self.status {} # name - executed or not self.context {} def validate(self): for task in self.tasks.values(): for dep in task.deps: if dep not in self.tasks: raise ValueError(f任務(wù) {task.name} 依賴了不存在的任務(wù) {dep}) async def _process_ready_tasks(self): ready [] for name, task in self.tasks.items(): if name in self.status: continue if all(dep in self.status for dep in task.deps): ready.append(task) return ready async def run(self): self.validate() while len(self.status) len(self.tasks): ready await self._process_ready_tasks() if not ready: raise RuntimeError(存在循環(huán)依賴或無法調(diào)度的任務(wù)) await asyncio.gather(*[task.execute(self.context) for task in ready]) for task in ready: self.status[task.name] True return self.context這個(gè)編排器的核心是_process_ready_tasks每次只找“所有依賴都已完成”的任務(wù)并發(fā)執(zhí)行。執(zhí)行完一批更新狀態(tài)繼續(xù)下一批。6.4 測試編排器下面模擬一個(gè)多智能體異步編排場景Task A調(diào)研需求無依賴耗時(shí) 3 秒Task B技術(shù)預(yù)研依賴 A耗時(shí) 2 秒Task C競品分析無依賴耗時(shí) 2 秒Task D生成方案依賴 B 和 C耗時(shí) 2 秒# demo_orchestrator.py import asyncio from orchestrator import AsyncOrchestrator, Task async def research_requirement(): await asyncio.sleep(3) return 需求調(diào)研報(bào)告 async def tech_preview(需求調(diào)研報(bào)告): await asyncio.sleep(2) return f技術(shù)預(yù)研基于{需求調(diào)研報(bào)告} async def competitor_analysis(): await asyncio.sleep(2) return 競品分析報(bào)告 async def final_plan(技術(shù)預(yù)研, 競品分析報(bào)告): await asyncio.sleep(2) return f最終方案{技術(shù)預(yù)研} {競品分析報(bào)告} async def main(): orchestrator AsyncOrchestrator([ Task(name調(diào)研, deps[], cororesearch_requirement), Task(name預(yù)研, deps[調(diào)研], corotech_preview), Task(name競品, deps[], corocompetitor_analysis), Task(name方案, deps[預(yù)研, 競品], corofinal_plan), ]) results await orchestrator.run() print(results) if __name__ __main__: asyncio.run(main())執(zhí)行過程會充分體現(xiàn)異步編排的優(yōu)勢調(diào)研和競品分析并行啟動調(diào)研完成后預(yù)研開始競品完成后方案等待預(yù)研和競品都完成最后才執(zhí)行?？偤臅r(shí)從串行的 9 秒壓縮到 7 秒任務(wù)越多這種優(yōu)勢越明顯。6.5 把異步編排接入多智能體系統(tǒng)回到多智能體場景第 5 節(jié)的 Worker 執(zhí)行是串行的現(xiàn)在可以把每個(gè) Worker 封裝成 asyncio 任務(wù)交給編排器調(diào)度。思路是每個(gè) Worker 的run方法改成異步版本。根據(jù)任務(wù)依賴關(guān)系構(gòu)造 Task 列表。編排器管理整個(gè)執(zhí)行過程。Supervisor 在最后一個(gè)“匯總?cè)蝿?wù)”中接收所有 Worker 結(jié)果。這里不重復(fù)貼全部代碼核心改動是使用asyncio.to_thread把已有的同步 Agent.run 包裝成可異步執(zhí)行的任務(wù)asyncio.to_thread(worker_agent.run, sub_task)在真實(shí)生產(chǎn)環(huán)境中Agent 的run方法往往本身就是異步的因?yàn)樗鼉?nèi)部要調(diào)模型接口、調(diào)外部工具而這些 I/O 操作天然適合異步。7. 常見問題與排查思路多智能體系統(tǒng)開發(fā)中遇到的坑通常集中在循環(huán)控制、工具調(diào)用、消息管理、異步調(diào)度等幾個(gè)方面。下面整理一份排查清單。問題現(xiàn)象常見原因解決思路Agent 陷入死循環(huán)缺少終止條件模型反復(fù)調(diào)用工具設(shè)置 max_steps在 system prompt 中明確“不需要調(diào)用工具時(shí)就輸出最終結(jié)果”Agent 不調(diào)用工具直接編答案工具 description 不清晰或模型誤判自己知道答案強(qiáng)化工具描述觸發(fā)條件在 prompt 里加 few-shot 示例工具參數(shù)解析失敗模型返回的 JSON 參數(shù)與 schema 不一致捕獲 json.JSONDecodeError把錯(cuò)誤信息以 tool 消息回傳給模型讓它重新生成多個(gè) Agent 消息串了多個(gè) Agent 共用了同一個(gè) messages 列表確保每個(gè) Agent 實(shí)例持有獨(dú)立的 messages不要用全局消息列表API 請求超時(shí)模型響應(yīng)過慢或網(wǎng)絡(luò)不穩(wěn)定設(shè)置 timeout 參數(shù)實(shí)現(xiàn)指數(shù)退避重試長任務(wù)改異步異步編排卡死任務(wù)存在循環(huán)依賴或初始化失敗在 run() 開頭調(diào)用 validate()準(zhǔn)備超時(shí)機(jī)制成本失控工具循環(huán)次數(shù)過多模型越試越偏限制 max_steps優(yōu)先使用小模型對大模型調(diào)用設(shè)置每日預(yù)算和日志監(jiān)控7.1 更深入地看如何調(diào)試 Agent 循環(huán)調(diào)試 Agent 循環(huán)最直接的辦法是打印每一個(gè)步驟的消息內(nèi)容。在Agent.run中加一行日志就能看到每次模型返回的消息、工具調(diào)用、工具結(jié)果print(f[{self.name}] step {step}: {message})生產(chǎn)環(huán)境建議換成結(jié)構(gòu)化日志記錄步驟號、模型名、tokens、工具名稱、耗時(shí)。7.2 異步編排器常見陷阱異步編排最隱蔽的問題出現(xiàn)在共享狀態(tài)上。多個(gè)并發(fā)任務(wù)同時(shí)讀寫同一個(gè) dict可能出現(xiàn)數(shù)據(jù)覆蓋。我上面的示例里編排器用一個(gè) context 字典保存所有任務(wù)結(jié)果但每個(gè)任務(wù)寫入的是自己的 key因此并發(fā)安全。如果業(yè)務(wù)需要多個(gè) Worker 操作同一個(gè)全局狀態(tài)建議引入鎖或者使用不可變數(shù)據(jù)。另一個(gè)常見問題是某個(gè)任務(wù)異常后asyncio.gather會立即拋出異常導(dǎo)致其他正常任務(wù)被取消。真實(shí)場景中更推薦捕獲異常并記錄到結(jié)果中讓錯(cuò)誤任務(wù)留給上層處理而不是中斷整個(gè)鏈路。8. 最佳實(shí)踐與工程建議代碼能跑通只是第一步真正要落地到工程中還需要關(guān)注以下這些實(shí)踐。8.1 讓每個(gè) Agent 的職責(zé)保持單一不要試圖讓一個(gè) Agent 什么都干。“寫文章”和“查數(shù)據(jù)”是兩種能力“規(guī)劃”和“執(zhí)行”是兩種職責(zé)。一個(gè) Agent 的 system prompt 只描述一個(gè)核心目標(biāo)工具列表也保持精簡。工具太多會顯著增加模型選錯(cuò)工具的概率。8.2 設(shè)計(jì)穩(wěn)定的工具接口工具的參數(shù)盡量使用基礎(chǔ)類型字符串、數(shù)字、布爾值。避免讓模型生成復(fù)雜嵌套對象。工具的描述寫清楚觸發(fā)條件和參數(shù)含義。工具內(nèi)部做好輸入校驗(yàn)不要信任模型生成的任何字段。8.3 日志與可觀測性優(yōu)先多智能體系統(tǒng)比傳統(tǒng) API 服務(wù)復(fù)雜得多一次用戶請求可能觸發(fā) 10 次以上模型調(diào)用。沒有日志出了問題幾乎無法排查。建議記錄每一次模型請求的模型名、prompt 摘要、tokens、耗時(shí)。每一次工具調(diào)用的函數(shù)名、參數(shù)、返回值。每一次任務(wù)分發(fā)的來源和去向。全局鏈路 ID把一次用戶請求的所有日志串起來。8.4 使用模型分級策略強(qiáng)模型貴且慢弱模型便宜且快。工程上最常見的優(yōu)化是把任務(wù)按難度分級規(guī)劃任務(wù)、工具選擇、最終匯總使用強(qiáng)模型。簡單抽取、格式化、關(guān)鍵詞判斷使用小模型。一個(gè)系統(tǒng)如果全部調(diào)用最強(qiáng)模型成本會直線上升如果全部調(diào)用小模型拆解任務(wù)時(shí)又容易出錯(cuò)。8.5 安全邊界不可忽視Agent 一旦接了工具就等于把代碼執(zhí)行能力交給了模型。這里有三條底線最小權(quán)限原則Agent 的 API Key、數(shù)據(jù)庫賬號、文件系統(tǒng)權(quán)限只開到完成當(dāng)前任務(wù)所需的最小范圍。防止提示注入如果工具會讀取外部網(wǎng)頁或用戶上傳文件外部內(nèi)容里可能包含惡意指令。必須區(qū)分“系統(tǒng)指令”和“不可信外部內(nèi)容”后者只能作為數(shù)據(jù)處理不能直接作為指令執(zhí)行。生產(chǎn)環(huán)境操作必須審計(jì)所有涉及刪除、寫入、發(fā)布的動作記錄操作人、操作對象和結(jié)果。高風(fēng)險(xiǎn)操作建議加人工確認(rèn)環(huán)節(jié)。8.6 從原型到生產(chǎn)關(guān)注性能指標(biāo)原型階段只需要一個(gè) Agent 循環(huán)生產(chǎn)階段則要考慮吞吐量和延遲。異步編排在原型階段可能只帶來幾秒的優(yōu)化但在生產(chǎn)環(huán)境、任務(wù)數(shù)量較大時(shí)是決定性因素。建議在系統(tǒng)設(shè)計(jì)初期就把異步能力考慮進(jìn)去。9. 總結(jié)與學(xué)習(xí)路線到這里我們已經(jīng)從零實(shí)現(xiàn)了一個(gè)包含子智能體、多智能體協(xié)作和異步任務(wù)編排的最小系統(tǒng)。回看整條鏈路核心能力其實(shí)就三件事設(shè)計(jì)好單個(gè) Agent 的循環(huán)規(guī)劃好多個(gè) Agent 之間的分工與消息傳遞用異步編排把無依賴任務(wù)并行起來。配合一個(gè)理解“何時(shí)調(diào)度誰、何時(shí)停止循環(huán)、失敗如何恢復(fù)”的 Harness 框架一個(gè)可用的多智能體系統(tǒng)就能穩(wěn)定運(yùn)行。如果接下來你想繼續(xù)深入建議按這個(gè)順序?qū)W習(xí)長期記憶讓 Agent 在多次任務(wù)之間記住關(guān)鍵信息而不是每次從空白開始。工具調(diào)用可靠性研究 structured output、function calling 的進(jìn)階用法減少參數(shù)解析錯(cuò)誤。任務(wù)評估給輸出結(jié)果設(shè)計(jì)自動化評分機(jī)制用評估集驗(yàn)收每次 Prompt 改動。生產(chǎn)級 Harness 框架去閱讀 LangGraph、AutoGen、CrewAI 或者各家模型團(tuán)隊(duì)推出的 Agent Harness 實(shí)現(xiàn)對比它們的調(diào)度、重試、狀態(tài)持久化方案。異步任務(wù)隊(duì)列如果任務(wù)量再上臺階把進(jìn)程內(nèi) asyncio 換成消息隊(duì)列實(shí)現(xiàn)分布式多智能體系統(tǒng)。多智能體開發(fā)是一個(gè)典型的“看起來容易、做起來細(xì)節(jié)極多”的方向。建議你先跑通本文的代碼再自己換一個(gè)業(yè)務(wù)場景比如寫技術(shù)日報(bào)、自動整理會議紀(jì)要、做競品信息匯總把 Supervisor、Worker、異步編排三個(gè)模塊重新組合一遍。只有親手改過一輪遇到報(bào)錯(cuò)并排查過才能真正理解 Harness 為什么是多智能體系統(tǒng)的核心基礎(chǔ)設(shè)施。如果在運(yùn)行過程中遇到問題歡迎在評論區(qū)留言交流。