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