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