Python AI CrewAI工作流實戰：多角色Agent協作的5個核心模式

AI与大数据

當單Agent扛不住：多角色協作才是AI工作流的正確打開方式

上週一個做內容營運的朋友跟我吐槽：他用單個Agent做自動化選題分析，讓一個Agent同時負責資料抓取、趨勢分析、內容生成和排版——結果輸出一塌糊塗，分析報告裡混著營銷文案，資料圖表格式錯亂，Prompt越寫越長卻越來越失控。這不是個例。單Agent能力邊界有限、複雜任務分解困難、多角色職責劃分不清、協作流程混亂導致輸出品質崩塌——這四個痛點幾乎困擾著每一個嘗試AI工作流的開發者。

CrewAI的核心思路是「角色分工+流程編排」：每個Agent專注一件事，透過Sequential或Hierarchical流程串聯，用Tool擴展能力邊界。本文將從5個核心模式出發，帶你從零構建生產級CrewAI工作流。

核心概念速查

概念 說明 核心作用
CrewAI 多角色Agent協作框架 角色化Agent定義，流程化任務編排
Agent角色 具有特定職責和目標的智慧體 每個Agent專注單一領域，避免職責混亂
Task任務 Agent需要完成的具體工作 定義輸入輸出、依賴關係和預期結果
Crew團隊 Agent和Task的容器與編排器 管理Agent協作流程和任務執行
Tool工具 Agent可呼叫的外部能力 搜尋、程式碼執行、API呼叫等能力擴展
Sequential流程 順序執行模式 任務按定義順序依次執行，上游輸出傳遞給下游
Hierarchical流程 層級管理模式 Manager Agent分配任務，適合複雜決策場景

問題分析：多角色Agent協作的5大挑戰

挑戰1：角色定義與職責劃分。給Agent起個名字容易，定義清晰的職責邊界難。很多開發者把「資料分析Agent」寫成既抓資料又做分析還寫報告的全能選手，結果跟單Agent沒區別。好的角色定義應該是：一個Agent只做一件事，做到極致。

挑戰2：任務依賴與執行順序。任務A的輸出是任務B的輸入，但B又依賴任務C的結果——這種DAG依賴關係如果手動管理，很容易出現死鎖或資料斷裂。CrewAI的Sequential流程適合線性依賴，Hierarchical流程適合複雜依賴，但選錯流程模式會導致整個工作流卡住。

挑戰3：上下文傳遞與資訊共享。Agent之間如何傳遞中間結果？全量傳遞會導致Token爆炸，選擇性傳遞又可能丟失關鍵資訊。更棘手的是，當多個Agent並行工作時，如何保證它們看到一致的上下文視圖？

挑戰4：工具分配與衝突。搜尋工具給誰？程式碼執行工具給誰？如果兩個Agent都有搜尋工具，會不會重複呼叫浪費Token？工具分配不合理，輕則效率低下，重則產生矛盾結果。

挑戰5：輸出品質與一致性。每個Agent獨立生成內容，風格、格式、深度可能完全不同。最終拼接出來的報告，前言不搭後語。缺乏統一的品質標準和格式約束，多Agent協作反而不如單Agent。

模式1：CrewAI基礎Agent與Task定義

一切從定義Agent和Task開始。這是CrewAI最基礎但最關鍵的模式——角色定義的品質直接決定協作效果。

from crewai import Agent, Task, Crew, LLM

llm = LLM(model="gpt-4o", temperature=0.3)

