給多智能體流水線裝上 if-else 和 whileAgno Workflow 工作流編排入門【免費(fèi)下載鏈接】agnoBuild, run, and manage agent platforms.項(xiàng)目地址: https://gitcode.com/GitHub_Trending/ag/agno想讓研究 Agent 查完資料后只有內(nèi)容涉及具體數(shù)據(jù)時(shí)才調(diào)起事實(shí)核查 Agent否則直接進(jìn)寫(xiě)作——這個(gè) if-else 在代碼里到底怎么寫(xiě)Agno Workflow 就是為此而生的一個(gè)用普通 Python 代碼描述多智能體流水線的引擎條件、循環(huán)、并行、路由全是可編程的結(jié)構(gòu)而不是畫(huà)在流程圖上的虛線。 把任務(wù)串成一條線順序步驟怎么跑Workflow 接收輸入后按列表順序逐個(gè)執(zhí)行 steps 里的成員。Step 的身份很寬松Agent、Team、普通 Python 函數(shù)都能直接入列函數(shù)只需讀 StepInput、返回 StepOutput就能和 Agent 平起平坐地參與工作流編排。銜接靠的是 previous_step_content上一步產(chǎn)出的內(nèi)容自動(dòng)流入下一步研究 Agent 寫(xiě)下的素材不用手動(dòng)傳遞寫(xiě)作 Agent 直接就能讀到。最小例子看 sequence_of_steps.py研究→寫(xiě)作兩步就是一條完整的 Agno 工作流引擎實(shí)現(xiàn)本身在 workflow.py。 給流水線加 if-elseCondition 與 Router 的分工Condition 像是流水線上的門衛(wèi)手里攥著 evaluator 這把尺返回 True內(nèi)部 steps 照常執(zhí)行返回 False整段被跳過(guò)流程直接走到下一站。經(jīng)典用法是事實(shí)核查門禁——摘要里出現(xiàn)數(shù)據(jù)顯示百分比這類事實(shí)性措辭才放行事實(shí)核查 Agent否則省掉這次調(diào)用參考 condition_basic.py。想給攔截時(shí)留個(gè)后路就配上 else_steps不滿足條件時(shí)改走另一小段。Router 更像分揀臺(tái)給它一組 choices 候選步驟selector 函數(shù)按輸入內(nèi)容動(dòng)態(tài)挑出要走的一條或多條路。比如主題命中 tech、AI、startup 就派 HackerNews 研究員否則交給通用網(wǎng)頁(yè)研究員兩條路最后都匯回發(fā)布步驟見(jiàn) router_basic.py。兩者別混Condition 回答做不做是二值判斷Router 回答派誰(shuí)去做是多路分發(fā)。流程需要攔截用前者需要分流用后者。 結(jié)果不夠好就再來(lái)一遍Agno 循環(huán)執(zhí)行Agno Workflow 循環(huán)結(jié)構(gòu)steps 反復(fù)執(zhí)行end_condition 決定退出時(shí)機(jī)單次生成的質(zhì)量看運(yùn)氣Loop 把這件事交給重試。它包一組步驟反復(fù)執(zhí)行每輪結(jié)束后調(diào)用 end_condition返回 True 就退出返回 False 就接著下一輪。例子里兩個(gè)研究步驟被圈進(jìn) Loop評(píng)估邏輯是輸出不足 200 字符就重跑夠長(zhǎng)了才放行給創(chuàng)作環(huán)節(jié)見(jiàn) loop_basic.py。光有 end_condition 還不夠穩(wěn)max_iterations 是配套的保險(xiǎn)絲評(píng)估函數(shù)萬(wàn)一失靈最多跑 3 輪也會(huì)強(qiáng)制收手。兩個(gè)參數(shù)配合才是質(zhì)量與成本的雙保險(xiǎn)。還有一個(gè)累積式變體開(kāi)啟迭代累積后每輪輸出會(huì)喂給下一輪當(dāng)輸入研究?jī)?nèi)容逐輪疊加而不是每輪從頭重來(lái)。? 無(wú)依賴的任務(wù)一起跑Agno 并行工作流兩個(gè)互不相干的研究任務(wù)串行著跑純屬浪費(fèi)。Parallel 同時(shí)發(fā)起它收到的所有步驟等全部完成后自動(dòng)合并結(jié)果再交給下游。前提是分支之間沒(méi)有數(shù)據(jù)依賴A 分支的輸入不能來(lái)自 B 分支的產(chǎn)出否則還是老老實(shí)實(shí)排成順序。parallel_basic.py 里研究 HackerNews和研究網(wǎng)頁(yè)并行跑完合并后的素材一次性交給寫(xiě)作和審閱。這套結(jié)構(gòu)還能繼續(xù)疊加控制邏輯分支內(nèi)部可以塞 Condition比如只讓某一類輸入觸發(fā)某個(gè)分支Loop 里也可以放 Parallel失敗的路徑整組重來(lái)組合玩法參考 parallel_with_condition.py。?? 少寫(xiě)代碼的兩個(gè)捷徑CEL 表達(dá)式與全局叫停手寫(xiě) evaluator 函數(shù)有點(diǎn)啰嗦時(shí)CEL 表達(dá)式可以直接頂替一行 evaluatorinput.contains(urgent) 就能完成輸入含 urgent 走緊急通道的判斷框架識(shí)別出這是表達(dá)式而非 Python 可調(diào)用對(duì)象自行解析執(zhí)行。表達(dá)式里能引用 input、previous_step、session_state 等上下文條件、循環(huán)退出、路由選擇統(tǒng)統(tǒng)可以聲明式配置。規(guī)則越簡(jiǎn)單越適合 CEL邏輯復(fù)雜了再換回函數(shù)。示例在 cel_basic.py需先 pip install cel-python。另一個(gè)捷徑是提前終止步驟返回 StepOutput 時(shí)把 stop 置 True剩余步驟全部跳過(guò)工作流以 Early Stopped 收尾。它和 Condition 跳過(guò)是兩碼事——前者叫停整條線后者只是繞過(guò)一段路。early_stopping 目錄里有安全掃描攔截部署的例子掃出 VULNERABLE 就直接收工?？刂平Y(jié)構(gòu)適用場(chǎng)景關(guān)鍵參數(shù)Steps順序接力stepsCondition二值攔截evaluator、else_stepsRouter多路分流choices、selectorLoop重試打磨end_condition、max_iterationsParallel無(wú)依賴并發(fā)steps先跑通 01_basic_workflows/ 里的順序步驟再往線上加 Condition、Loop、Parallel對(duì)應(yīng)子目錄分別是 02_conditional_execution/、03_loop_execution/、04_parallel_execution/。四種結(jié)構(gòu)疊加完畢你的多智能體流水線就有了剎車和方向盤——剩下的只是決定每一步派誰(shuí)上場(chǎng)?！久赓M(fèi)下載鏈接】agnoBuild, run, and manage agent platforms.項(xiàng)目地址: https://gitcode.com/GitHub_Trending/ag/agno創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考