一、真实问题：三Agent协作崩盘实录

“半夜两点，客服Agent回复了用户‘可以退款’，质检Agent却同时发送‘不符合退款条件’，挽留Agent又弹出“赠送优惠券”。用户截图微博，第二天公关部炸了。”

这就是我团队去年在智能客服系统中踩的坑。三个Agent各自独立部署，通过Redis广播订阅通信，没有任何协作流程控制。结果就是数据不一致、决策冲突、用户情绪负值。

当时我们用了CrewAI v0.30.0，但把它当成了“多线程脚本调度器”，根本没有理解多Agent协作流程设计的本质。后来花了三周重构，测试了四种协作模式，才稳定下来。

二、问题分析：为什么简单的多Agent会崩？

多Agent协作的核心挑战：任务依赖和上下文一致。

上述案例中三个Agent的业务依赖链是：

• 客服Agent需要先识别用户意图
• 质检Agent必须等待客服回复后检查合规性
• 挽留Agent只有在质检不通过时才触发

而原始方案中三个Agent并行启动，没有等待、没有顺序、没有结果传递，自然混乱。

三、两种主流协作模式对比

我们重点测试了CrewAI支持的两种模式：

在我们的客服场景中，Sequential模式虽然简单但无法处理质检失败后的“分支”，而Hierarchical可以动态决定是否触发挽留。最终我们采用“分层+顺序混合”的方案：主线用顺序链，质检失败后由Manager Agent触发挽留分支。

四、完整代码实现（Python + YAML）

以下代码基于CrewAI 0.30.0、Python 3.11.5、OpenAI GPT-4-turbo（也可换本地模型）。

4.1 项目结构

crew_project/
├── config/
│   ├── agents.yaml
│   └── tasks.yaml
├── agents/
│   └── custom_agent.py
├── main.py
└── requirements.txt

4.2 YAML配置（多Agent和任务定义）

# config/agents.yaml
agents:
  - name: "客服Agent"
    role: "customer_service"
    goal: "准确理解用户需求，给出合规的初步回复"
    backstory: "你是顶尖的客服AI，擅长情感分析和快速应答"
    llm: "gpt-4-turbo"
    allow_delegation: false

  - name: "质检Agent"
    role: "quality_inspector"
    goal: "检查客服回复是否符合公司政策，标记风险"
    backstory: "你是严格的质量审查员，绝不放过任何违规"
    llm: "gpt-4-turbo"
    allow_delegation: false

  - name: "挽留Agent"
    role: "retention_specialist"
    goal: "当质检不过时，提供最有诚意的挽留方案"
    backstory: "你是客户留存专家，擅长赠送优惠券、升级服务"
    llm: "gpt-4-turbo"
    allow_delegation: false

# config/tasks.yaml
tasks:
  - name: "基础回复"
    agent: "客服Agent"
    description: "用户输入: {user_message}，请生成专业客服回复，附加风险评估分数(0-10)"
    expected_output: "JSON格式，包含reply和risk_score"

  - name: "质检审查"
    agent: "质检Agent"
    description: "基于客服回复和用户原始信息，判定是否合规。输入：{客服输出}"
    expected_output: "JSON格式，包含passed(布尔)和reason"

  - name: "挽留处理"
    agent: "挽留Agent"
    description: "质检未通过时执行。输入：{客服回复}和{质检报告}，生成挽留话术和优惠券方案"
    expected_output: "JSON格式，包含retention_message和offer"

4.3 Python主程序（分工协作流程）

# main.py
import json
from crewai import Agent, Task, Crew, Process
from crewai_tools import tool

# 模拟工具：查询用户历史
@tool("get_user_history")
def get_user_history(user_id: str) -> str:
    """返回用户历史投诉记录和消费等级"""
    mock = {
        "001": {"tier": "gold", "complaints": 2, "total_spent": 12000},
        "002": {"tier": "silver", "complaints": 0, "total_spent": 3000}
    }
    return json.dumps(mock.get(user_id, {"tier": "normal", "complaints": 0, "total_spent": 0}))

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