researcher = Agent(
    role="市場研究分析師",
    goal="深入分析指定領域的市場趨勢和競爭格局",
    backstory="你是一位擁有10年經驗的市場研究分析師，擅長從海量資訊中提取關鍵洞察，你的分析報告曾被多家500強企業採用",
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

writer = Agent(
    role="技術內容撰寫專家",
    goal="將技術分析轉化為清晰易懂的專業文章",
    backstory="你是一位資深技術作家，曾在頂級科技媒體擔任主編，擅長將複雜技術概念用簡潔有力的語言表達",
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

researchTask = Task(
    description="分析{topic}領域的市場現狀，包括：1)主要參與者及其市場份額 2)技術發展趨勢 3)未來6個月的預測",
    expected_output="一份結構化的市場分析報告，包含資料支撐和趨勢判斷",
    agent=researcher,
)

writeTask = Task(
    description="基於市場分析報告，撰寫一篇面向技術決策者的深度文章，要求：1)核心觀點明確 2)資料引用準確 3)可讀性強",
    expected_output="一篇1500字左右的專業技術文章，Markdown格式",
    agent=writer,
)

關鍵原則：role決定視角，goal決定方向，backstory決定深度。backstory不是裝飾，它直接影響Agent的推理風格和輸出品質。一個有「10年經驗」backstory的Agent，輸出會比沒有backstory的Agent更專業、更有深度。

模式2：Sequential順序執行流程

Sequential是最直觀的流程模式：任務按定義順序依次執行，上游任務的輸出自動作為下游任務的上下文。

from crewai import Agent, Task, Crew, Process, LLM

llm = LLM(model="gpt-4o", temperature=0.3)

researcher = Agent(
    role="資料研究專員",
    goal="收集並整理指定主題的原始資料和關鍵資訊",
    backstory="你是一位嚴謹的資料研究專員，擅長從多源資料中提取可靠資訊",
    llm=llm,
)

analyst = Agent(
    role="資料分析專家",
    goal="對原始資料進行深度分析，發現趨勢和模式",
    backstory="你是一位資深資料分析師，擅長統計分析和趨勢預測",
    llm=llm,
)

editor = Agent(
    role="內容審核編輯",
    goal="審核分析報告的準確性和可讀性，確保輸出品質",
    backstory="你是一位嚴格的技術編輯，對資料準確性和邏輯一致性有極高要求",
    llm=llm,
)

researchTask = Task(
    description="收集關於{topic}的最新市場資料，包括市場規模、增長率、主要玩家",
    expected_output="結構化的原始資料摘要，包含資料來源",
    agent=researcher,
)

analysisTask = Task(
    description="基於研究資料，進行趨勢分析和競爭格局評估",
    expected_output="包含圖表描述和關鍵發現的分析報告",
    agent=analyst,
)

reviewTask = Task(
    description="審核分析報告，檢查資料準確性、邏輯一致性和可讀性，提出修改建議或確認發佈",
    expected_output="最終版報告，附帶審核意見",
    agent=editor,
)

crew = Crew(
    agents=[researcher, analyst, editor],
    tasks=[researchTask, analysisTask, reviewTask],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "2026年AI Agent框架"})
print(result)

Sequential流程的適用場景：線性依賴鏈——每個任務的輸出嚴格依賴前一個任務。不適合有並行子任務或複雜條件分支的場景。注意：如果中間任務失敗，後續任務不會執行，務必在Task中設定expected_output做品質把關。

模式3：Hierarchical層級管理流程

當任務依賴關係複雜、需要動態決策時，Hierarchical流程讓Manager Agent充當「專案經理」，自動分配任務和協調執行。

from crewai import Agent, Task, Crew, Process, LLM

llm = LLM(model="gpt-4o", temperature=0.2)

manager = Agent(
    role="專案經理",
    goal="協調團隊成員完成複雜專案，確保任務分配合理、輸出品質達標",
    backstory="你是一位經驗豐富的技術專案經理，擅長拆解複雜任務、合理分配資源、把控專案節奏",
    llm=llm,
    allow_delegation=True,
)

researcher = Agent(
    role="行業研究分析師",
    goal="深入分析行業趨勢和競爭格局",
    backstory="你是一位專注行業研究的分析師，擅長從宏觀視角把握趨勢",
    llm=llm,
)

technologist = Agent(
    role="技術架構師",
    goal="評估技術方案的可行性和創新性",
    backstory="你是一位全端技術架構師，對主流技術棧和新興技術有深入理解",
    llm=llm,
)

business_analyst = Agent(
    role="商業分析師",
    goal="評估商業價值和投資回報",
    backstory="你是一位資深商業分析師，擅長財務建模和商業可行性評估",
    llm=llm,
)

tasks = [
    Task(
        description="對{topic}進行全面的行業分析，包括市場規模、增長趨勢、競爭格局",
        expected_output="行業分析報告，包含市場規模資料和競爭格局圖",
        agent=researcher,
    ),
    Task(
        description="評估{topic}相關的核心技術方案，分析技術可行性和創新點",
        expected_output="技術評估報告，包含架構建議和風險分析",
        agent=technologist,
    ),
    Task(
        description="基於行業和技術分析，評估商業價值和投資回報，給出建議",
        expected_output="商業評估報告，包含ROI分析和投資建議",
        agent=business_analyst,
    ),
]

crew = Crew(
    agents=[researcher, technologist, business_analyst],
    tasks=tasks,
    process=Process.hierarchical,
    manager_agent=manager,
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "AI Agent基礎設施"})

