作：agency-agents框架編排實戰(zhàn))
如果你關注 AI 應用開發(fā)一定已經感覺到一個明顯風向大家慢慢不再滿足于“一個對話框 一個超長 Prompt”的單體智能體而是開始研究怎么讓多個智能體像一家小型公司那樣分工協(xié)作。去年我花了不少時間做多智能體編排前后試過好幾套思路最后沉淀出了一個叫“agency-agents”的輕量級框架。它的核心想法很簡單把復雜任務拆給不同角色的 Agent讓調度器決定誰在什么節(jié)點干活、上下文怎么傳、結果怎么匯總整體效果就像運營一家數字代理機構所以才取了這么個名字。這篇文章我會從設計動機講起然后是核心架構拆解、最小可運行實現(xiàn)、一個完整的內容生產流水線案例最后附上我在實際使用中踩過的坑。內容會偏工程但不會只丟代碼我會把每一步“為什么這么做”也講清楚。無論你是在研究多智能體框架想手寫一套自己的 Agent 編排系統(tǒng)還是只是想把大模型接進正式工作流這篇文章應該都能給你一些超出文檔之外的實際經驗。1. agency-agents 到底做了什么從一個痛點說起1.1 單 Agent 的天花板為什么必須拆先說說我最初遇到的真實現(xiàn)象。早期我用單個 Agent 去完成類似“幫我寫一份新品推廣方案”的任務時通常做法是準備一個非常長的系統(tǒng)提示詞里面包含你要扮演什么角色、項目背景、目標用戶、必須包含哪些章節(jié)、輸出格式要求、相關競品資料、語氣風格限制……這種“大而全”的 Prompt 初看效率很高一個請求全搞定但實際用一段時間就會發(fā)現(xiàn)問題。最大的問題是上下文競爭。角色設定和背景資料擠在一起模型注意力會被分散經常出現(xiàn)“前面說了面向年輕用戶后面調研部分又寫中年人市場”這種內部打架的情況。想改其中一個細節(jié)比如調整語氣風格你會擔心動到其他部分的穩(wěn)定性最后只能小心翼翼地重寫整段提示詞。而且當任務包含明顯不同性質的工作時比如先做資料檢索、再做競品分析、最后寫文案單個 Agent 很難保證每個環(huán)節(jié)都達到專業(yè)水準。我的實測是讓它先“認真檢索”再“動筆寫”它經常檢索完還沉浸在資料里輸出一段像報告不像文案的東西。某次我嘗試用一個超長 Pipeline 式 Prompt 跑一輪完整的市場調研加內容產出結果模型把戰(zhàn)略分析寫成了產品功能列表邏輯完全亂了。這時候我開始意識到真正靠譜的路徑不是把任務越堆越滿而是把任務拆開讓不同角色各管一段。這就是“agency-agents”最早的產品動機用多個分工明確的 Agent 模擬一個真實代理公司的協(xié)作方式而不是祈禱一個全知全能的大模型把一件事一次性做完美。1.2 拆成“數字代理機構”agency-agents 的核心思路你可以把 agency-agents 想象成一家小型廣告公司有客戶經理負責接需求有策略總監(jiān)負責拆解目標有調研專員去搜資料有文案或設計負責產出還有質檢人員做最終檢查。這些角色各自獨立但通過公司內部的交接流程協(xié)同起來。放到代碼層面這家“公司”由四個核心組件構成Agent角色實體只負責完成一項特定類型的工作內部封裝提示詞模板、模型調用、工具調用邏輯。Registry注冊中心所有 Agent 的身份登記處系統(tǒng)能通過 Agent 名稱查到它的實現(xiàn)、描述和配置。Dispatcher調度器相當于項目經理接收任務后按路由規(guī)則決定發(fā)給哪個 Agent并根據反饋決定下一步給誰。Context / Memory上下文與記憶一套跨 Agent 傳遞信息的機制保證每個環(huán)節(jié)都能拿到前面環(huán)節(jié)的結果且不會被無效信息淹沒。一個典型流程是這樣的任務進入 Dispatcher它先看任務類型如果命中“市場方案類”就先把任務派給 planning agent 拆解成步驟拆完確認需要做外部調研便再派給 research agent拿到調研結果后發(fā)給 writing agent 寫初稿最后給 reviewing agent 做質檢如果質檢不通過就打回修改通過則把最終結果寫回任務結果區(qū)。之所以選擇這種架構而不是直接寫一段“讓多個 Agent 通過自然語言互相聊天”的代碼原因很實際可調試性和可替換性。聊天式的多 Agent 協(xié)作看著很炫但一旦出現(xiàn)問題你很難說是哪個 Agent 理解錯了而通過 Dispatcher 控制編排順序每個環(huán)節(jié)的輸入輸出都有明確的記錄和格式出了錯直接把日志翻出來很容易定位到是哪一步、哪個 Agent、哪個提示詞出了問題。后面我的所有項目幾乎都在吃這個架構紅利。2. 核心架構拆解注冊中心、調度器與上下文傳遞2.1 讓每個 Agent 只做一件事角色定義與注冊機制在 agency-agents 里每個 Agent 被設計成只做一件事。我認為這是整個框架最重要的約束。舉例來說“研究員”就只負責收集和整理信息并輸出結構化摘要“寫手”就只負責根據摘要寫初稿“校對”就只負責提出修改意見。誰都不許越界誰都不許順手把別人的活也干了。代碼上我定義了一個非常薄的基礎類核心思想是“輸入一個任務對象輸出一個結果對象”# base_agent.py from abc import ABC, abstractmethod from dataclasses import dataclass, field dataclass class Task: id: str type: str payload: dict metadata: dict field(default_factorydict) dataclass class AgentResult: status: str # success / failed / needs_revision output: dict trace: list field(default_factorylist) class BaseAgent(ABC): name: str base description: str def __init__(self, llm_client, toolsNone, configNone): self.llm llm_client self.tools tools or {} self.config config or {} abstractmethod async def run(self, task: Task, context: Context) - AgentResult: ...然后通過一個注冊表把 Agent 的字符串名字和類關聯(lián)起來# registry.py AGENT_REGISTRY {} def register_agent(name): def decorator(cls): AGENT_REGISTRY[name] cls return cls return decorator這樣在別處定義 Agent 時一行裝飾器就能完成注冊register_agent(planner) class PlanningAgent(BaseAgent): async def run(self, task: Task, context): # 調用模型把任務拆解成步驟 steps await self.llm.chat(...) return AgentResult(statussuccess, output{steps: steps})為什么非要搞一套注冊機制而不是直接用 import 引用類因為在實際項目中Agent 的數量會從兩三個漲到十幾個如果每個 Agent 都在代碼里被手動引用就會出現(xiàn)大量重復的初始化代碼而且新增一個角色時要改多處調用點。注冊表本質上是一個“配置即代碼”的模式Agent 的實現(xiàn)和調度邏輯完全解耦后續(xù)甚至可以做成配置化注冊。實測下來這種方式在 Agent 數量超過 5 個后節(jié)省的維護成本非常明顯。2.2 調度器怎么知道“下一步該誰上”路由與狀態(tài)機Dispatcher 是整個系統(tǒng)的大腦。它的核心任務是維護一個執(zhí)行 DAG有向無環(huán)圖也就是把一條條 Agent 之間的流轉關系定義清楚。例如固定鏈式planner - researcher - writer - reviewer這種萬能流程。分支路由當任務類型是product時走產品組是content時走內容組。條件回溯當 reviewer 返回needs_revision時回給 writer但最多回兩次。實現(xiàn)時不需要太重一個簡單的狀態(tài)表就夠用# dispatcher.py class Dispatcher: def __init__(self, registry): self.registry registry async def dispatch(self, task: Task, context: Context, flow_map: dict[str, list[str]]): current_step flow_map.get(task.type, flow_map.get(default)) for agent_name in current_step: agent_cls self.registry.get(agent_name) agent agent_cls(...) result await agent.run(task, context) # 寫入上下文供后續(xù) Agent 使用 context.add_result(agent_name, result) # 特殊邏輯reviewer 打回修改 if result.status needs_revision and context.revision_count 2: context.revision_count 1 # 跳回上一個 agent return await self.dispatch(task, context, {default: current_step[:-1]}) return context關鍵點在于Agent 之間絕不能直接互相調用。哪怕寫手需要調研員的數據也不能在代碼里import ResearcherAgent然后直接調用它的方法而是要把調研結果寫到 Context 里再由 Dispatcher 在下一步把寫手喚醒并傳給它。這樣做的好處是每個 Agent 都可以獨立測試替換實現(xiàn)時不會牽連其他角色而且執(zhí)行路徑一目了然。項目中實際發(fā)生過多條 Agent 互相調到 Python 遞歸棧溢出的情況后來規(guī)定“Agent 只能見 Context不能見彼此”這個問題就根除了。2.3 上下文與記憶多 Agent 之間怎么傳話多 Agent 之間的“傳話”是另一個極其容易翻車的點。如果每個 Agent 都拿到完整的全部聊天記錄很快 Context 就會爆掉而且模型會分不清哪些是當前任務要處理的。agency-agents 的 Context 設計分兩層。短期上下文保存當前任務從開始到現(xiàn)在所有 Agent 的結果摘要每個 Agent 完成工作后把關鍵信息按固定 schema 寫進context.results長期記憶則是把過往任務沉淀到外部存儲Redis、向量庫或 JSON 文件里便于后續(xù)任務檢索。兩層的接口可以統(tǒng)一成簡單的get和set。# context.py class Context: def __init__(self, task_id: str): self.task_id task_id self.results {} self.revision_count 0 def add_result(self, agent_name: str, result: AgentResult): # 只保留結構化摘要不存完整原文 self.results[agent_name] { status: result.status, output: result.output, } def get_all_results(self): return self.results在實現(xiàn)時有一個很隱蔽的細節(jié)不要把完整原文丟進 Context。最開始我把 researcher 找到的幾萬字原始資料全部塞進context.results結果 writer 收到任務時光傳參就要傳幾萬 token既慢又貴。后來改為讓 researcher 必須輸出“結構化摘要”并且只輸出對當前寫稿真正有用的那部分信息。這相當于逼迫每個 Agent 做完信息壓縮再交接效果立竿見影——速度提升、費用下降、效果還更穩(wěn)定。建議在你這邊的實現(xiàn)中把“Agent 輸出必須比輸入短”作為一條硬性約定。3. 從零實現(xiàn)環(huán)境準備、代碼骨架與關鍵配置3.1 環(huán)境與依賴Python 3.10 asyncio 一個大模型接口這套框架我用 Python 3.10 實現(xiàn)異步用的是標準庫 asyncio外部依賴控制在很克制的范圍內。具體依賴如下# requirements.txt aiohttp3.9 PyYAML6.0 pydantic2.0 jinja23.0 redis5.0 # 可選用于長期記憶與隊列第三方包我只保留了“發(fā) HTTP 請求、讀配置、做模板渲染、做數據校驗”這幾件事其余都靠標準庫。之所以刻意保持輕量是因為 Agent 編排系統(tǒng)本身的復雜度不低如果底層再鋪一層重量級框架出了問題排查難度會成倍增加。異步框架選了 aiohttp沒有特別深層的原因主要是它穩(wěn)定而且對大模型接口的流式響應支持良好。如果你更熟悉 httpx完全可以用 httpx 替代。所有 Agent 共享同一個大模型客戶端封裝客戶端負責處理 API 調用、超時重試、流式輸出。實際項目中我把具體的模型調用隔離在llm_client.py這樣一個文件里切換模型供應商時只需要改這個文件。3.2 最小可運行骨架讓兩個 Agent 先跑起來演進一個新系統(tǒng)切忌一上來就寫一堆復雜流程。我先做了一個只有兩個 Agent 的最小 Demo一個 planner 負責把用戶需求拆成步驟一個 executor 負責按步驟返回模擬結果。跑通之后再逐步加入真正的模型調用和流程跳轉。# main.py import asyncio from context import Context from dispatcher import Dispatcher from registry import AGENT_REGISTRY, register_agent register_agent(planner) class PlannerAgent(BaseAgent): async def run(self, task: Task, context: Context): plan [市場調研, 方案撰寫, 審核修訂] return AgentResult(statussuccess, output{plan: plan}) register_agent(executor) class ExecutorAgent(BaseAgent): async def run(self, task: Task, context: Context): # 實際項目中這里會調用 LLM 或外部工具 result f已完成計劃: {context.results[planner][output][plan]} return AgentResult(statussuccess, output{detail: result}) async def main(): registry AGENT_REGISTRY task Task(id1, typedemo, payload{requirement: 做一份方案}) ctx Context(task_id1) dispatcher Dispatcher(registry) await dispatcher.dispatch(task, ctx, flow_map{demo: [planner, executor]}) print(ctx.results) asyncio.run(main())這個 Demo 的價值不在于功能而在于驗證流程骨架是否跑通。我就是從這一步開始逐步把BaseAgent內部的模型調用邏輯填進去把flow_map擴充成真正的搜索、寫作、審核鏈路。如果你是自己搭框架也一定要留一個這么小的“呼吸測試”每次改動大結構后先跑這個最小版本確認沒有破壞基礎鏈路。3.3 YAML 配置別把 Prompt 寫死在代碼里Agent 系統(tǒng)的提示詞是整個系統(tǒng)里變化最頻繁的部分。我的經驗是Prompt 絕對不能硬編碼在 Python 文件里否則改一句話就要重新部署一次。agency-agents 里所有 Agent 的提示詞和參數都放在 YAML 配置文件中# agents.yaml agents: planner: model: gpt-4o-mini temperature: 0.3 system_prompt: | 你是一名經驗豐富的項目策劃專家。請把用戶需求拆解為可執(zhí)行的步驟。 要求 1. 步驟不超過 5 個 2. 每個步驟必須包含目標描述 3. 直接輸出 JSON 數組不要額外解釋 writer: model: gpt-4o temperature: 0.7 system_prompt: | 你是一名資深內容創(chuàng)作者。請結合調研資料完成文案撰寫。 要求文風簡潔有力突出核心賣點。加載時用 YAML Jinja2 模板引擎。為什么不用 f-string 直接拼 Prompt因為 Agent 的 Prompt 里經常要嵌入多份調研摘要、用戶畫像、歷史記錄用 f-string 很容易寫出長到難以維護的表達式而且轉義問題很煩人。Jinja2 模板天然支持變量循環(huán)和條件判斷寫起來清楚還能提前在模板里加上默認值。# prompt_renderer.py from jinja2 import Environment, FileSystemLoader env Environment(loaderFileSystemLoader(prompts)) template env.get_template(writer_prompt.md.j2) prompt template.render(requirementtask.payload[requirement])這里埋一個我在實際項目里踩過的坑讓模型輸出的格式盡量保持“結構化但簡單”。剛開始我要求 Researcher 輸出包含多個嵌套層級、帶評分、帶優(yōu)先級的復雜 JSON模型倒是照著輸出了但我解析時經常遇到模型生成多余逗號、把null寫成None等問題。后來統(tǒng)一規(guī)定Agent 輸出的 JSON 最多兩層嵌套數組項控制在 20 個以內。換來的穩(wěn)定性不值那點信息量損失。3.4 接入外部能力的掛載點工具調用怎么設計一個純靠模型“腦補”的 Agent能力天花板非常明顯。遲早得讓 Agent 能搜索、能讀網頁、能查數據庫。我在框架設計里給 Agent 加了一個 tools 接口用意是讓 Agent 從“只會對話”進化為“能動手干活”。工具調用統(tǒng)一走 schema工具名稱 輸入參數字典返回必須是一個 JSON。# tool.py class Tool: async def execute(self, params: dict) - dict: raise NotImplementedError # 模擬一個搜索工具 class SearchTool(Tool): async def execute(self, params: dict): query params.get(query, ) # 真實項目里這里接搜索引擎 API return {results: [{title: f關于{query}的前幾條摘要, url: example.com}]}在 BaseAgent 內部模型輸出中如果包含工具調用指令就執(zhí)行對應工具并返回結果。這一段是標準套路我就不展開完整代碼了。但有一個方向建議不要一開始就把工具系統(tǒng)做得特別復雜先只加三個以內的工具比如搜索引擎和網頁正文提取、內部知識庫檢索跑通之后再慢慢增多。工具太多會讓 Agent 的意圖選擇準確率明顯下降這個小細節(jié)等到后面踩坑部分我會再提。4. 實戰(zhàn)用 agency-agents 搭建“選題—調研—寫作—質檢”內容流水線4.1 場景設定給一個小型營銷團隊做內容生產這次實戰(zhàn)我模擬的是一個真實發(fā)生在我工作里的場景某小型營銷工作室每周要產出一篇面向行業(yè)客戶的產品推廣長文。之前流程是人工搜集資料、人工寫初稿、人工審核一周一篇都很勉強。我們把這套流程搬上 agency-agents定義四個角色選題策劃根據目標讀者和產品賣點輸出選題方向和寫作大綱。調研員根據大綱去檢索資料輸出事實摘要和引用來源。寫手結合大綱和摘要寫初稿。審核員按清單檢查內容準確性、邏輯連貫性和表達力度給出修改意見。這個組合已經能覆蓋絕大多數內容類任務你也可以換成“產品經理 技術調研 文檔工程師 QA”來寫技術方案。4.2 Agent 分工與路由配置四類 Agent 的注冊代碼不復雜直接用第一節(jié)定義的基礎類擴展即可。真正的關鍵是 flow_map 怎么寫。我的設計是第一版用固定鏈式planner - researcher - writer - reviewerreviewer 返回needs_revision時跳回writer允許最多三次重試超過次數直接放行并標記“需人工復核”這個“三次上限”非常重要沒有它會出現(xiàn)無限循環(huán)燒錢的問題后面的踩坑部分會細講。Flow 配置放在 YAML 里起了作用# flow.yaml flows: content: steps: - planner - researcher - writer - reviewer max_revision: 3 fallback: human4.3 關鍵代碼串聯(lián)整個流水線主程序的核心就是在dispatch中傳入 flow 配置。注意我在ReviewerAgent中設置了一個檢查邏輯它既要判斷輸出內容是否滿足“包含核心關鍵詞”“有數據支撐”“無事實性錯誤”還要控制是否needs_revision。register_agent(reviewer) class ReviewerAgent(BaseAgent): async def run(self, task: Task, context): writer_output context.results[writer][output] # 這里調用 LLM 進行質量評估并返回 json: {passed: bool, comments: [...]} evaluation await self.llm.chat(...) if not evaluation[passed]: return AgentResult(statusneeds_revision, outputevaluation) return AgentResult(statussuccess, outputevaluation)運行時只需要在主入口里加載配置并調度async def run_flow(user_requirement: str): task Task(iduuid4().hex, typecontent, payload{requirement: user_requirement}) ctx Context(task_idtask.id) flow_config load_flow_config(flow.yaml) await dispatcher.dispatch(task, ctx, flow_config) return ctx.results整套系統(tǒng)跑起來后最開始的效果并不理想。調研員返回的資料要么太泛、要么太舊寫手寫出的稿子經常和產品賣點脫節(jié)。后來我逐步給每個 Agent 的 Prompt 里加入了“目標讀者鎖定”“競品關鍵詞映射表”這類約束情況才明顯改善。這里的重點不是代碼本身而是多 Agent 系統(tǒng)的成功與否很大程度取決于每個 Agent 的輸入輸出邊界定義得清不清楚這個在系統(tǒng)里是由 Prompt 和輸出格式共同保證的。4.4 調優(yōu)記錄如何把單次任務從 3 分鐘壓到 40 秒流水線跑通之后我發(fā)現(xiàn)單次任務耗時太長。最初版本是四個 Agent 串行執(zhí)行每個 Agent 調用一次大模型正常情況要 3 分鐘。我做了三件事終于把平均耗時壓到 40 秒上下第一步把調研員改成“內部并行檢索”一次觸發(fā) 3 個搜索請求等全部返回后一次性匯總。使用asyncio.gather幾乎是零成本提速。第二步給寫手配置一個更快的小模型承擔 70% 的常規(guī)內容生成審核員不滿意再切大模型。用“高中低配模型分層”的方式整體費用和速度最均衡。第三步減少串聯(lián)等待。讓審核員在檢查時就生成一份“修改建議清單”writer 接收到清單后直接定位到對應段落重寫而不是整篇重寫。效果比整篇重新生成要好得多。這一輪優(yōu)化給我的最大體會是多 Agent 系統(tǒng)不能一直加分堆邏輯它需要階段性做減法。把慢的環(huán)節(jié)并行掉把貴的環(huán)節(jié)便宜掉把重復的環(huán)節(jié)合并掉系統(tǒng)的價值才能體現(xiàn)出來。純堆 Agent 數量不一定帶來效率提升反而會讓調度成本、延遲和出錯概率一起變大。5. 老坑實錄死循環(huán)、并發(fā)限流、上下文爆掉5.1 死循環(huán)審核不通過時怎么辦第一次上線審核鏈路時我犯過一個挺蠢的錯誤審核不通過就直接讓寫手重寫寫手重寫之后審核又覺得不行無限循環(huán)下去。在真實測試中角色分配越明確這個循環(huán)越容易發(fā)生因為每個 Agent 都會堅定地堅持自己的判斷誰都不妥協(xié)。大模型不會累但你會看著賬單心疼。解決辦法就是前面提到過的“最多重試三次三次不過就轉人工”。在 Dispatcher 里增加一個簡單的全局計數每進入 writer 一次revision_count 1當達到上限時強制讓 reviewer 輸出“已通過但備注需人工確認”。同時在日志系統(tǒng)里記錄失敗原因便于分析瓶頸。總結成一句話所有帶反饋回路的流程必須設計終止條件否則系統(tǒng)遲早失控。5.2 并發(fā)限流頻繁 429 錯誤真正跑起來后第二個高頻問題是模型服務返回 429請求過多或者超時。多 Agent 并行化之后不僅沒解決這個問題反而讓情況更嚴重。我一開始天真地以為“并發(fā)越高越快”結果同時發(fā)出 10 個請求瞬間觸發(fā)了服務商的限流整個任務全部失敗重來比串行還慢。后來在llm_client里加了一個簡單的令牌桶限速器全局所有 Agent 共享保證每秒請求數不超過規(guī)定閾值。import asyncio import time class RateLimiter: def __init__(self, max_requests: int, per_seconds: float 1.0): self.max_requests max_requests self.period per_seconds self.tokens max_requests self.updated_at time.monotonic() async def acquire(self): while True: now time.monotonic() self.tokens min(self.max_requests, self.tokens (now - self.updated_at) / self.period * self.max_requests) self.updated_at now if self.tokens 1: self.tokens - 1 return await asyncio.sleep(0.05)這個限速器的粒度是整個進程級別同一時刻所有 Agent 共用一把鎖。如果是多機部署可以換成 Redis 分布式限流。這里要提醒的是具體閾值不能隨便拍腦袋我用的是“先設置低值、壓測后逐步上調”的方法找到一個既能跑得快又不觸發(fā) 429 的安全區(qū)間。5.3 上下文爆掉長文檔與多 Agent 記憶第三個高頻問題是 Context 無限膨脹。最初設計 Context 時以為把數據都放內存字典里就沒事結果跑了幾輪長文寫作任務內存上漲很快而且單次任務的 token 消耗也高得驚人。解決策略分兩層。短期上下文這邊我給每次寫入 Context 的內容設置了“摘要化”約束研究員輸出的內容必須壓縮到 500 字以內寫手寫作時讀到的是精煉摘要而不是海量原文。長期記憶這邊用向量庫存儲“任務級別的歷史結論”當下次遇到相近任務時只把最相關的 3 到 5 條歷史摘要注入到 Context 中而不是把所有歷史記錄都帶進去。順手還解決了一個容易被忽視的問題任務在傳遞長文檔時不應該傳文檔全文而是傳文檔 ID其他 Agent 需要時再按 ID 去文檔庫加載。這個設計初看只是省 token實際上還附帶解決了緩存一致性問題避免每個 Agent 各拿一份并各自腦補內容。5.4 工具返回格式不一致接口協(xié)議的坑最后還有一個細節(jié)工具調用多了以后“什么格式的返回結果”會成為災難。搜索工具返回的是數組數據庫查詢工具返回的是列表文件讀取工具返回的是字符串而 Agent 的 Prompt 里要求它統(tǒng)一處理這些數據模型很容易被格式帶偏經常出現(xiàn)“列表里套 JSON 字符串再套 Markdown”這種奇奇怪怪的嵌套。我后來規(guī)定所有工具輸出必須是統(tǒng)一的ToolResult結構內容字段是一個字符串但允許包含 Markdown結構化數據字段是一個 JSON 對象原始數據字段是數組或字典。Agent 的 Prompt 一律只讀取這兩種固定字段其他字段不做解析。這個約束極大降低了模型的解析錯誤率也讓調試日志變得清晰。如果到后期工具數量繼續(xù)增長建議順便引入 JSON Schema 做校驗而不是靠“約定”來約束。6. 后續(xù)擴展把模型層做實、把部署做穩(wěn)6.1 部署形態(tài)Docker Compose 一鍵啟動服務單體腳本跑通只是第一步實際要落地到一個團隊使用部署形態(tài)必須穩(wěn)定。我的做法是把主服務、Redis、日志系統(tǒng)用 Docker Compose 打包在一起同時把配置和 Prompt 模板掛載到外部目錄這樣改 Prompt 不用重新構建鏡像。# docker-compose.yml services: engine: build: . ports: - 8080:8080 volumes: - ./config:/app/config:ro - ./prompts:/app/prompts:ro environment: - LOG_LEVELINFO depends_on: - redis redis: image: redis:7-alpine volumes: - redis-data:/data這里有一個實戰(zhàn)建議Docker 容器里的 Python 進程要多開幾個 worker因為大模型接口的等待時間很長單 worker 很容易并發(fā)撐不住。我用的是 multiprocessing asyncio 的組合主進程收任務子進程跑 Agent 循環(huán)Redis 做任務隊列。這個組合方案雖然有些老土但勝在穩(wěn)定幾乎不需要額外維護。6.2 Agent 的可觀測性日志與鏈路追蹤多 Agent 系統(tǒng)的可觀測性比單 Agent 重要得多。原因很簡單單 Agent 出問題復盤時看一個輸入輸出就夠了多 Agent 出問題你可能需要看十幾個環(huán)節(jié)的輸入輸出如果沒有一個統(tǒng)一的 trace_id問題幾乎沒法排查。agency-agents 的做法是在 Task 創(chuàng)建時生成一個trace_id貫穿所有后續(xù)調用和日志每個 Agent 的日志輸出都帶上自己的名稱和 trace_id。日志格式直接用 JSON方便導入日志分析平臺{trace_id: 3f2a..., agent: writer, event: start, ts: 1720000000} {trace_id: 3f2a..., agent: writer, event: llm_call, model: gpt-4o-mini, latency_ms: 2100}千萬別小看這一招。有一次調研員總是返回過時數據我排查了半天沒有頭緒后來把搜索工具調用的 trace_id 對應日志一查發(fā)現(xiàn)是工具傳參時少了時間范圍限制導致模型默認用當前時間之前的所有結果。沒有鏈路日志這種跨組件問題可能要折騰好幾天。6.3 模塊化方向插件、協(xié)議與市場最后聊聊這套系統(tǒng)的未來擴展方向。我做這套框架時給 Agent 定義的接口足夠簡潔也意味著它可以很容易演變成一個半開放的插件系統(tǒng)。每種 Agent 可以對應一個 Python 包包內自帶配置模板和 Prompt 模板注冊中心負責加載。更進一步可以把 Agent 的輸入輸出協(xié)議標準化成 JSON Schema這樣不同團隊開發(fā)的 Agent 可以互相兼容整個“數字代理機構”就有機會變成一個生態(tài)。我目前的想法是往“協(xié)議優(yōu)先”的方向演進完全不管具體實現(xiàn)是什么只約好消息格式和接口契約A 團隊寫的調研員可以被 B 團隊的項目直接使用。當然這也帶來新的問題比如能力的可靠性評價、惡意 Agent 的隔離、消息的權限控制等等。不過這些屬于平臺化階段的事情當前階段最重要的是把單機的 Agent 編排邏輯磨得足夠穩(wěn)?；氐轿易约哼@套“agency-agents”框架至少有一點我確定用“代理機構”的思維去設計多 Agent 系統(tǒng)遠比給單個 Agent 堆砌更多 Prompt 要健康得多。如今每當我接到一個新任務第一反應不再是“我該怎么寫一個完美的提示詞”而是“這需要一個什么樣的團隊團隊里每個人干什么誰先誰后哪里需要檢查復核”。這種思維轉變大概就是這個項目給我留下的最實在的收獲。以后如果再投入時間我會優(yōu)先把 Agent 的“身份記憶”和“長期工作習慣”做深讓每個 Agent 在多次任務中越用越順手而不僅僅是完成一次孤立的工作。 從單Agent到多智能體協(xié)作：agency-agents框架編排實戰(zhàn)
如果你關注 AI 應用開發(fā)一定已經感覺到一個明顯風向大家慢慢不再滿足于“一個對話框 一個超長 Prompt”的單體智能體而是開始研究怎么讓多個智能體像一家小型公司那樣分工協(xié)作。去年我花了不少時間做多智能體編排前后試過好幾套思路最后沉淀出了一個叫“agency-agents”的輕量級框架。它的核心想法很簡單把復雜任務拆給不同角色的 Agent讓調度器決定誰在什么節(jié)點干活、上下文怎么傳、結果怎么匯總整體效果就像運營一家數字代理機構所以才取了這么個名字。這篇文章我會從設計動機講起然后是核心架構拆解、最小可運行實現(xiàn)、一個完整的內容生產流水線案例最后附上我在實際使用中踩過的坑。內容會偏工程但不會只丟代碼我會把每一步“為什么這么做”也講清楚。無論你是在研究多智能體框架想手寫一套自己的 Agent 編排系統(tǒng)還是只是想把大模型接進正式工作流這篇文章應該都能給你一些超出文檔之外的實際經驗。1. agency-agents 到底做了什么從一個痛點說起1.1 單 Agent 的天花板為什么必須拆先說說我最初遇到的真實現(xiàn)象。早期我用單個 Agent 去完成類似“幫我寫一份新品推廣方案”的任務時通常做法是準備一個非常長的系統(tǒng)提示詞里面包含你要扮演什么角色、項目背景、目標用戶、必須包含哪些章節(jié)、輸出格式要求、相關競品資料、語氣風格限制……這種“大而全”的 Prompt 初看效率很高一個請求全搞定但實際用一段時間就會發(fā)現(xiàn)問題。最大的問題是上下文競爭。角色設定和背景資料擠在一起模型注意力會被分散經常出現(xiàn)“前面說了面向年輕用戶后面調研部分又寫中年人市場”這種內部打架的情況。想改其中一個細節(jié)比如調整語氣風格你會擔心動到其他部分的穩(wěn)定性最后只能小心翼翼地重寫整段提示詞。而且當任務包含明顯不同性質的工作時比如先做資料檢索、再做競品分析、最后寫文案單個 Agent 很難保證每個環(huán)節(jié)都達到專業(yè)水準。我的實測是讓它先“認真檢索”再“動筆寫”它經常檢索完還沉浸在資料里輸出一段像報告不像文案的東西。某次我嘗試用一個超長 Pipeline 式 Prompt 跑一輪完整的市場調研加內容產出結果模型把戰(zhàn)略分析寫成了產品功能列表邏輯完全亂了。這時候我開始意識到真正靠譜的路徑不是把任務越堆越滿而是把任務拆開讓不同角色各管一段。這就是“agency-agents”最早的產品動機用多個分工明確的 Agent 模擬一個真實代理公司的協(xié)作方式而不是祈禱一個全知全能的大模型把一件事一次性做完美。1.2 拆成“數字代理機構”agency-agents 的核心思路你可以把 agency-agents 想象成一家小型廣告公司有客戶經理負責接需求有策略總監(jiān)負責拆解目標有調研專員去搜資料有文案或設計負責產出還有質檢人員做最終檢查。這些角色各自獨立但通過公司內部的交接流程協(xié)同起來。放到代碼層面這家“公司”由四個核心組件構成Agent角色實體只負責完成一項特定類型的工作內部封裝提示詞模板、模型調用、工具調用邏輯。Registry注冊中心所有 Agent 的身份登記處系統(tǒng)能通過 Agent 名稱查到它的實現(xiàn)、描述和配置。Dispatcher調度器相當于項目經理接收任務后按路由規(guī)則決定發(fā)給哪個 Agent并根據反饋決定下一步給誰。Context / Memory上下文與記憶一套跨 Agent 傳遞信息的機制保證每個環(huán)節(jié)都能拿到前面環(huán)節(jié)的結果且不會被無效信息淹沒。一個典型流程是這樣的任務進入 Dispatcher它先看任務類型如果命中“市場方案類”就先把任務派給 planning agent 拆解成步驟拆完確認需要做外部調研便再派給 research agent拿到調研結果后發(fā)給 writing agent 寫初稿最后給 reviewing agent 做質檢如果質檢不通過就打回修改通過則把最終結果寫回任務結果區(qū)。之所以選擇這種架構而不是直接寫一段“讓多個 Agent 通過自然語言互相聊天”的代碼原因很實際可調試性和可替換性。聊天式的多 Agent 協(xié)作看著很炫但一旦出現(xiàn)問題你很難說是哪個 Agent 理解錯了而通過 Dispatcher 控制編排順序每個環(huán)節(jié)的輸入輸出都有明確的記錄和格式出了錯直接把日志翻出來很容易定位到是哪一步、哪個 Agent、哪個提示詞出了問題。后面我的所有項目幾乎都在吃這個架構紅利。2. 核心架構拆解注冊中心、調度器與上下文傳遞2.1 讓每個 Agent 只做一件事角色定義與注冊機制在 agency-agents 里每個 Agent 被設計成只做一件事。我認為這是整個框架最重要的約束。舉例來說“研究員”就只負責收集和整理信息并輸出結構化摘要“寫手”就只負責根據摘要寫初稿“校對”就只負責提出修改意見。誰都不許越界誰都不許順手把別人的活也干了。代碼上我定義了一個非常薄的基礎類核心思想是“輸入一個任務對象輸出一個結果對象”# base_agent.py from abc import ABC, abstractmethod from dataclasses import dataclass, field dataclass class Task: id: str type: str payload: dict metadata: dict field(default_factorydict) dataclass class AgentResult: status: str # success / failed / needs_revision output: dict trace: list field(default_factorylist) class BaseAgent(ABC): name: str base description: str def __init__(self, llm_client, toolsNone, configNone): self.llm llm_client self.tools tools or {} self.config config or {} abstractmethod async def run(self, task: Task, context: Context) - AgentResult: ...然后通過一個注冊表把 Agent 的字符串名字和類關聯(lián)起來# registry.py AGENT_REGISTRY {} def register_agent(name): def decorator(cls): AGENT_REGISTRY[name] cls return cls return decorator這樣在別處定義 Agent 時一行裝飾器就能完成注冊register_agent(planner) class PlanningAgent(BaseAgent): async def run(self, task: Task, context): # 調用模型把任務拆解成步驟 steps await self.llm.chat(...) return AgentResult(statussuccess, output{steps: steps})為什么非要搞一套注冊機制而不是直接用 import 引用類因為在實際項目中Agent 的數量會從兩三個漲到十幾個如果每個 Agent 都在代碼里被手動引用就會出現(xiàn)大量重復的初始化代碼而且新增一個角色時要改多處調用點。注冊表本質上是一個“配置即代碼”的模式Agent 的實現(xiàn)和調度邏輯完全解耦后續(xù)甚至可以做成配置化注冊。實測下來這種方式在 Agent 數量超過 5 個后節(jié)省的維護成本非常明顯。2.2 調度器怎么知道“下一步該誰上”路由與狀態(tài)機Dispatcher 是整個系統(tǒng)的大腦。它的核心任務是維護一個執(zhí)行 DAG有向無環(huán)圖也就是把一條條 Agent 之間的流轉關系定義清楚。例如固定鏈式planner - researcher - writer - reviewer這種萬能流程。分支路由當任務類型是product時走產品組是content時走內容組。條件回溯當 reviewer 返回needs_revision時回給 writer但最多回兩次。實現(xiàn)時不需要太重一個簡單的狀態(tài)表就夠用# dispatcher.py class Dispatcher: def __init__(self, registry): self.registry registry async def dispatch(self, task: Task, context: Context, flow_map: dict[str, list[str]]): current_step flow_map.get(task.type, flow_map.get(default)) for agent_name in current_step: agent_cls self.registry.get(agent_name) agent agent_cls(...) result await agent.run(task, context) # 寫入上下文供后續(xù) Agent 使用 context.add_result(agent_name, result) # 特殊邏輯reviewer 打回修改 if result.status needs_revision and context.revision_count 2: context.revision_count 1 # 跳回上一個 agent return await self.dispatch(task, context, {default: current_step[:-1]}) return context關鍵點在于Agent 之間絕不能直接互相調用。哪怕寫手需要調研員的數據也不能在代碼里import ResearcherAgent然后直接調用它的方法而是要把調研結果寫到 Context 里再由 Dispatcher 在下一步把寫手喚醒并傳給它。這樣做的好處是每個 Agent 都可以獨立測試替換實現(xiàn)時不會牽連其他角色而且執(zhí)行路徑一目了然。項目中實際發(fā)生過多條 Agent 互相調到 Python 遞歸棧溢出的情況后來規(guī)定“Agent 只能見 Context不能見彼此”這個問題就根除了。2.3 上下文與記憶多 Agent 之間怎么傳話多 Agent 之間的“傳話”是另一個極其容易翻車的點。如果每個 Agent 都拿到完整的全部聊天記錄很快 Context 就會爆掉而且模型會分不清哪些是當前任務要處理的。agency-agents 的 Context 設計分兩層。短期上下文保存當前任務從開始到現(xiàn)在所有 Agent 的結果摘要每個 Agent 完成工作后把關鍵信息按固定 schema 寫進context.results長期記憶則是把過往任務沉淀到外部存儲Redis、向量庫或 JSON 文件里便于后續(xù)任務檢索。兩層的接口可以統(tǒng)一成簡單的get和set。# context.py class Context: def __init__(self, task_id: str): self.task_id task_id self.results {} self.revision_count 0 def add_result(self, agent_name: str, result: AgentResult): # 只保留結構化摘要不存完整原文 self.results[agent_name] { status: result.status, output: result.output, } def get_all_results(self): return self.results在實現(xiàn)時有一個很隱蔽的細節(jié)不要把完整原文丟進 Context。最開始我把 researcher 找到的幾萬字原始資料全部塞進context.results結果 writer 收到任務時光傳參就要傳幾萬 token既慢又貴。后來改為讓 researcher 必須輸出“結構化摘要”并且只輸出對當前寫稿真正有用的那部分信息。這相當于逼迫每個 Agent 做完信息壓縮再交接效果立竿見影——速度提升、費用下降、效果還更穩(wěn)定。建議在你這邊的實現(xiàn)中把“Agent 輸出必須比輸入短”作為一條硬性約定。3. 從零實現(xiàn)環(huán)境準備、代碼骨架與關鍵配置3.1 環(huán)境與依賴Python 3.10 asyncio 一個大模型接口這套框架我用 Python 3.10 實現(xiàn)異步用的是標準庫 asyncio外部依賴控制在很克制的范圍內。具體依賴如下# requirements.txt aiohttp3.9 PyYAML6.0 pydantic2.0 jinja23.0 redis5.0 # 可選用于長期記憶與隊列第三方包我只保留了“發(fā) HTTP 請求、讀配置、做模板渲染、做數據校驗”這幾件事其余都靠標準庫。之所以刻意保持輕量是因為 Agent 編排系統(tǒng)本身的復雜度不低如果底層再鋪一層重量級框架出了問題排查難度會成倍增加。異步框架選了 aiohttp沒有特別深層的原因主要是它穩(wěn)定而且對大模型接口的流式響應支持良好。如果你更熟悉 httpx完全可以用 httpx 替代。所有 Agent 共享同一個大模型客戶端封裝客戶端負責處理 API 調用、超時重試、流式輸出。實際項目中我把具體的模型調用隔離在llm_client.py這樣一個文件里切換模型供應商時只需要改這個文件。3.2 最小可運行骨架讓兩個 Agent 先跑起來演進一個新系統(tǒng)切忌一上來就寫一堆復雜流程。我先做了一個只有兩個 Agent 的最小 Demo一個 planner 負責把用戶需求拆成步驟一個 executor 負責按步驟返回模擬結果。跑通之后再逐步加入真正的模型調用和流程跳轉。# main.py import asyncio from context import Context from dispatcher import Dispatcher from registry import AGENT_REGISTRY, register_agent register_agent(planner) class PlannerAgent(BaseAgent): async def run(self, task: Task, context: Context): plan [市場調研, 方案撰寫, 審核修訂] return AgentResult(statussuccess, output{plan: plan}) register_agent(executor) class ExecutorAgent(BaseAgent): async def run(self, task: Task, context: Context): # 實際項目中這里會調用 LLM 或外部工具 result f已完成計劃: {context.results[planner][output][plan]} return AgentResult(statussuccess, output{detail: result}) async def main(): registry AGENT_REGISTRY task Task(id1, typedemo, payload{requirement: 做一份方案}) ctx Context(task_id1) dispatcher Dispatcher(registry) await dispatcher.dispatch(task, ctx, flow_map{demo: [planner, executor]}) print(ctx.results) asyncio.run(main())這個 Demo 的價值不在于功能而在于驗證流程骨架是否跑通。我就是從這一步開始逐步把BaseAgent內部的模型調用邏輯填進去把flow_map擴充成真正的搜索、寫作、審核鏈路。如果你是自己搭框架也一定要留一個這么小的“呼吸測試”每次改動大結構后先跑這個最小版本確認沒有破壞基礎鏈路。3.3 YAML 配置別把 Prompt 寫死在代碼里Agent 系統(tǒng)的提示詞是整個系統(tǒng)里變化最頻繁的部分。我的經驗是Prompt 絕對不能硬編碼在 Python 文件里否則改一句話就要重新部署一次。agency-agents 里所有 Agent 的提示詞和參數都放在 YAML 配置文件中# agents.yaml agents: planner: model: gpt-4o-mini temperature: 0.3 system_prompt: | 你是一名經驗豐富的項目策劃專家。請把用戶需求拆解為可執(zhí)行的步驟。 要求 1. 步驟不超過 5 個 2. 每個步驟必須包含目標描述 3. 直接輸出 JSON 數組不要額外解釋 writer: model: gpt-4o temperature: 0.7 system_prompt: | 你是一名資深內容創(chuàng)作者。請結合調研資料完成文案撰寫。 要求文風簡潔有力突出核心賣點。加載時用 YAML Jinja2 模板引擎。為什么不用 f-string 直接拼 Prompt因為 Agent 的 Prompt 里經常要嵌入多份調研摘要、用戶畫像、歷史記錄用 f-string 很容易寫出長到難以維護的表達式而且轉義問題很煩人。Jinja2 模板天然支持變量循環(huán)和條件判斷寫起來清楚還能提前在模板里加上默認值。# prompt_renderer.py from jinja2 import Environment, FileSystemLoader env Environment(loaderFileSystemLoader(prompts)) template env.get_template(writer_prompt.md.j2) prompt template.render(requirementtask.payload[requirement])這里埋一個我在實際項目里踩過的坑讓模型輸出的格式盡量保持“結構化但簡單”。剛開始我要求 Researcher 輸出包含多個嵌套層級、帶評分、帶優(yōu)先級的復雜 JSON模型倒是照著輸出了但我解析時經常遇到模型生成多余逗號、把null寫成None等問題。后來統(tǒng)一規(guī)定Agent 輸出的 JSON 最多兩層嵌套數組項控制在 20 個以內。換來的穩(wěn)定性不值那點信息量損失。3.4 接入外部能力的掛載點工具調用怎么設計一個純靠模型“腦補”的 Agent能力天花板非常明顯。遲早得讓 Agent 能搜索、能讀網頁、能查數據庫。我在框架設計里給 Agent 加了一個 tools 接口用意是讓 Agent 從“只會對話”進化為“能動手干活”。工具調用統(tǒng)一走 schema工具名稱 輸入參數字典返回必須是一個 JSON。# tool.py class Tool: async def execute(self, params: dict) - dict: raise NotImplementedError # 模擬一個搜索工具 class SearchTool(Tool): async def execute(self, params: dict): query params.get(query, ) # 真實項目里這里接搜索引擎 API return {results: [{title: f關于{query}的前幾條摘要, url: example.com}]}在 BaseAgent 內部模型輸出中如果包含工具調用指令就執(zhí)行對應工具并返回結果。這一段是標準套路我就不展開完整代碼了。但有一個方向建議不要一開始就把工具系統(tǒng)做得特別復雜先只加三個以內的工具比如搜索引擎和網頁正文提取、內部知識庫檢索跑通之后再慢慢增多。工具太多會讓 Agent 的意圖選擇準確率明顯下降這個小細節(jié)等到后面踩坑部分我會再提。4. 實戰(zhàn)用 agency-agents 搭建“選題—調研—寫作—質檢”內容流水線4.1 場景設定給一個小型營銷團隊做內容生產這次實戰(zhàn)我模擬的是一個真實發(fā)生在我工作里的場景某小型營銷工作室每周要產出一篇面向行業(yè)客戶的產品推廣長文。之前流程是人工搜集資料、人工寫初稿、人工審核一周一篇都很勉強。我們把這套流程搬上 agency-agents定義四個角色選題策劃根據目標讀者和產品賣點輸出選題方向和寫作大綱。調研員根據大綱去檢索資料輸出事實摘要和引用來源。寫手結合大綱和摘要寫初稿。審核員按清單檢查內容準確性、邏輯連貫性和表達力度給出修改意見。這個組合已經能覆蓋絕大多數內容類任務你也可以換成“產品經理 技術調研 文檔工程師 QA”來寫技術方案。4.2 Agent 分工與路由配置四類 Agent 的注冊代碼不復雜直接用第一節(jié)定義的基礎類擴展即可。真正的關鍵是 flow_map 怎么寫。我的設計是第一版用固定鏈式planner - researcher - writer - reviewerreviewer 返回needs_revision時跳回writer允許最多三次重試超過次數直接放行并標記“需人工復核”這個“三次上限”非常重要沒有它會出現(xiàn)無限循環(huán)燒錢的問題后面的踩坑部分會細講。Flow 配置放在 YAML 里起了作用# flow.yaml flows: content: steps: - planner - researcher - writer - reviewer max_revision: 3 fallback: human4.3 關鍵代碼串聯(lián)整個流水線主程序的核心就是在dispatch中傳入 flow 配置。注意我在ReviewerAgent中設置了一個檢查邏輯它既要判斷輸出內容是否滿足“包含核心關鍵詞”“有數據支撐”“無事實性錯誤”還要控制是否needs_revision。register_agent(reviewer) class ReviewerAgent(BaseAgent): async def run(self, task: Task, context): writer_output context.results[writer][output] # 這里調用 LLM 進行質量評估并返回 json: {passed: bool, comments: [...]} evaluation await self.llm.chat(...) if not evaluation[passed]: return AgentResult(statusneeds_revision, outputevaluation) return AgentResult(statussuccess, outputevaluation)運行時只需要在主入口里加載配置并調度async def run_flow(user_requirement: str): task Task(iduuid4().hex, typecontent, payload{requirement: user_requirement}) ctx Context(task_idtask.id) flow_config load_flow_config(flow.yaml) await dispatcher.dispatch(task, ctx, flow_config) return ctx.results整套系統(tǒng)跑起來后最開始的效果并不理想。調研員返回的資料要么太泛、要么太舊寫手寫出的稿子經常和產品賣點脫節(jié)。后來我逐步給每個 Agent 的 Prompt 里加入了“目標讀者鎖定”“競品關鍵詞映射表”這類約束情況才明顯改善。這里的重點不是代碼本身而是多 Agent 系統(tǒng)的成功與否很大程度取決于每個 Agent 的輸入輸出邊界定義得清不清楚這個在系統(tǒng)里是由 Prompt 和輸出格式共同保證的。4.4 調優(yōu)記錄如何把單次任務從 3 分鐘壓到 40 秒流水線跑通之后我發(fā)現(xiàn)單次任務耗時太長。最初版本是四個 Agent 串行執(zhí)行每個 Agent 調用一次大模型正常情況要 3 分鐘。我做了三件事終于把平均耗時壓到 40 秒上下第一步把調研員改成“內部并行檢索”一次觸發(fā) 3 個搜索請求等全部返回后一次性匯總。使用asyncio.gather幾乎是零成本提速。第二步給寫手配置一個更快的小模型承擔 70% 的常規(guī)內容生成審核員不滿意再切大模型。用“高中低配模型分層”的方式整體費用和速度最均衡。第三步減少串聯(lián)等待。讓審核員在檢查時就生成一份“修改建議清單”writer 接收到清單后直接定位到對應段落重寫而不是整篇重寫。效果比整篇重新生成要好得多。這一輪優(yōu)化給我的最大體會是多 Agent 系統(tǒng)不能一直加分堆邏輯它需要階段性做減法。把慢的環(huán)節(jié)并行掉把貴的環(huán)節(jié)便宜掉把重復的環(huán)節(jié)合并掉系統(tǒng)的價值才能體現(xiàn)出來。純堆 Agent 數量不一定帶來效率提升反而會讓調度成本、延遲和出錯概率一起變大。5. 老坑實錄死循環(huán)、并發(fā)限流、上下文爆掉5.1 死循環(huán)審核不通過時怎么辦第一次上線審核鏈路時我犯過一個挺蠢的錯誤審核不通過就直接讓寫手重寫寫手重寫之后審核又覺得不行無限循環(huán)下去。在真實測試中角色分配越明確這個循環(huán)越容易發(fā)生因為每個 Agent 都會堅定地堅持自己的判斷誰都不妥協(xié)。大模型不會累但你會看著賬單心疼。解決辦法就是前面提到過的“最多重試三次三次不過就轉人工”。在 Dispatcher 里增加一個簡單的全局計數每進入 writer 一次revision_count 1當達到上限時強制讓 reviewer 輸出“已通過但備注需人工確認”。同時在日志系統(tǒng)里記錄失敗原因便于分析瓶頸。總結成一句話所有帶反饋回路的流程必須設計終止條件否則系統(tǒng)遲早失控。5.2 并發(fā)限流頻繁 429 錯誤真正跑起來后第二個高頻問題是模型服務返回 429請求過多或者超時。多 Agent 并行化之后不僅沒解決這個問題反而讓情況更嚴重。我一開始天真地以為“并發(fā)越高越快”結果同時發(fā)出 10 個請求瞬間觸發(fā)了服務商的限流整個任務全部失敗重來比串行還慢。后來在llm_client里加了一個簡單的令牌桶限速器全局所有 Agent 共享保證每秒請求數不超過規(guī)定閾值。import asyncio import time class RateLimiter: def __init__(self, max_requests: int, per_seconds: float 1.0): self.max_requests max_requests self.period per_seconds self.tokens max_requests self.updated_at time.monotonic() async def acquire(self): while True: now time.monotonic() self.tokens min(self.max_requests, self.tokens (now - self.updated_at) / self.period * self.max_requests) self.updated_at now if self.tokens 1: self.tokens - 1 return await asyncio.sleep(0.05)這個限速器的粒度是整個進程級別同一時刻所有 Agent 共用一把鎖。如果是多機部署可以換成 Redis 分布式限流。這里要提醒的是具體閾值不能隨便拍腦袋我用的是“先設置低值、壓測后逐步上調”的方法找到一個既能跑得快又不觸發(fā) 429 的安全區(qū)間。5.3 上下文爆掉長文檔與多 Agent 記憶第三個高頻問題是 Context 無限膨脹。最初設計 Context 時以為把數據都放內存字典里就沒事結果跑了幾輪長文寫作任務內存上漲很快而且單次任務的 token 消耗也高得驚人。解決策略分兩層。短期上下文這邊我給每次寫入 Context 的內容設置了“摘要化”約束研究員輸出的內容必須壓縮到 500 字以內寫手寫作時讀到的是精煉摘要而不是海量原文。長期記憶這邊用向量庫存儲“任務級別的歷史結論”當下次遇到相近任務時只把最相關的 3 到 5 條歷史摘要注入到 Context 中而不是把所有歷史記錄都帶進去。順手還解決了一個容易被忽視的問題任務在傳遞長文檔時不應該傳文檔全文而是傳文檔 ID其他 Agent 需要時再按 ID 去文檔庫加載。這個設計初看只是省 token實際上還附帶解決了緩存一致性問題避免每個 Agent 各拿一份并各自腦補內容。5.4 工具返回格式不一致接口協(xié)議的坑最后還有一個細節(jié)工具調用多了以后“什么格式的返回結果”會成為災難。搜索工具返回的是數組數據庫查詢工具返回的是列表文件讀取工具返回的是字符串而 Agent 的 Prompt 里要求它統(tǒng)一處理這些數據模型很容易被格式帶偏經常出現(xiàn)“列表里套 JSON 字符串再套 Markdown”這種奇奇怪怪的嵌套。我后來規(guī)定所有工具輸出必須是統(tǒng)一的ToolResult結構內容字段是一個字符串但允許包含 Markdown結構化數據字段是一個 JSON 對象原始數據字段是數組或字典。Agent 的 Prompt 一律只讀取這兩種固定字段其他字段不做解析。這個約束極大降低了模型的解析錯誤率也讓調試日志變得清晰。如果到后期工具數量繼續(xù)增長建議順便引入 JSON Schema 做校驗而不是靠“約定”來約束。6. 后續(xù)擴展把模型層做實、把部署做穩(wěn)6.1 部署形態(tài)Docker Compose 一鍵啟動服務單體腳本跑通只是第一步實際要落地到一個團隊使用部署形態(tài)必須穩(wěn)定。我的做法是把主服務、Redis、日志系統(tǒng)用 Docker Compose 打包在一起同時把配置和 Prompt 模板掛載到外部目錄這樣改 Prompt 不用重新構建鏡像。# docker-compose.yml services: engine: build: . ports: - 8080:8080 volumes: - ./config:/app/config:ro - ./prompts:/app/prompts:ro environment: - LOG_LEVELINFO depends_on: - redis redis: image: redis:7-alpine volumes: - redis-data:/data這里有一個實戰(zhàn)建議Docker 容器里的 Python 進程要多開幾個 worker因為大模型接口的等待時間很長單 worker 很容易并發(fā)撐不住。我用的是 multiprocessing asyncio 的組合主進程收任務子進程跑 Agent 循環(huán)Redis 做任務隊列。這個組合方案雖然有些老土但勝在穩(wěn)定幾乎不需要額外維護。6.2 Agent 的可觀測性日志與鏈路追蹤多 Agent 系統(tǒng)的可觀測性比單 Agent 重要得多。原因很簡單單 Agent 出問題復盤時看一個輸入輸出就夠了多 Agent 出問題你可能需要看十幾個環(huán)節(jié)的輸入輸出如果沒有一個統(tǒng)一的 trace_id問題幾乎沒法排查。agency-agents 的做法是在 Task 創(chuàng)建時生成一個trace_id貫穿所有后續(xù)調用和日志每個 Agent 的日志輸出都帶上自己的名稱和 trace_id。日志格式直接用 JSON方便導入日志分析平臺{trace_id: 3f2a..., agent: writer, event: start, ts: 1720000000} {trace_id: 3f2a..., agent: writer, event: llm_call, model: gpt-4o-mini, latency_ms: 2100}千萬別小看這一招。有一次調研員總是返回過時數據我排查了半天沒有頭緒后來把搜索工具調用的 trace_id 對應日志一查發(fā)現(xiàn)是工具傳參時少了時間范圍限制導致模型默認用當前時間之前的所有結果。沒有鏈路日志這種跨組件問題可能要折騰好幾天。6.3 模塊化方向插件、協(xié)議與市場最后聊聊這套系統(tǒng)的未來擴展方向。我做這套框架時給 Agent 定義的接口足夠簡潔也意味著它可以很容易演變成一個半開放的插件系統(tǒng)。每種 Agent 可以對應一個 Python 包包內自帶配置模板和 Prompt 模板注冊中心負責加載。更進一步可以把 Agent 的輸入輸出協(xié)議標準化成 JSON Schema這樣不同團隊開發(fā)的 Agent 可以互相兼容整個“數字代理機構”就有機會變成一個生態(tài)。我目前的想法是往“協(xié)議優(yōu)先”的方向演進完全不管具體實現(xiàn)是什么只約好消息格式和接口契約A 團隊寫的調研員可以被 B 團隊的項目直接使用。當然這也帶來新的問題比如能力的可靠性評價、惡意 Agent 的隔離、消息的權限控制等等。不過這些屬于平臺化階段的事情當前階段最重要的是把單機的 Agent 編排邏輯磨得足夠穩(wěn)?；氐轿易约哼@套“agency-agents”框架至少有一點我確定用“代理機構”的思維去設計多 Agent 系統(tǒng)遠比給單個 Agent 堆砌更多 Prompt 要健康得多。如今每當我接到一個新任務第一反應不再是“我該怎么寫一個完美的提示詞”而是“這需要一個什么樣的團隊團隊里每個人干什么誰先誰后哪里需要檢查復核”。這種思維轉變大概就是這個項目給我留下的最實在的收獲。以后如果再投入時間我會優(yōu)先把 Agent 的“身份記憶”和“長期工作習慣”做深讓每個 Agent 在多次任務中越用越順手而不僅僅是完成一次孤立的工作。