从一次线上事故说起

凌晨1点，售后客服系统报警。一个用户投诉工单在三个Agent之间来回踢了12轮，最后被错误路由到退款流程。用户等了一个小时，退款没到账，投诉升级。
排查日志发现：质检Agent认为「退款」关键词命中退款意图，情绪识别Agent判断用户「愤怒」，于是把工单升级给退款组。但用户真正的问题是收到破损商品，想「换货」而不是「退款」。

这个案例暴露的问题很简单——多Agent协作如果只按关键词路由、没有全局状态管理，必然出错。当时我们用的是自研编排脚本，通过Redis队列把Agent串起来，状态靠塞进JSON字段传递。代码越来越长，排查靠翻日志，改一个Agent的逻辑影响另外三个。

两个月后，我们基于CrewAI重构了这套流程。本文记录完整的方案对比、代码实现、压测数据和避坑经验。环境版本：Python 3.11.8、CrewAI 0.80.0、FastAPI 0.115.2、Redis 7.2.4、MySQL 8.0.35。

问题拆解：这个场景到底卡在哪

以售后工单处理为例，核心链路如下：

用户提交工单 → 意图分类 → 信息抽取 → 情绪判断 → 路由决策 → 生成回复 → 人工审核 → 归档

这套链路有三个核心问题：

• 状态同步困难：每个环节需要前序环节的输出，且偶尔需要回退重试
• 决策逻辑脆弱：条件分支一多，代码变成if-else泥潭
• Agent间数据冲突：同一个字段不同Agent可能给出不同结果（比如「退款」和「换货」同时命中）

方案对比：两种多Agent编排路线

路线A：LangGraph图编排

LangGraph 0.2.x版，用图结构定义节点和边。把每个Agent当作一个节点，API调用定义为「路由器」节点。这是目前的主流做法，适合「你能把流程完整画成图」的场景。

from langgraph.graph import StateGraph, END

graph = StateGraph(AgentState)
graph.add_node("classify", classify_agent)
graph.add_node("extract", extract_agent)
graph.add_node("route", route_agent)
graph.add_edge("classify", "extract")
graph.add_conditional_edges(
    "route",
    route_logic,
    {"refund": "refund_handler", "exchange": "exchange_handler", "END": END}
)
app = graph.compile()

优势：图结构可视化清晰，节点间数据流明确定义为State。适合复杂分支、需要人工干预、需要精确控制每一步的场景。

痛点：工程量大。每个节点要写胶水代码，状态管理得自己定义（State类型、Pydantic模型、回退逻辑）。我们写了将近3000行才算跑通。
另一个痛点是：真实业务中多个Agent需要并发执行（比如意图分类和情绪识别互不依赖），LangGraph的StateGraph在0.2.x版本里并发要手动写法，用invoke并发也行，但状态合并、冲突解决都得自己写。

路线B：CrewAI角色协作编排

CrewAI 0.80.0的核心思路：把Agent定义成「角色 + 目标 + 技能」，通过Task和Process来编排执行顺序。
我们最终选了这条路。原因有三：

• 声明式配置：Agent和Task用YAML配置，业务人员能看懂，改流程不用动代码
• 内置协作协议：Agent之间可以互相委托任务，支持上下文传递，省去自定义状态机
• Sequential和Hierarchical两种模式覆盖大部分业务场景

两种路线的数据对比

我们用一个测试集做了对比：500条真实脱敏工单，环境为4核8G云服务器，OpenAI GPT-4o-mini模型，temperature=0。

指标	LangGraph方案	CrewAI方案
代码量（核心编排层）	约2800行	约600行（含配置）
首轮开发耗时	10人日	4人日
单工单平均处理耗时	3分30秒	41秒
准确率（人工复核）	86.4%	87.2%
增加一个新环节平均耗时	2~3小时	20分钟（只改YAML）

注：LangGraph的耗时高主要因为我们实现了并发调用、状态合并和冲突消解。CrewAI在0.80版本内部对并发处理有优化（具体机制下文会讲）。

最终方案：CrewAI实现客服工单Agent流水线

架构设计

用户输入
    │
    ▼
┌─────────────┐
│ 入口 Agent  │  意图分类 + 实体抽取（并发）
└─────────────┘
    │
    ▼
┌─────────────┐
│ 策略 Agent  │  路由决策（仅调用LLM一次，不重复分析）
└─────────────┘
    │
    ▼
┌─────────────┐
│ 执行 Agent  │  根据路由结果调用对应API或生成回复
└─────────────┘
    │
    ▼
┌─────────────┐
│ 校验 Agent  │  检查API返回结果/回复质量，必要时触发二次调用
└─────────────┘
    │
    ▼
    输出

代码实现

第一步，用YAML定义Agent。这一步的好处是后面改Agent参数（比如换模型、调温度）不需要改代码。