Hierarchical流程的核心優勢：Manager Agent可以根據中間結果動態調整任務分配和執行順序。適用場景：任務間存在條件依賴、需要動態決策、團隊規模超過3人。注意：Manager本身消耗Token，小團隊簡單任務用Sequential更高效。

模式4：自訂Tool與Agent整合

CrewAI的Tool機制讓Agent突破純文字推理的限制，能夠搜尋網頁、執行程式碼、呼叫API。自訂Tool是連接Agent與外部系統的橋樑。

from crewai import Agent, Task, Crew, LLM
from crewai.tools import tool
from typing import Dict

@tool("search_market_data")
def searchMarketData(query: str) -> str:
    """搜尋市場資料，輸入搜尋關鍵詞，返回相關市場資料摘要"""
    import requests
    try:
        response = requests.get(
            "https://api.example.com/market/search",
            params={"q": query, "limit": 5},
            timeout=10,
        )
        response.raise_for_status()
        data = response.json()
        results = []
        for item in data.get("results", []):
            results.append(f"- {item['title']}: {item['summary']}")
        return "\n".join(results) if results else "未找到相關資料"
    except Exception as e:
        return f"搜尋失敗: {str(e)}"

@tool("calculate_metrics")
def calculateMetrics(data: str) -> str:
    """計算業務指標，輸入JSON格式的原始資料，返回計算結果"""
    import json
    try:
        parsed = json.loads(data)
        total = sum(item.get("value", 0) for item in parsed.get("items", []))
        avg = total / max(len(parsed.get("items", [])), 1)
        return json.dumps({"total": total, "average": round(avg, 2), "count": len(parsed.get("items", []))})
    except Exception as e:
        return f"計算失敗: {str(e)}"

llm = LLM(model="gpt-4o", temperature=0.3)

researcher = Agent(
    role="市場資料研究員",
    goal="透過搜尋工具獲取最新市場資料並整理分析",
    backstory="你是一位善於利用工具獲取資料的研究員，擅長從搜尋結果中提煉關鍵資訊",
    llm=llm,
    tools=[searchMarketData],
)

analyst = Agent(
    role="資料分析師",
    goal="對原始資料進行計算和深度分析",
    backstory="你是一位精通資料分析的專家，擅長透過計算工具快速得出關鍵指標",
    llm=llm,
    tools=[calculateMetrics],
)

researchTask = Task(
    description="使用搜尋工具獲取{topic}的最新市場資料",
    expected_output="市場資料摘要，包含資料來源",
    agent=researcher,
)

analysisTask = Task(
    description="使用計算工具對市場資料進行分析，計算關鍵指標",
    expected_output="包含關鍵指標和分析結論的報告",
    agent=analyst,
)

crew = Crew(
    agents=[researcher, analyst],
    tasks=[researchTask, analysisTask],
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "AI Agent市場"})

Tool設計原則：一個Tool只做一件事，函數名和docstring要清晰——LLM根據它們決定何時呼叫。返回值必須是字串，異常要在Tool內部捕獲並返回友好錯誤資訊，不要讓異常傳播到Agent層。

模式5：生產級CrewAI工作流（含監控）

生產環境需要考慮：錯誤處理、重試機制、執行監控、成本控制。這個模式將前面所有模式整合為一個可部署的生產級工作流。

import logging
import time
from datetime import datetime
from crewai import Agent, Task, Crew, Process, LLM
from crewai.tools import tool

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("crewai_production")

@tool("web_search")
def webSearch(query: str) -> str:
    """搜尋網際網路獲取最新資訊"""
    import requests
    try:
        response = requests.get(
            "https://api.search.service/v1/search",
            params={"q": query, "num": 5},
            headers={"Authorization": "Bearer YOUR_API_KEY"},
            timeout=15,
        )
        response.raise_for_status()
        return response.json().get("summary", "無搜尋結果")
    except requests.Timeout:
        return "搜尋超時，請稍後重試"
    except requests.RequestException as e:
        return f"搜尋錯誤: {str(e)}"

