AI Agent工作流手冊(cè)：5種常見(jiàn)模式實(shí)現(xiàn)與應(yīng)用，助力生產(chǎn)環(huán)境穩(wěn)定

很多人認(rèn)為使用AI Agent就是直接扔個(gè)提示詞過(guò)去，然后等結(jié)果。做實(shí)驗(yàn)這樣是沒(méi)問(wèn)題的，但要是想在生產(chǎn)環(huán)境穩(wěn)定輸出高質(zhì)量結(jié)果，這套玩法就不行了。

核心問(wèn)題是這種隨意的提示方式根本擴(kuò)展不了。你會(huì)發(fā)現(xiàn)輸出結(jié)果亂七八糟，質(zhì)量完全不可控，還浪費(fèi)計(jì)算資源。

真正有效的做法是設(shè)計(jì)結(jié)構(gòu)化的Agent工作流。

那些做得好的團(tuán)隊(duì)從來(lái)不指望一個(gè)提示詞解決所有問(wèn)題。他們會(huì)把復(fù)雜任務(wù)拆解成步驟，根據(jù)不同輸入選擇合適的模型，然后持續(xù)驗(yàn)證輸出質(zhì)量，直到結(jié)果達(dá)標(biāo)。

本文會(huì)詳細(xì)介紹5種常用的的Agent工作流模式，每種都有完整的實(shí)現(xiàn)代碼和使用場(chǎng)景分析。看完你就知道每種模式解決什么問(wèn)題，什么時(shí)候用，以及為什么能產(chǎn)生更好的效果。

模式一：串行鏈?zhǔn)教幚?/p>

鏈?zhǔn)教幚淼暮诵乃悸肥前岩粋€(gè)LLM調(diào)用的輸出直接作為下一個(gè)調(diào)用的輸入。比起把所有邏輯塞進(jìn)一個(gè)巨大的提示詞，拆分成小步驟要靠譜得多。

道理很簡(jiǎn)單：步驟越小，出錯(cuò)的概率越低。鏈?zhǔn)教幚淼扔诮o模型提供了明確的推理路徑，而不是讓它自己瞎猜。

如果不用鏈?zhǔn)教幚恚憧赡軙?huì)經(jīng)常會(huì)遇到輸出冗長(zhǎng)混亂、前后邏輯不一致、錯(cuò)誤率偏高的問(wèn)題。有了鏈?zhǔn)教幚恚恳徊蕉伎梢詥为?dú)檢查，整個(gè)流程的可控性會(huì)大幅提升。

from typing import List from helpers import run_llm def serial_chain_workflow(input_query: str, prompt_chain : List[str]) -> List[str]: """運(yùn)行一系列LLM調(diào)用來(lái)處理`input_query`， 使用`prompt_chain`中指定的提示詞列表。 """ response_chain = [] response = input_query for i, prompt in enumerate(prompt_chain): print(f"Step {i+1}") response = run_llm(f"{prompt}\nInput:\n{response}", model='meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo') response_chain.append(response) print(f"{response}\n") return response_chain # 示例 question = "Sally earns $12 an hour for babysitting. Yesterday, she just did 50 minutes of babysitting. How much did she earn?" prompt_chain = ["""Given the math problem, ONLY extract any relevant numerical information and how it can be used.""", """Given the numberical information extracted, ONLY express the steps you would take to solve the problem.""", """Given the steps, express the final answer to the problem."""] responses = serial_chain_workflow(question, prompt_chain)

模式二：智能路由

路由系統(tǒng)的作用是決定不同類型的輸入應(yīng)該交給哪個(gè)模型處理。

現(xiàn)實(shí)情況是，不是每個(gè)查詢都需要?jiǎng)佑媚阕顝?qiáng)大、最昂貴的模型。簡(jiǎn)單任務(wù)用輕量級(jí)模型就夠了，復(fù)雜任務(wù)才需要重型武器。路由機(jī)制確保資源分配的合理性。

沒(méi)有路由的話，要么在簡(jiǎn)單任務(wù)上浪費(fèi)資源，要么用不合適的模型處理復(fù)雜問(wèn)題導(dǎo)致效果很差。

實(shí)現(xiàn)路由需要先定義輸入的分類標(biāo)準(zhǔn)，比如簡(jiǎn)單查詢、復(fù)雜推理、受限領(lǐng)域等，然后為每個(gè)類別指定最適合的模型或處理流程。這樣做的好處是成本更低、響應(yīng)更快、質(zhì)量更穩(wěn)定，因?yàn)槊糠N任務(wù)都有專門(mén)的工具來(lái)處理。