# agents.yaml
intake_agent:
  role: >
    用户意图分类和关键信息抽取专家
  goal: >
    识别用户的问题类型（退款/换货/物流/维修），抽取订单号和商品名称
  backstory: >
    你是一名资深的客服调度专家，能够从用户描述中快速判断意图并提取关键信息。
  llm: gpt-4o-mini
  temperature: 0.1
  max_iterations: 3
  verbose: true

route_agent:
  role: >
    工单路由决策专家
  goal: >
    根据用户意图和信息完整度，决定执行路径：直接回复、退款处理、换货处理、人工介入
  backstory: >
    客服主管，熟悉所有业务流程和升级规则。仅在信息不足时提问，不重复用户已提供的内容。
  llm: gpt-4o-mini
  temperature: 0
  max_iterations: 2
  verbose: true

executor_agent:
  role: >
    工单执行专员
  goal: >
    执行路由决策，调用对应的业务API或生成答复话术
  backstory: >
    你的工作是完成具体的售后动作。始终基于已有信息，不自作主张改变流程。
  llm: gpt-4o-mini
  temperature: 0.2
  max_iterations: 3
  verbose: true

quality_agent:
  role: >
    质检校验专员
  goal: >
    检查执行结果是否准确、完整，发现错误及时纠正
  backstory: >
    你是客服质检员，检查回复是否解决问题。发现信息缺失或决策错误时，明确指出并给出修正建议。
  llm: gpt-4o-mini
  temperature: 0
  max_iterations: 1
  verbose: true

第二步，定义Task。每个Task绑定Agent，并声明输入上下文。

# tasks.yaml
classify_and_extract:
  description: >
    分析用户的工单描述，完成两项工作：
    1. 意图分类：从 [退款, 换货, 物流, 维修, 其他] 中选择一个
    2. 信息抽取：提取订单号、商品名、问题描述的关键字段
    用户输入：{user_query}
  expected_output: >
    JSON格式，包含intent、order_id、product_name、issue_description字段。
  agent: intake_agent

make_route_decision:
  description: >
    基于意图分类和抽取结果，决定执行路径。
    规则：
    - 意图为"退款"且订单号存在 → 执行退款API
    - 意图为"换货"且商品名存在 → 执行换货API
    - 信息缺失 → 生成追问话术
    - 情绪激烈或涉及金额超过5000元 → 转人工
    输入上下文：{classify_and_extract.output}
  expected_output: >
    决策结果JSON，包含action字段和reason字段。
  agent: route_agent

execute_action:
  description: >
    根据路由决策执行具体动作。
    输入上下文：{make_route_decision.output}
    调用对应的 API 或生成回复话术。API 调用失败时，说明原因并返回重试。
  expected_output: >
    包含execution_status、response_message、retry_flag字段的JSON。
  agent: executor_agent

quality_check:
  description: >
    校验执行结果。检查response_message是否解决用户问题，execution_status是否为success。
    如果retry_flag为true，解释失败原因。
    输入上下文：{execute_action.output}
  expected_output: >
    包含qa_status（pass/fail）、final_response、notes字段的JSON。
  agent: quality_agent

第三步，主程序编排。这是CrewAI的核心：用Crew对象把Agent和Task绑起来，定义Process为sequential。

# crew_workflow.py
import json
from crewai import Agent, Crew, Task, Process
from langchain_openai import ChatOpenAI

# 加载YAML配置
import yaml
with open("agents.yaml", "r") as f:
    agents_config = yaml.safe_load(f)
with open("tasks.yaml", "r") as f:
    tasks_config = yaml.safe_load(f)

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.1,
    api_key="your-api-key",  # 实际应用中使用环境变量
)

def build_agents():
    intake = Agent(
        role=agents_config["intake_agent"]["role"],
        goal=agents_config["intake_agent"]["goal"],
        backstory=agents_config["intake_agent"]["backstory"],
        llm=llm,
        temperature=agents_config["intake_agent"]["temperature"],
        max_iterations=agents_config["intake_agent"]["max_iterations"],
        verbose=True,
    )
    router = Agent(
        role=agents_config["route_agent"]["role"],
        goal=agents_config["route_agent"]["goal"],
        backstory=agents_config["route_agent"]["backstory"],
        llm=llm,
        temperature=0,
        max_iterations=2,
        verbose=True,
    )
    executor = Agent(
        role=agents_config["executor_agent"]["role"],
        goal=agents_config["executor_agent"]["goal"],
        backstory=agents_config["executor_agent"]["backstory"],
        llm=llm,
        temperature=0.2,
        max_iterations=3,
        verbose=True,
    )
    quality = Agent(
        role=agents_config["quality_agent"]["role"],
        goal=agents_config["quality_agent"]["goal"],
        backstory=agents_config["quality_agent"]["backstory"],
        llm=llm,
        temperature=0,
        max_iterations=1,
        verbose=True,
    )
    return [intake, router, executor, quality]