class CrewAIProductionWorkflow:
    def __init__(self):
        self.llm = LLM(model="gpt-4o", temperature=0.2, max_tokens=4096)
        self.executionLog = []
        self._setupAgents()
        self._setupTasks()
        self._setupCrew()

    def _setupAgents(self):
        self.researcher = Agent(
            role="首席研究分析師",
            goal="透過多源搜尋獲取全面、準確的市場資訊",
            backstory="你是一位擁有15年經驗的研究總監，擅長從海量資訊中篩選高價值洞察",
            llm=self.llm,
            tools=[webSearch],
            max_iter=5,
            verbose=True,
        )
        self.writer = Agent(
            role="高級內容策略師",
            goal="將研究結果轉化為高品質的專業內容",
            backstory="你是一位獲過獎的內容策略師，擅長將複雜資訊轉化為引人入勝的敘事",
            llm=self.llm,
            max_iter=3,
            verbose=True,
        )
        self.reviewer = Agent(
            role="品質審核專家",
            goal="確保輸出內容的準確性、完整性和專業性",
            backstory="你是一位以嚴格著稱的審核專家，絕不放過任何事實錯誤或邏輯漏洞",
            llm=self.llm,
            max_iter=2,
            verbose=True,
        )

    def _setupTasks(self):
        self.tasks = [
            Task(
                description="對{topic}進行全面的市場研究，使用搜尋工具獲取最新資料",
                expected_output="結構化市場研究報告，包含資料來源和關鍵發現",
                agent=self.researcher,
            ),
            Task(
                description="基於研究報告，撰寫面向決策者的深度分析文章",
                expected_output="2000字專業分析文章，Markdown格式，包含資料引用",
                agent=self.writer,
            ),
            Task(
                description="審核文章的事實準確性、邏輯一致性和專業深度",
                expected_output="審核通過的文章終稿，附審核意見",
                agent=self.reviewer,
            ),
        ]

    def _setupCrew(self):
        self.crew = Crew(
            agents=[self.researcher, self.writer, self.reviewer],
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True,
            max_rpm=10,
        )

    def run(self, topic: str) -> dict:
        startTime = time.time()
        logger.info(f"[{datetime.now().isoformat()}] 工作流啟動: {topic}")
        try:
            result = self.crew.kickoff(inputs={"topic": topic})
            duration = time.time() - startTime
            executionRecord = {
                "topic": topic,
                "status": "success",
                "duration": round(duration, 2),
                "timestamp": datetime.now().isoformat(),
                "result_length": len(str(result)),
            }
            self.executionLog.append(executionRecord)
            logger.info(f"工作流完成，耗時{duration:.2f}秒")
            return {"status": "success", "result": str(result), "metadata": executionRecord}
        except Exception as e:
            duration = time.time() - startTime
            errorRecord = {
                "topic": topic,
                "status": "failed",
                "error": str(e),
                "duration": round(duration, 2),
                "timestamp": datetime.now().isoformat(),
            }
            self.executionLog.append(errorRecord)
            logger.error(f"工作流失敗: {str(e)}")
            return {"status": "failed", "error": str(e), "metadata": errorRecord}

    def getStats(self) -> dict:
        total = len(self.executionLog)
        success = sum(1 for r in self.executionLog if r["status"] == "success")
        avgDuration = sum(r["duration"] for r in self.executionLog) / max(total, 1)
        return {"total_runs": total, "success_rate": f"{success/max(total,1)*100:.1f}%", "avg_duration": f"{avgDuration:.2f}s"}

workflow = CrewAIProductionWorkflow()
result = workflow.run("2026年AI Agent框架市場")
print(result)
print(workflow.getStats())

生產級要點：max_iter限制Agent迭代次數防止無限迴圈，max_rpm控制API呼叫頻率，executionLog記錄每次執行的元資料用於監控和最佳化，異常捕獲確保單次失敗不影響整體服務。

避坑指南：5大常見陷阱

❌ 陷阱1：Agent職責定義過於寬泛 ✅ 一個Agent只做一件事。「全能Agent」跟單Agent沒區別，反而多了協調開銷。role要具體到「市場資料搜尋專員」而非「分析師」。

❌ 陷阱2：Sequential流程處理複雜依賴 ✅ 任務間有條件分支或並行需求時，用Hierarchical流程。Sequential只能線性執行，強行用它處理複雜依賴會導致邏輯混亂。

❌ 陷阱3：Tool返回非字串型別 ✅ CrewAI Tool必須返回字串。返回dict或list會導致Agent解析失敗。在Tool內部做JSON序列化，異常時返回友好錯誤資訊。

❌ 陷阱4：忽略max_iter導致Agent無限迴圈 ✅ 設定max_iter=3~5限制Agent迭代次數。沒有限制時，Agent可能反覆呼叫Tool陷入迴圈，快速消耗Token。

❌ 陷阱5：不監控Token消耗和執行耗時 ✅ 生產環境必須記錄每次執行的Token消耗和耗時。CrewAI的Token消耗隨Agent數量和迭代次數指數增長，不監控等於燒錢。