from pydantic import BaseModel, Field from typing import Literal, Dict from helpers import run_llm, JSON_llm def router_workflow(input_query: str, routes: Dict[str, str]) -> str: """給定一個(gè)`input_query`和包含每個(gè)選項(xiàng)和詳細(xì)信息的`routes`字典。 為任務(wù)選擇最佳模型并返回模型的響應(yīng)。 """ ROUTER_PROMPT = """Given a user prompt/query: {user_query}, select the best option out of the following routes: {routes}. Answer only in JSON format.""" # 從路由字典創(chuàng)建模式 class Schema(BaseModel): route: Literal[tuple(routes.keys())] reason: str = Field( description="Short one-liner explanation why this route was selected for the task in the prompt/query." ) # 調(diào)用LLM選擇路由 selected_route = JSON_llm( ROUTER_PROMPT.format(user_query=input_query, routes=routes), Schema ) print( f"Selected route:{selected_route['route']}\nReason: {selected_route['reason']}\n" ) # 在選定的路由上使用LLM。 # 也可以為每個(gè)路由使用不同的提示詞。 response = run_llm(user_prompt=input_query, model=selected_route["route"]) print(f"Response: {response}\n") return response prompt_list = [ "Produce python snippet to check to see if a number is prime or not.", "Plan and provide a short itenary for a 2 week vacation in Europe.", "Write a short story about a dragon and a knight.", ] model_routes = { "Qwen/Qwen2.5-Coder-32B-Instruct": "Best model choice for code generation tasks.", "Gryphe/MythoMax-L2-13b": "Best model choice for story-telling, role-playing and fantasy tasks.", "Qwen/QwQ-32B-Preview": "Best model for reasoning, planning and multi-step tasks", } for i, prompt in enumerate(prompt_list): print(f"Task {i+1}: {prompt}\n") print(20 * "==") router_workflow(prompt, model_routes)

模式三：并行處理

大部分人習(xí)慣讓LLM一個(gè)任務(wù)一個(gè)任務(wù)地處理。但如果任務(wù)之間相互獨(dú)立，完全可以并行執(zhí)行然后合并結(jié)果，這樣既節(jié)省時(shí)間又能提升輸出質(zhì)量。

并行處理的思路是把大任務(wù)分解成可以同時(shí)進(jìn)行的小任務(wù)，等所有部分都完成后再把結(jié)果整合起來(lái)。

比如做代碼審查的時(shí)候，可以讓一個(gè)模型專門(mén)檢查安全問(wèn)題，另一個(gè)關(guān)注性能優(yōu)化，第三個(gè)負(fù)責(zé)代碼可讀性，最后把所有反饋合并成完整的審查報(bào)告。文檔分析也是類似的思路：把長(zhǎng)報(bào)告按章節(jié)拆分，每個(gè)部分單獨(dú)總結(jié)，再合并成整體摘要。文本分析任務(wù)中，情感分析、實(shí)體提取、偏見(jiàn)檢測(cè)可以完全并行進(jìn)行。

不用并行處理的話，不僅速度慢，還容易讓單個(gè)模型負(fù)擔(dān)過(guò)重，導(dǎo)致輸出混亂或者前后不一致。并行方式讓每個(gè)模型專注于自己擅長(zhǎng)的部分，最終結(jié)果會(huì)更準(zhǔn)確，也更容易維護(hù)。