def build_tasks(agents, user_query):
    intake_agent, router_agent, executor_agent, quality_agent = agents
    
    task1 = Task(
        description=tasks_config["classify_and_extract"]["description"].format(user_query=user_query),
        expected_output=tasks_config["classify_and_extract"]["expected_output"],
        agent=intake_agent,
    )
    task2 = Task(
        description=tasks_config["make_route_decision"]["description"],
        expected_output=tasks_config["make_route_decision"]["expected_output"],
        agent=router_agent,
        context=[task1],  # 声明依赖task1的输出
    )
    task3 = Task(
        description=tasks_config["execute_action"]["description"],
        expected_output=tasks_config["execute_action"]["expected_output"],
        agent=executor_agent,
        context=[task1, task2],
    )
    task4 = Task(
        description=tasks_config["quality_check"]["description"],
        expected_output=tasks_config["quality_check"]["expected_output"],
        agent=quality_agent,
        context=[task2, task3],
    )
    return [task1, task2, task3, task4]

def run_crew(user_query: str):
    agents = build_agents()
    tasks = build_tasks(agents, user_query)
    
    crew = Crew(
        agents=agents,
        tasks=tasks,
        process=Process.sequential,
        verbose=True,
    )
    
    result = crew.kickoff(inputs={"user_query": user_query})
    return result

if __name__ == "__main__":
    sample_query = "我上周买的电饭煲收到了但是盖子坏了，订单号是PO20240115，想换一个"
    output = run_crew(sample_query)
    print("=== 最终结果 ===")
    print(output)

这里注意一个关键点：context=[task1]声明了任务间的数据依赖。CrewAI在Sequential模式下会根据context自动组装上下文传给下一个Task。
这也解释了为什么单工单耗时从3分半降到了41秒——我们没有把上下文全部塞进一个巨大的prompt里，而是每个Task只拿它需要的字段。context声明机制帮助控制了token长度。

并发优化：让不依赖的Agent并行执行

CrewAI 0.80.0支持在Sequential流程内声明并行Task吗？答案是：不直接支持。官方推荐的并行方式是「多个Crew实例并行」，或者在一个Task的description里通过「多个子任务」交给同一个Agent。但实际业务中，意图分类和情绪识别是两个独立Agent干的活。
我们的做法：FastAPI异步接口里用asyncio并发起两个Crew（一个做意图+信息抽取，一个做情绪+风险标记），最后合并结果再进入路由Crew。

# fastapi_app.py
import asyncio
import json
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from concurrent.futures import ThreadPoolExecutor

app = FastAPI(title="CrewAI 客服工单处理", version="1.0.0")
executor = ThreadPoolExecutor(max_workers=4)

class TicketRequest(BaseModel):
    user_query: str
    ticket_id: str = ""

def run_intent_crew(query: str):
    """意图分类和信息抽取Crew"""
    from crew_workflow import run_crew
    result = run_crew(query)  # 实际使用独立的运行函数
    return {"intent_result": json.dumps(str(result))}

def run_sentiment_crew(query: str):
    """情绪分析Crew：分析用户情绪和风险等级"""
    from crewai import Agent, Task, Crew, Process
    sentiment_agent = Agent(
        role="情绪分析专家",
        goal="判断用户情绪状态（平静/不满/愤怒/激烈）和风险等级",
        backstory="客服心理分析师",
        llm="gpt-4o-mini",
        temperature=0,
    )
    sentiment_task = Task(
        description=f"""分析以下用户消息的情绪和风险等级：
        用户消息：{query}
        输出JSON格式：{{"emotion": "...", "risk_level": "low/medium/high", "suggestion": "..."}}""",
        expected_output="包含emotion, risk_level, suggestion的JSON",
        agent=sentiment_agent,
    )
    crew = Crew(agents=[sentiment_agent], tasks=[sentiment_task], process=Process.sequential)
    result = crew.kickoff()
    return {"sentiment_result": json.dumps(str(result))}

@app.post("/process_ticket")
async def process_ticket(request: TicketRequest):
    loop = asyncio.get_event_loop()
    
    # 并发执行两个独立分析任务
    intent_future = loop.run_in_executor(executor, run_intent_crew, request.user_query)
    sentiment_future = loop.run_in_executor(executor, run_sentiment_crew, request.user_query)
    
    intent_result, sentiment_result = await asyncio.gather(intent_future, sentiment_future)
    
    # 合并结果进入路由决策Crew
    merged_context = f"""
    用户查询：{request.user_query}
    意图分析：{intent_result}
    情绪分析：{sentiment_result}
    """
    
    # 调用路由Crew
    final_decision = await loop.run_in_executor(executor, run_route_crew, merged_context)
    return {"ticket_id": request.ticket_id, "result": final_decision}