# 创建Agent对象
agents = {}
for a in agents_config:
    agents[a["name"]] = Agent(
        role=a["role"],
        goal=a["goal"],
        backstory=a["backstory"],
        llm=a["llm"],
        allow_delegation=a["allow_delegation"]
    )

# 创建Task对象（绑定Agent）
tasks = []
for t in tasks_config:
    agent_name = t["agent"]
    task_obj = Task(
        description=t["description"],
        expected_output=t["expected_output"],
        agent=agents[agent_name]
    )
    tasks.append(task_obj)

# 创建Crew - 使用Hierarchical流程，设置Manager Agent
manager_agent = Agent(
    role="流程协调员",
    goal="按顺序分配任务，当质检失败时插入挽留任务",
    backstory="你是经验丰富的项目管理者",
    llm="gpt-4-turbo",
    allow_delegation=True
)

crew = Crew(
    agents=list(agents.values()),
    tasks=tasks,
    manager_agent=manager_agent,
    process=Process.hierarchical,  # 关键：分层协调
    verbose=True
)

# 执行
user_id = "001"
user_message = "我对你们的产品质量很失望，要求退款"
result = crew.kickoff(inputs={"user_message": user_message})
print("Final result:", result)

4.4 客户端调用示例（Python）

# client.py
import requests
import json

url = "http://localhost:8000/crew"
payload = {
    "user_id": "001",
    "user_message": "产品质量差，我要投诉"
}
headers = {"Content-Type": "application/json"}
resp = requests.post(url, json=payload, headers=headers)
data = resp.json()
print(json.dumps(data, indent=2))
# 输出示例：
# {
#   "客服回复": "很抱歉给您带来不好的体验，我们支持退款。",
#   "质检结果": {"passed": false, "reason": "退款金额超过1000元需经理审批"},
#   "挽留": {"message": "赠送您200元优惠券和VIP服务", "offer": "200_coupon"}
# }

五、效果数据：10万次压测对比

测试环境：8vCPU 32GB RAM GPU (A10G)，OpenAI API延迟平均200ms。数据如下：

我们的混合模式在一致性和时间之间取得平衡，失败率从原始35%降低到4%。

六、避坑指南（这一条条都是真金白银的教训）

坑1：Agent角色定义不清晰导致任务混乱

最初我们把“质检Agent”的goal写成“检查并修复回复”，结果它擅自修改了客服内容。正确做法：每个Agent只能产生自己职责范围内的输出，不允许跨域修改其他Agent的输出。在CrewAI中，通过Task的expected_output格式限制。

坑2：上下文变量冲突和覆盖

多个Task使用相同的变量名（如“output”）导致下游拿到的是错误数据。解决方案：每个Task的输出变量名必须唯一，且用task名称前缀。可以用json字段名来隔离，如客服回复字段为"cs_reply"，质检为"qc_result"。

坑3：Hierarchical模式下的Manager Agent成为性能瓶颈

Manager Agent每次都要调用LLM来决策任务分配，额外增加200-500ms。减轻方法：将常规路径固化到顺序链中，只在分支时使用Manager。我们通过自定义Process类实现（代码略）。

坑4：任务依赖形成死循环

比如挽留Agent又去调用质检Agent。在CrewAI默认的hierarchical流程中，Manager可能会产生循环依赖。避坑：在Task描述中明确禁止循环调用，或者设置max_iterations=3。

坑5：Agent记忆混乱导致上下文丢失

CrewAI v0.30默认每个Agent有记忆，但任务之间上下文不会自动传递。需要手动在kickoff时通过inputs参数传入。我们使用了一个全局上下文管理器来维护共享数据。

七、总结

CrewAI的多Agent协作不是简单的“把agent扔进去跑”。你需要根据业务依赖图选择流程模式，严格控制上下文隔离，并仔细设计每个Agent的边界。我们的混合方案已经在生产环境稳定运行四个月，处理了超过50万次请求，故障率低于0.5%。

如果你也在用CrewAI，立刻去检查你的Agent配置和Task依赖，别等到凌晨两点被用户截图挂微博。