import asyncio from typing import List from helpers import run_llm, run_llm_parallel async def parallel_workflow(prompt : str, proposer_models : List[str], aggregator_model : str, aggregator_prompt: str): """運(yùn)行并行LLM調(diào)用鏈來(lái)處理`input_query`， 使用`models`中指定的模型列表。 返回最終聚合器模型的輸出。 """ # 從建議器模型收集中間響應(yīng) proposed_responses = await asyncio.gather(*[run_llm_parallel(prompt, model) for model in proposer_models]) # 使用聚合器模型聚合響應(yīng) final_output = run_llm(user_prompt=prompt, model=aggregator_model, system_prompt=aggregator_prompt + "\n" + "\n".join(f"{i+1}. {str(element)}" for i, element in enumerate(proposed_responses) )) return final_output, proposed_responses reference_models = [ "microsoft/WizardLM-2-8x22B", "Qwen/Qwen2.5-72B-Instruct-Turbo", "google/gemma-2-27b-it", "meta-llama/Llama-3.3-70B-Instruct-Turbo", ] user_prompt = """Jenna and her mother picked some apples from their apple farm. Jenna picked half as many apples as her mom. If her mom got 20 apples, how many apples did they both pick?""" aggregator_model = "deepseek-ai/DeepSeek-V3" aggregator_system_prompt = """You have been provided with a set of responses from various open-source models to the latest user query. Your task is to synthesize these responses into a single, high-quality response. It is crucial to critically evaluate the information provided in these responses, recognizing that some of it may be biased or incorrect. Your response should not simply replicate the given answers but should offer a refined, accurate, and comprehensive reply to the instruction. Ensure your response is well-structured, coherent, and adheres to the highest standards of accuracy and reliability. Responses from models:""" async def main(): answer, intermediate_reponses = await parallel_workflow(prompt = user_prompt, proposer_models = reference_models, aggregator_model = aggregator_model, aggregator_prompt = aggregator_system_prompt) for i, response in enumerate(intermediate_reponses): print(f"Intermetidate Response {i+1}:\n\n{response}\n") print(f"Final Answer: {answer}\n")

模式四：編排器-工作器架構(gòu)

這種模式的特點(diǎn)是用一個(gè)編排器模型來(lái)規(guī)劃整個(gè)任務(wù)，然后把具體的子任務(wù)分配給不同的工作器模型執(zhí)行。

編排器的職責(zé)是分析任務(wù)需求，決定執(zhí)行順序，你不需要事先設(shè)計(jì)好完整的工作流程。工作器模型各自處理分配到的任務(wù)，編排器負(fù)責(zé)把所有輸出整合成最終結(jié)果。

這種架構(gòu)在很多場(chǎng)景下都很實(shí)用。寫(xiě)博客文章的時(shí)候，編排器可以把任務(wù)拆分成標(biāo)題設(shè)計(jì)、內(nèi)容大綱、具體章節(jié)寫(xiě)作，然后讓專門(mén)的工作器處理每個(gè)部分，最后組裝成完整文章。開(kāi)發(fā)程序時(shí)，編排器負(fù)責(zé)分解成環(huán)境配置、核心功能實(shí)現(xiàn)、測(cè)試用例編寫(xiě)等子任務(wù)，不同的工作器生成對(duì)應(yīng)的代碼片段。數(shù)據(jù)分析報(bào)告也是類似思路：編排器識(shí)別出需要數(shù)據(jù)概覽、關(guān)鍵指標(biāo)計(jì)算、趨勢(shì)分析等部分，工作器分別生成內(nèi)容，編排器最后整合成完整報(bào)告。

這種方式的好處是減少了人工規(guī)劃的工作量，同時(shí)保證復(fù)雜任務(wù)的有序進(jìn)行。編排器處理任務(wù)管理，每個(gè)工作器專注于自己的專業(yè)領(lǐng)域，整個(gè)流程既有條理又有效率。