实测数据：串行两个分析（意图+情绪）平均耗时38秒；并行后平均耗时21秒，耗时降低了44.7%。

效果数据：重构前后对比

上线运行28天，统计12000条真实工单。

指标	自研脚本	CrewAI重构后	变化
平均处理耗时	4分12秒	41秒	↓83.7%
首次路由准确率	76.3%	87.2%	↑10.9%
错误路由率	11.4%	3.8%	↓66.7%
人工介入率	28.9%	12.1%	↓58.1%
API调用成本/单	约0.046美元	约0.031美元	↓32.6%
平均token消耗/单	约6800	约4300	↓36.8%

成本下降的原因：CrewAI的context机制只传递task间必要的数据，避免把前面的所有原始输出全部塞给后面的Agent。而自研脚本里，我们当时图省事，直接传了全量JSON。

避坑指南：实际遇到的7个坑

以下每个坑都是自己踩过的，不是从文档抄的。

坑1：process=Process.hierarchical 的 manager LLM

Hierarchical模式下，manager_llm如果不显式指定，CrewAI会用默认的OpenAI模型。如果你用的是国内模型或者代理，会出现「调用了一个不存在的模型名」报错。
解决：显式声明manager_llm。

crew = Crew(
    agents=agents,
    tasks=tasks,
    process=Process.hierarchical,
    manager_llm=llm,  # 必须指定
)

坑2：Agent的role_playing可能导致输出格式不稳定

CrewAI给Agent预设的backstory会让LLM「入戏」。但我们遇到过一次：质检Agent入戏太深，输出里自带语气词和评价「用户说得很有道理呢」，直接导致解析JSON失败。
解决：所有Agent的backstory里都加一句「只输出结构化JSON，不要任何解释性文字」。同时expected_output里写清楚格式样例。

坑3：max_iterations设置不当导致「Agent死循环」

如果max_iterations设置太大（比如默认的20），且Task的description里没写清楚「信息不足时直接询问用户」，Agent会反复调用工具重试。有次执行了17次工具调用，API账单翻了三倍。
解决：把max_iterations设到1~3，并在Task描述里写「信息不足时返回UNKNOWN字段」。

坑4：context传递的是「任务对象」不是「输出值」

在Task的description里引用{task1.output}，只有在kickoff执行时才会动态替换。但如果在Crew外想单独测试某个Task，需要手动构造上下文。
解决：写单元测试时，使用task.execute(agent=agent, context={"task1_output": "..."})方式测试单个Task。

坑5：YAML配置里不要用「中文冒号」

这个坑非常低级但浪费了我两个小时。在agents.yaml里写role: '客服助手：负责分类'，YAML解析把中文冒号当作普通字符，没问题。但如果你不小心在缩进前混用了全角空格，CrewAI的配置加载器直接抛异常，而且报错信息是指向下一行，不是出错行。
解决：用python -c "import yaml; yaml.safe_load(open('agents.yaml'))"先检查配置格式。

坑6：Hierarchical模式下manager Agent会「抢活」

Hierarchical模式下，manager Agent位于最上层，它有权决定「把任务分配给谁」。有一次用户只问了一个物流问题，manager居然把任务同时分配给了四个Agent，导致响应时间从40秒飙到130秒。
解决：在Task的description里明确写「这是一个简单任务，直接输出答案，不需要委派」。或者，对确定性高的流程，直接用Sequential。

坑7：耗时长不一定是LLM慢，可能是工具调用超时

我们的执行Agent需要调用内部订单API，FastAPI的默认超时是5秒。但当Agent连续两次调用同一API时，总耗时可能超过10秒。CrewAI的Agent在工具调用失败后会重试，重试次数默认等于max_iterations。
解决：外部API的时间设置参考值：HTTP连接2秒、读取5秒，整体不超8秒。同时把Agent工具函数里加上超时参数requests.get(url, timeout=(2, 5))。

什么时候用 CrewAI，什么时候放弃它

最后说点实在的判断标准。

建议用CrewAI的场景：

• Agent角色固定、流程相对明确（客服、内容审核、报告生成）
• 需要快速搭建多Agent MVP验证效果
• 希望业务人员能通过改YAML调整Agent行为

建议放弃CrewAI，回到LangGraph或自研的场景：

• 流程中出现状态循环（A→B→A），Sequential模式会死循环
• 需要细粒度控制token使用量（CrewAI内部会注入一些系统prompt，token消耗不是最省的）
• 需要严格的审计追踪（每一步的输入输出记录不够细）
• Agent数量超过10个，协作关系变成网状——CrewAI对这种复杂拓扑支持有限

我们的真实建议：先用CrewAI做POC，验证效果后如果发现编排灵活性不够，再迁到LangGraph。两个框架都支持导出/导入Agent参数，迁移成本不高。