報錯排查：10大常見錯誤

錯誤現象 可能原因 排查方式 解決方案
Agent無限迴圈呼叫Tool 未設定max_iter 檢查Agent設定 設定max_iter=3~5，限制迭代次數
Task輸出為空 expected_output定義不清 檢查Task描述 明確expected_output格式和內容要求
Sequential流程卡住 任務間資料傳遞斷裂 檢查上游Task輸出 確保上游輸出格式匹配下游輸入預期
Tool呼叫超時 外部API回應慢 檢查Tool內timeout設定 增加timeout，新增重試邏輯
Hierarchical流程Token爆炸 Manager過度委託 檢查Manager日誌 限制Manager委託次數，精簡任務描述
Agent輸出格式不一致 缺少格式約束 檢查expected_output 在Task描述中明確輸出格式模板
Crew kickoff報KeyError inputs參數缺失 檢查inputs字典 確保inputs包含所有{variable}佔位符
LLM呼叫429錯誤 API速率限制 檢查API配額 設定max_rpm，配置多Key輪換
Tool返回型別錯誤 返回了非字串 檢查Tool返回值 確保Tool返回str，用json.dumps序列化
記憶體佔用持續增長 Agent歷史未清理 監控記憶體使用 定期重啟Crew實例，限制上下文長度

進階最佳化技巧

1. 動態Agent建立。不要硬編碼所有Agent，根據任務型別動態建立。例如根據輸入主題自動選擇需要哪些專業Agent，減少不必要的Token消耗。用工廠函數封裝Agent建立邏輯，保持程式碼整潔。

2. 快取中間結果。對耗時長的Task（如搜尋、API呼叫）啟用結果快取，相同輸入直接返回快取。可以用@lru_cache裝飾Tool函數，或在外部維護Redis快取層。快取TTL建議設為1小時，避免資料過時。

3. 分層錯誤處理。Tool層捕獲網路異常返回友好資訊，Task層檢查expected_output格式，Crew層捕獲整體執行異常。三層防護確保單點故障不擴散。關鍵Task增加重試機制，最多重試2次。

4. Prompt工程最佳化。Agent的backstory和Task的description就是Prompt。遵循「角色-目標-約束」三段式：先定義身份，再明確目標，最後列出約束條件。避免模糊描述如「分析一下」，改為「列出3個關鍵趨勢並給出資料支撐」。

5. 成本分檔策略。簡單任務用GPT-4o-mini，複雜推理用GPT-4o，程式碼生成用DeepSeek-Coder。根據Task複雜度分配不同LLM，月度成本可降低60%以上。在Agent定義時指定不同LLM即可。

對比分析：CrewAI vs LangGraph vs AutoGen vs OpenAI Swarm

特性 CrewAI LangGraph AutoGen OpenAI Swarm
核心理念 角色化Agent+流程編排 狀態圖+條件邊 多Agent對話 輕量級Handoff
學習曲線 低 高 中 低
流程控制 Sequential/Hierarchical 自訂DAG 對話驅動 函數路由
工具整合 @tool裝飾器 LangChain Tool 函數註冊 函數定義
狀態管理 自動上下文傳遞 顯式State 訊息歷史 上下文變數
生產就緒 ⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐ ⭐⭐
適用場景 內容生產、研究分析 複雜工作流、RAG 多Agent討論 快速原型
推薦度 ⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐ ⭐⭐⭐

選型建議：需要角色化協作和快速上手選CrewAI；需要精細狀態控制和複雜DAG選LangGraph；需要多Agent自由討論選AutoGen；需要輕量快速原型選Swarm。CrewAI在角色定義的直觀性和流程編排的簡潔性上優勢明顯，是內容生產和研究分析場景的首選。

線上工具推薦

總結與展望

CrewAI工作流的核心不是簡單的Agent堆疊，而是角色精準定義、流程合理編排、工具高效整合三大原則的落地。5個核心模式——基礎定義、Sequential流程、Hierarchical流程、自訂Tool、生產級監控——覆蓋了從原型到生產的完整路徑。記住：一個Agent只做一件事、流程模式匹配任務結構、Tool是Agent的能力邊界，才能構建真正高效的多角色AI協作系統。未來CrewAI將深度整合MCP協定和更多企業級Tool，多Agent協作將從「能用」走向「好用」。

延伸閱讀

本站提供瀏覽器本地工具，免註冊即可試用 →

#CrewAI工作流#AI Agent编排#多角色协作#Agent框架#2026#AI与大数据