import asyncio import json from pydantic import BaseModel, Field from typing import Literal, List from helpers import run_llm_parallel, JSON_llm ORCHESTRATOR_PROMPT = """ 分析這個(gè)任務(wù)并將其分解為2-3種不同的方法： 任務(wù): {task} 提供分析： 解釋你對(duì)任務(wù)的理解以及哪些變化會(huì)有價(jià)值。 關(guān)注每種方法如何服務(wù)于任務(wù)的不同方面。 除了分析之外，提供2-3種處理任務(wù)的方法，每種都有簡(jiǎn)要描述： 正式風(fēng)格: 技術(shù)性和精確地寫(xiě)作，專注于詳細(xì)規(guī)范 對(duì)話風(fēng)格: 以友好和引人入勝的方式寫(xiě)作，與讀者建立聯(lián)系 混合風(fēng)格: 講述包含技術(shù)細(xì)節(jié)的故事，將情感元素與規(guī)范相結(jié)合 僅返回JSON輸出。 """ WORKER_PROMPT = """ 基于以下內(nèi)容生成內(nèi)容： 任務(wù): {original_task} 風(fēng)格: {task_type} 指導(dǎo)原則: {task_description} 僅返回你的響應(yīng)： [你的內(nèi)容在這里，保持指定的風(fēng)格并完全滿足要求。] """ task = """為新的環(huán)保水瓶寫(xiě)一個(gè)產(chǎn)品描述。 目標(biāo)受眾是有環(huán)保意識(shí)的千禧一代，關(guān)鍵產(chǎn)品特性是：無(wú)塑料、保溫、終身保修 """ class Task(BaseModel): type: Literal["formal", "conversational", "hybrid"] description: str class TaskList(BaseModel): analysis: str tasks: List[Task] = Field(..., default_factory=list) async def orchestrator_workflow(task : str, orchestrator_prompt : str, worker_prompt : str): """使用編排器模型將任務(wù)分解為子任務(wù)，然后使用工作器模型生成并返回響應(yīng)。""" # 使用編排器模型將任務(wù)分解為子任務(wù) orchestrator_response = JSON_llm(orchestrator_prompt.format(task=task), schema=TaskList) # 解析編排器響應(yīng) analysis = orchestrator_response["analysis"] tasks= orchestrator_response["tasks"] print("\n=== ORCHESTRATOR OUTPUT ===") print(f"\nANALYSIS:\n{analysis}") print(f"\nTASKS:\n{json.dumps(tasks, indent=2)}") worker_model = ["meta-llama/Llama-3.3-70B-Instruct-Turbo"]*len(tasks) # 從工作器模型收集中間響應(yīng) return tasks , await asyncio.gather(*[run_llm_parallel(user_prompt=worker_prompt.format(original_task=task, task_type=task_info['type'], task_description=task_info['description']), model=model) for task_info, model in zip(tasks,worker_model)]) async def main(): task = """為新的環(huán)保水瓶寫(xiě)一個(gè)產(chǎn)品描述。 目標(biāo)受眾是有環(huán)保意識(shí)的千禧一代，關(guān)鍵產(chǎn)品特性是：無(wú)塑料、保溫、終身保修 """ tasks, worker_resp = await orchestrator_workflow(task, orchestrator_prompt=ORCHESTRATOR_PROMPT, worker_prompt=WORKER_PROMPT) for task_info, response in zip(tasks, worker_resp): print(f"\n=== WORKER RESULT ({task_info['type']}) ===\n{response}\n") asyncio.run(main())

模式五：評(píng)估器-優(yōu)化器循環(huán)

這種模式的核心是通過(guò)反饋循環(huán)來(lái)持續(xù)改進(jìn)輸出質(zhì)量。

具體機(jī)制是一個(gè)模型負(fù)責(zé)生成內(nèi)容，另一個(gè)評(píng)估器模型按照預(yù)設(shè)的標(biāo)準(zhǔn)檢查輸出質(zhì)量。如果沒(méi)達(dá)標(biāo)，生成器根據(jù)反饋進(jìn)行修改，評(píng)估器再次檢查，這個(gè)過(guò)程一直重復(fù)到輸出滿足要求為止。

在代碼生成場(chǎng)景中，生成器寫(xiě)出代碼后，評(píng)估器會(huì)檢查語(yǔ)法正確性、算法效率、代碼風(fēng)格等方面，發(fā)現(xiàn)問(wèn)題就要求重寫(xiě)。營(yíng)銷文案也是類似流程：生成器起草內(nèi)容，評(píng)估器檢查字?jǐn)?shù)限制、語(yǔ)言風(fēng)格、信息準(zhǔn)確性，不合格就繼續(xù)改。數(shù)據(jù)報(bào)告的制作過(guò)程中，生成器產(chǎn)出分析結(jié)果，評(píng)估器驗(yàn)證數(shù)據(jù)完整性和結(jié)論的邏輯性。

如果沒(méi)有這套評(píng)估優(yōu)化機(jī)制，輸出質(zhì)量會(huì)很不穩(wěn)定，需要大量人工檢查和修正。有了評(píng)估器-優(yōu)化器循環(huán)，可以自動(dòng)保證結(jié)果符合預(yù)期標(biāo)準(zhǔn)，減少重復(fù)的手工干預(yù)。

