最近在折騰一個 AI Agent 項目技能越加越多調(diào)度卻越來越亂。原本只是想給助手加幾個 Skills結(jié)果發(fā)現(xiàn)真正卡脖子的不是“能不能調(diào)用”而是“該先調(diào)誰、怎么并發(fā)、失敗了怎么降級”。折騰一段時間之后我用 130 行核心代碼做了一個叫DeepAgentForce的輕量調(diào)度內(nèi)核專門解決 Agent Skills 的“絲滑”編排問題現(xiàn)在已經(jīng)開源。這 130 行代碼干的事情其實非常具體把技能聲明成帶依賴關(guān)系的節(jié)點(diǎn)自動拓?fù)渑判蚰懿⑿械募寄芙^不排隊依賴沒滿足的技能先等著隨時超時、隨時降級全程狀態(tài)可觀測。如果你在做 Agent 應(yīng)用或者準(zhǔn)備給現(xiàn)有助手塞一堆工具函數(shù)這個實現(xiàn)思路能讓你少踩很多坑。1. 為什么要給 Agent 做“Skills 調(diào)度”以及我踩了什么坑1.1 先聊清楚什么是 Agent SkillsAgent Skills 這個概念最近在圈子里被反復(fù)討論。它跟普通 Function Calling 最大的區(qū)別是函數(shù)是原子操作而 Skill 是一個“能力包”。一個 Skill 可以包含提示詞模板、子步驟、校驗規(guī)則、甚至內(nèi)部的一小段工作流。舉個例子一個“酒店推薦”技能它內(nèi)部可能要查庫存、算價格、比對評分、生成推薦理由這些子邏輯不需要 Agent 主腦去理解。Agent 只需要知道“我注冊了一個酒店推薦技能輸入是預(yù)算和城市輸出是推薦列表”然后就可以在合適的時機(jī)調(diào)用它。這種封裝方式很像螺絲刀套裝里的批頭——Agent 是手柄Skills 是不同規(guī)格的批頭任務(wù)來了換一個就行。好處是主腦不需要承載所有領(lǐng)域知識壞處是當(dāng)你有十幾個批頭的時候選哪個、按什么順序用、能不能同時用兩個就成了新的問題。1.2 我最初串行調(diào)用的慘痛教訓(xùn)我最早實現(xiàn)技能編排的方式非常原始if-else 串聯(lián)。用戶問“幫我推薦一個適合周末短途旅行的海邊城市”我就先調(diào)天氣技能再調(diào)交通技能最后調(diào)住宿技能一個接一個排著隊來。很快問題就暴露了。第一是順序經(jīng)常搞錯。有一次我寫反了依賴先跑去查住宿再查天氣結(jié)果給用戶推了一個周末有暴雨的城市。第二是慢每個技能調(diào)用都得花兩三秒三個技能串起來就是七八秒用戶早就沒了耐心。第三是上下文爆炸每個技能執(zhí)行的時候我把前面所有技能的完整輸出都塞給大模型結(jié)果 Prompt 越來越長后面技能的輸出質(zhì)量明顯變差。更難受的是維護(hù)性。技能一多if-else 就變成了蜘蛛網(wǎng)加一個新技能得想著放在哪個分支里刪一個舊技能還得擔(dān)心別的地方有沒有引用。跟幾個做 Agent 的朋友聊下來發(fā)現(xiàn)大家?guī)缀醵甲吡艘槐檫@條路只是有人把鍋甩給“模型太笨”有人開始懷疑架構(gòu)。問題不在模型在我的調(diào)度策略。我需要一個機(jī)制讓技能自己告訴我它們之間的依賴關(guān)系然后由調(diào)度器統(tǒng)一安排執(zhí)行順序。DeepAgentForce 就是從這個需求里長出來的。2. DeepAgentForce 的核心設(shè)計130 行怎么裝下這么多功能2.1 用聲明式描述技能而不是寫一堆 if-elseDeepAgentForce 的第一條設(shè)計原則是Skill 自己交代元數(shù)據(jù)。我不用去讀你的函數(shù)體你只需要在定義技能的時候聲明“我依賴誰”“我的優(yōu)先級是什么”“超時多久”“我輸入什么”。我用一個裝飾器搞定這件事skill( nameweather_query, dependencies[], priority20, timeout15, description查詢指定未來一周的天氣情況, ) async def weather_query(city: str, **ctx): ...這里面的字段含義很明確name技能的唯一標(biāo)識其他技能通過這個名字聲明依賴dependencies這個技能執(zhí)行前必須完成的技能名列表priority同一批次可執(zhí)行技能中的排序依據(jù)數(shù)值越小越先跑timeout單個技能的最長執(zhí)行時間超時直接判定失敗description給大模型規(guī)劃器看的說明也是自動路由的依據(jù)。為什么選用聲明式而不是把調(diào)用邏輯寫死在調(diào)度器里因為聲明式數(shù)據(jù)可以被檢查、被規(guī)劃、被可視化。調(diào)度器不需要理解“天氣查詢”的業(yè)務(wù)邏輯只需要知道它是一個沒有依賴的節(jié)點(diǎn)、優(yōu)先級是 20。這也讓后面接入大模型做意圖規(guī)劃變得非常簡單——模型讀 description 就夠了。2.2 一張隱形的執(zhí)行圖依賴怎么從“亂序”變成“有序”技能之間的依賴關(guān)系本質(zhì)是一個有向無環(huán)圖DAG。每個技能是一個節(jié)點(diǎn)依賴關(guān)系是邊?！熬频晖扑]”依賴“預(yù)算計算”和“天氣查詢”那么圖上就有兩條邊從后面兩個節(jié)點(diǎn)指向它。調(diào)度器拿到一個規(guī)劃列表后會先做兩件事建圖、算入度。入度是“我還差多少個依賴技能沒完成”的計數(shù)。入度為零意味著所有前置技能都跑完了這個節(jié)點(diǎn)立刻進(jìn)入可執(zhí)行隊列。這個邏輯我用一個很短的函數(shù)實現(xiàn)def build_graph(plan): nodes set(plan) inc {n: 0 for n in nodes} edges {} for n in nodes: for dep in SKILL_REGISTRY[n].dependencies: if dep not in nodes: continue edges.setdefault(dep, []).append(n) inc[n] 1 return inc, edges建完圖之后調(diào)度器會用一個隊列維護(hù)所有入度為零的節(jié)點(diǎn)誰先完成誰去給后置節(jié)點(diǎn)“解鎖”。這跟餐廳后廚出餐是一個套路蒸飯的時候同時切菜切完菜才能下鍋炒炒完才能裝盤裝盤不等備菜。后廚不會傻傻地先做完一道菜再做下一道Agent Skills 調(diào)度也不應(yīng)該。如果你發(fā)現(xiàn)技能圖里存在循環(huán)依賴比如 A 等 B、B 等 A拓?fù)渑判驎苯邮?。我選擇在初始化階段就拒絕這類非法配置寧可拋異常也不跑到運(yùn)行期才暴露問題。2.3 調(diào)度器內(nèi)部的狀態(tài)流轉(zhuǎn)為什么能做到“絲滑”有了圖和入度剩下的就是一個清晰的狀態(tài)機(jī)。每個技能節(jié)點(diǎn)在任何時刻都處于以下狀態(tài)之一狀態(tài)含義PENDING入度不為零等待前置技能完成READY入度為零已進(jìn)入可執(zhí)行隊列等待 worker 領(lǐng)取RUNNING正在執(zhí)行DONE正常完成產(chǎn)物已寫入上下文FAILED執(zhí)行異?；虺瑫rSKIPPED前置失敗導(dǎo)致不再需要執(zhí)行所謂“絲滑”拆開看就是三個能力能并行的不排隊、能恢復(fù)的不放棄、能繞開的不死磕。能并行的不排隊靠的是多個 worker 從同一個隊列里取任務(wù)能恢復(fù)的不放棄靠的是超時和重試能繞開的不死磕靠的是降級規(guī)則——某個非關(guān)鍵技能失敗后調(diào)度器可以按配置決定是中斷整條鏈路還是跳過它繼續(xù)跑下游。每次狀態(tài)變化調(diào)度器會廣播一個事件。我在事件里帶上技能名、狀態(tài)、耗時和結(jié)果摘要這樣上層可以做實時日志、鏈路追蹤、甚至畫一個動態(tài)執(zhí)行的儀表盤。這也是 DeepAgentForce 被我用在正式環(huán)境時最舒服的地方問題定位從“猜”變成了“看”。3. 從零實現(xiàn)一個可運(yùn)行的最小版本3.1 目錄結(jié)構(gòu)與技能注冊先搭一個清爽的骨架DeepAgentForce 的目錄結(jié)構(gòu)非常簡潔核心就幾個文件deepagentforce/ ├── engine.py # 調(diào)度引擎核心約 130 行 ├── registry.py # 技能注冊表與裝飾器 ├── planner.py # 大模型規(guī)劃器可選 ├── demo.py # 示例入口 └── skills/ └── travel.py # 示例技能集registry.py負(fù)責(zé)維護(hù)一個全局技能注冊表所有技能定義都往這個表里登記。它同時承擔(dān)一個重要的校驗工作技能定義合法性檢查。如果兩個技能重名、依賴項引用了不存在的技能、或者出現(xiàn)循環(huán)依賴注冊階段就能發(fā)現(xiàn)。下面是注冊表的核心實現(xiàn)SKILL_REGISTRY {} def skill( name: str, dependencies: list None, priority: int 10, timeout: int 30, description: str , ): def decorator(fn): SKILL_REGISTRY[name] { name: name, fn: fn, dependencies: dependencies or [], priority: priority, timeout: timeout, description: description, } return fn return decorator這種全局注冊的寫法可能被一些人詬病“不夠優(yōu)雅”但在輕量框架里它非常實用技能函數(shù)不需要被調(diào)度器顯式引用只要模塊被 import 過裝飾器就會把它登記進(jìn)去。我在實際項目里會在啟動時統(tǒng)一導(dǎo)入技能包目錄確保所有技能都完成注冊。3.2 核心調(diào)度引擎逐段解析130 行拆開看真正干活的代碼都在engine.py里。我把它壓縮到大約 130 行下面這段是為了方便講解稍微展開過的版本但邏輯完全一致。import asyncio import time from collections import deque class SkillEngine: def __init__(self, max_workers3): self.max_workers max_workers self.ctx {} async def run(self, plan: list): # 1. 建圖、算入度 inc {n: 0 for n in plan} edges {} for node in plan: meta SKILL_REGISTRY[node] for dep in meta[dependencies]: if dep in plan: edges.setdefault(dep, []).append(node) inc[node] 1 self.inc inc self.edges edges self.status {n: PENDING for n in plan} # 2. 創(chuàng)建就緒隊列入度為零的先入隊 ready_q asyncio.Queue() for node in plan: if inc[node] 0: ready_q.put_nowait(node) self.status[node] READY workers [ asyncio.create_task(self._worker(ready_q)) for _ in range(self.max_workers) ] await ready_q.join() for w in workers: w.cancel() return self.ctx async def _worker(self, ready_q): while True: name await ready_q.get() try: await self._execute(name) finally: ready_q.task_done() async def _execute(self, name): meta SKILL_REGISTRY[name] self.status[name] RUNNING start time.time() try: result await asyncio.wait_for( meta[fn](**self._build_input(name)), timeoutmeta[timeout], ) self.ctx[name] result self._solve_dependencies(name, finishedTrue) except asyncio.TimeoutError: self._handle_failure(name, ftimeout after {meta[timeout]}s) except Exception as e: self._handle_failure(name, str(e)) finally: self.status[name] DONE print(f[skill] {name} - {self.status[name]} ({time.time() - start:.2f}s)) def _build_input(self, name): # 從 ctx 里篩出技能聲明需要的字段不讓歷史數(shù)據(jù)全部灌進(jìn)去 meta SKILL_REGISTRY[name] return {k: self.ctx[k] for k in meta[dependencies]} def _solve_dependencies(self, name, finishedTrue): if name not in self.edges: return for child in self.edges[name]: self.inc[child] - 1 if self.inc[child] 0: self.status[child] READY asyncio.create_task(self._execute(child))有幾個細(xì)節(jié)值得單獨(dú)拿出來說。ready_q.join()是關(guān)鍵的阻塞點(diǎn)——它會等待隊列中所有任務(wù)都調(diào)用過task_done()才返回保證了run()不會提前結(jié)束。_execute里我用asyncio.wait_for實現(xiàn)了超時控制比手動計時更可靠。它會在超時時直接拋出TimeoutError掛在技能函數(shù)上的網(wǎng)絡(luò)請求也會被取消避免了線程泄漏。_solve_dependencies用asyncio.create_task去啟動下游技能而沒有繼續(xù)走隊列這樣能減少一次出隊入隊的延遲技能之間銜接更緊湊。我實測過微小但可感知的提速在任務(wù)是輕量計算時尤其明顯。3.3 讓大模型參與規(guī)劃最小接入方式調(diào)度器本身不挑任務(wù)它只需要一份“該執(zhí)行哪些技能”的列表。這份列表可以由規(guī)則生成也可以由大模型動態(tài)規(guī)劃。我把后者做成了一個單獨(dú)的planner.py只在需要的時候引入。核心思路很簡單把技能注冊表里的name和description用 JSON 格式傳給模型讓它輸出一個技能執(zhí)行序列數(shù)組。我給模型設(shè)計的提示詞模板如下你是任務(wù)規(guī)劃器。以下是可以調(diào)用的技能列表 [{name}, {description}]每項以JSON呈現(xiàn) 請根據(jù)用戶請求輸出一個按執(zhí)行順序排列的技能名稱數(shù)組。 只能使用列表中存在的技能名稱不要解釋原因。輸出必須是合法JSON數(shù)組。 用戶請求{user_request}解析響應(yīng)的部分我寫得比較寬容先用正則從模型輸出里抓取第一對[...]再從里面解析 JSON。如果解析失敗就退回到默認(rèn)順序。這段兜底邏輯很重要因為模型輸出經(jīng)常帶一些“思考過程”之類的多余文本硬解析很容易翻車。老實說大模型規(guī)劃沒有想象中那么難接難點(diǎn)反而在“模型返回了沒注冊的技能名”或者“漏掉了關(guān)鍵技能”。我的處理策略是解析完后做一遍白名單過濾不在注冊表里的名字直接丟棄保證調(diào)度器永遠(yuǎn)拿到合法輸入。4. 用“旅行規(guī)劃”場景驗證調(diào)度效果4.1 一個標(biāo)準(zhǔn)的五技能旅行規(guī)劃示例為了驗證 DeepAgentForce 不是玩具我搭建了一個旅行規(guī)劃 Demo。技能清單如下技能名作用依賴優(yōu)先級預(yù)計耗時budget_calc計算預(yù)算上限無51sweather_query查詢目的地天氣無103shotel_recommender推薦酒店budget_calc, weather_query502sitinerary_builder生成行程weather_query404scopywriter生成種草文案hotel_recommender, itinerary_builder902s這個依賴結(jié)構(gòu)里budget_calc和weather_query互不依賴應(yīng)該同時執(zhí)行hotel_recommender要等它們兩個完成才能啟動itinerary_builder只需要等weather_query所以它實際上和hotel_recommender是同一批次可以并行copywriter最后跑。技能定義代碼大概是這樣的風(fēng)格skill(namebudget_calc, dependencies[], priority5, timeout3, description根據(jù)用戶輸入計算行程預(yù)算) async def budget_calc(person_count: int 2, **ctx): return {budget: 5000} skill(nameweather_query, dependencies[], priority10, timeout15, description查詢目的地天氣情況) async def weather_query(city: str 某南方沿海城市, **ctx): ... skill(namehotel_recommender, dependencies[budget_calc, weather_query], priority50, timeout8, description根據(jù)預(yù)算和天氣推薦酒店) async def hotel_recommender(**ctx): budget ctx[budget_calc][budget] weather ctx[weather_query][forecast] ...你會發(fā)現(xiàn)我在技能函數(shù)里并沒有寫任何調(diào)度邏輯只是通過**ctx去拿上游產(chǎn)物。依賴關(guān)系全部體現(xiàn)在裝飾器參數(shù)里代碼可讀性一下子高了很多。4.2 實際運(yùn)行日志長什么樣跑起來之后日志輸出是這樣的我做了精簡但格式是真實的[10:00:01] READY budget_calc (priority5) [10:00:01] READY weather_query (priority10) [10:00:01] RUN budget_calc [10:00:01] RUN weather_query [10:00:03] DONE budget_calc (2.05s) [10:00:04] DONE weather_query (3.10s) [10:00:04] READY hotel_recommender [10:00:04] READY itinerary_builder [10:00:04] RUN hotel_recommender [10:00:04] RUN itinerary_builder [10:00:08] DONE itinerary_builder (4.02s) [10:00:09] DONE hotel_recommender (5.01s) [10:00:09] READY copywriter [10:00:09] RUN copywriter [10:00:11] DONE copywriter (2.11s)這份日志非常有用。你能直接看到哪些技能被并行執(zhí)行了哪些在等依賴哪些技能耗時長哪些技能拖慢了整條鏈。我在正式環(huán)境里加了事件埋點(diǎn)把同樣的日志推到圖表面板上效果跟 CI 流水線視圖類似每次 Agent 跑完我都能看到一張執(zhí)行瀑布圖。4.3 串行和并行的實測對比數(shù)據(jù)我拿同一個任務(wù)做了三組對照分別用最樸素的 if-else 串行、DeepAgentForce 并發(fā)調(diào)度、以及不帶超時控制的并發(fā)調(diào)度來跑。執(zhí)行方式總耗時失敗處理備注if-else 串行14.2s無一個掛了全掛每個技能 2-5s累計相加并發(fā)調(diào)度無超時8.7s無可能卡死兩個無依賴技能并行但有風(fēng)險DeepAgentForce8.1s每次任務(wù)自動超時含日志和狀態(tài)機(jī)開銷從數(shù)據(jù)看并發(fā)調(diào)度相比串行的收益是實打?qū)嵉乃奈鍌€技能的鏈路能省下幾乎一半時間。更重要的是超時控制把最壞執(zhí)行時間從“不可控”變成了“可控”這一點(diǎn)在生產(chǎn)環(huán)境的權(quán)重比提速還高。還有個容易被忽略的收益是 token 花費(fèi)。因為_build_input只挑依賴項產(chǎn)物喂給下游技能而不是把歷史記錄全部塞進(jìn)去后面的技能看到的輸入更精簡模型輸出質(zhì)量也穩(wěn)定了。我統(tǒng)計過改造后單個任務(wù)的 token 開銷降了大概三成。5. 常見問題與調(diào)試心得5.1 技能互相搶占 / 優(yōu)先級不生效有朋友跑起來第一件事就是問我為什么我設(shè)置了 priority兩個技能還是亂序執(zhí)行原因是入隊的時候我直接按技能列表的順序 put沒有對同批次優(yōu)先級做排序。要在 READY 階段嚴(yán)格按 priority 執(zhí)行就得改用優(yōu)先隊列或者入隊前先把同批次節(jié)點(diǎn)按 priority 排序。我的做法是后者因為更直觀ready_items [node for node in plan if inc[node] 0] ready_items.sort(keylambda n: SKILL_REGISTRY[n][priority]) for node in ready_items: ready_q.put_nowait(node)注意priority只在“同批次可執(zhí)行節(jié)點(diǎn)”之間比較不代表它能跨越依賴鏈提前執(zhí)行。如果你希望某個技能無論如何都先跑從設(shè)計上就應(yīng)該把它放進(jìn)依賴鏈的最上游而不是依賴優(yōu)先級這個弱信號。5.2 上下文爆炸問題我最初版本里有一個偷懶的寫法把ctx整個傳給每個技能。技能只要寫**ctx就能拿到上游產(chǎn)物寫起來很爽代價是上下文像雪球一樣滾。等到第五個技能執(zhí)行的時候它收到的是前面四個技能的全部中間產(chǎn)物包括一些完全用不上的日志、調(diào)試信息、甚至大模型生成的中間文案。解決思路就是我在_build_input里做的依賴項白名單只把dependencies里聲明過的技能產(chǎn)物傳給當(dāng)前技能。對于真需要全局變量的場景我額外留了一個shared字段只有顯式聲明了才會注入。這個改動讓我后續(xù)再也不用為“上下文長度超限”頭疼。5.3 腳本卡死 / 假死卡死問題最典型的原因是技能內(nèi)部發(fā)起了網(wǎng)絡(luò)請求但沒設(shè)置 IO 超時。調(diào)度器層超時只能兜底異常場景如果你每個技能都不設(shè) timeout而每次請求又恰好永遠(yuǎn)掛起那個技能就會一直處于 RUNNING 狀態(tài)。我自己的排查套路是兩步第一步看engine.py日志里有沒有技能長時間停留在 RUNNING第二步在技能函數(shù)里給網(wǎng)絡(luò)調(diào)用再包一層asyncio.wait_for不要把調(diào)度器的超時當(dāng)作最后防線。調(diào)度器的 timeout 是保險絲不是剎車片。5.4 大模型規(guī)劃不穩(wěn)定怎么辦模型把用戶請求解析成技能列表時偶爾會輸出“思考過程JSON”的混合文本或者干脆返回一個空數(shù)組。我在 planner 里加了三次容錯正則提取 JSON 數(shù)組、JSON 解析失敗后二次嘗試、最終回退到默認(rèn)技能順序。從這幾次迭代看最有效的還是把提示詞寫死成約束性語言只允許輸出 JSON、只允許使用白名單技能、不要解釋。如果你接的模型支持 JSON Schema 強(qiáng)制約束直接開啟約束模式這類問題基本絕跡。我用下來的經(jīng)驗是與其期望模型穩(wěn)定不如把模型當(dāng)不穩(wěn)定組件然后在外面加一層確定性的解析和過濾護(hù)欄。5.5 常見問題速查表現(xiàn)象可能原因解決方案兩個技能總是先執(zhí)行一個同級 READY 未按 priority 排序入隊前排序技能收到大量無用歷史數(shù)據(jù)ctx 整體透傳按 dependencies 篩選輸入任務(wù)長時間 RUNNING技能內(nèi)部網(wǎng)絡(luò)請求無 IO 超時技能內(nèi)單獨(dú)加 wait_for模型輸出非法 JSON模型返回自由文本正則抽取 白名單過濾 默認(rèn)順序回退技能 A 失敗后整個鏈路中斷沒有降級策略配置 skip_on_failure 標(biāo)記非關(guān)鍵技能最后再分享一個我實際用下來的心得。DeepAgentForce 這套 130 行的調(diào)度內(nèi)核表面上是在做任務(wù)編排本質(zhì)上是在劃定“模型”和“機(jī)制”的邊界。大模型負(fù)責(zé)看意圖、定方向調(diào)度器負(fù)責(zé)盯節(jié)奏、管異常、保底線。把這兩件事拆開之后Agent 的穩(wěn)定性才真正可控。我自己后續(xù)會在流式輸出和條件分支上繼續(xù)迭代比如讓某個技能按前序技能的不同結(jié)果走不同分支同時也想把執(zhí)行瀑布圖開源出來。如果你也遇到過技能調(diào)度亂成一團(tuán)的問題可以先拿這套思路去改自己手上的小項目別急著上重框架130 行很多時候已經(jīng)夠用了。