from pydantic import BaseModel from typing import Literal from helpers import run_llm, JSON_llm task = """ 實(shí)現(xiàn)一個(gè)棧，包含： 1. push(x) 2. pop() 3. getMin() 所有操作都應(yīng)該是O(1)。 """ GENERATOR_PROMPT = """ 你的目標(biāo)是基于<用戶輸入>完成任務(wù)。如果有來(lái)自你之前生成的反饋， 你應(yīng)該反思它們來(lái)改進(jìn)你的解決方案 以下列格式簡(jiǎn)潔地輸出你的答案： 思路： [你對(duì)任務(wù)和反饋的理解以及你計(jì)劃如何改進(jìn)] 響應(yīng)： [你的代碼實(shí)現(xiàn)在這里] """ def generate(task: str, generator_prompt: str, context: str = "") -> tuple[str, str]: """基于反饋生成和改進(jìn)解決方案。""" full_prompt = f"{generator_prompt}\n{context}\n任務(wù): {task}" if context else f"{generator_prompt}\n任務(wù): {task}" response = run_llm(full_prompt, model="Qwen/Qwen2.5-Coder-32B-Instruct") print("\n## Generation start") print(f"Output:\n{response}\n") return response EVALUATOR_PROMPT = """ 評(píng)估以下代碼實(shí)現(xiàn)的： 1. 代碼正確性 2. 時(shí)間復(fù)雜度 3. 風(fēng)格和最佳實(shí)踐 你應(yīng)該只進(jìn)行評(píng)估，不要嘗試解決任務(wù)。 只有當(dāng)所有標(biāo)準(zhǔn)都得到滿足且你沒(méi)有進(jìn)一步改進(jìn)建議時(shí)，才輸出"PASS"。 如果有需要改進(jìn)的地方，請(qǐng)?zhí)峁┰敿?xì)反饋。你應(yīng)該指出需要改進(jìn)什么以及為什么。 只輸出JSON。 """ def evaluate(task : str, evaluator_prompt : str, generated_content: str, schema) -> tuple[str, str]: """評(píng)估解決方案是否滿足要求。""" full_prompt = f"{evaluator_prompt}\n原始任務(wù): {task}\n要評(píng)估的內(nèi)容: {generated_content}" # 構(gòu)建評(píng)估模式 class Evaluation(BaseModel): evaluation: Literal["PASS", "NEEDS_IMPROVEMENT", "FAIL"] feedback: str response = JSON_llm(full_prompt, Evaluation) evaluation = response["evaluation"] feedback = response["feedback"] print("## Evaluation start") print(f"Status: {evaluation}") print(f"Feedback: {feedback}") return evaluation, feedback def loop_workflow(task: str, evaluator_prompt: str, generator_prompt: str) -> tuple[str, list[dict]]: """持續(xù)生成和評(píng)估，直到評(píng)估器通過(guò)最后生成的響應(yīng)。""" # 存儲(chǔ)生成器的先前響應(yīng) memory = [] # 生成初始響應(yīng) response = generate(task, generator_prompt) memory.append(response) # 當(dāng)生成的響應(yīng)沒(méi)有通過(guò)時(shí)，繼續(xù)生成和評(píng)估 while True: evaluation, feedback = evaluate(task, evaluator_prompt, response) # 終止條件 if evaluation == "PASS": return response # 將當(dāng)前響應(yīng)和反饋添加到上下文中并生成新響應(yīng) context = "\n".join([ "Previous attempts:", *[f"- {m}" for m in memory], f"\nFeedback: {feedback}" ]) response = generate(task, generator_prompt, context) memory.append(response) loop_workflow(task, EVALUATOR_PROMPT, GENERATOR_PROMPT)

總結(jié)

這些結(jié)構(gòu)化工作流徹底改變了LLM的使用方式。

不再是隨便拋個(gè)提示詞然后碰運(yùn)氣，而是有章法地分解任務(wù)、合理分配模型資源、并行處理獨(dú)立子任務(wù)、智能編排復(fù)雜流程，再通過(guò)評(píng)估循環(huán)保證輸出質(zhì)量。

每種模式都有明確的適用場(chǎng)景，組合使用能讓你更高效地處理各種復(fù)雜任務(wù)。可以先從一個(gè)模式開(kāi)始熟悉，掌握之后再逐步引入其他模式。

當(dāng)你把路由、編排、并行處理、評(píng)估優(yōu)化這些機(jī)制組合起來(lái)使用時(shí)，就徹底告別了那種混亂、不可預(yù)測(cè)的提示方式，轉(zhuǎn)而獲得穩(wěn)定、高質(zhì)量、可用于生產(chǎn)環(huán)境的輸出結(jié)果。長(zhǎng)期來(lái)看，這種方法不僅節(jié)省時(shí)間，更重要的是給你對(duì)AI系統(tǒng)的完全控制權(quán)，讓每次輸出都在預(yù)期范圍內(nèi)，從根本上解決了臨時(shí)提示方式帶來(lái)的各種問(wèn)題。

掌握這些工作流模式，你就能充分發(fā)揮AI的潛力，穩(wěn)定地獲得高質(zhì)量結(jié)果。

https://avoid.overfit.cn/post/c0be5e7283234053b03fc894fc7c7342

作者：Paolo Perrone