实战案例与最佳实践
1. 实战案例：竞品分析自动化
场景描述
自动化完成竞品分析：搜索竞品信息 → 整理数据 → 生成对比报告。
完整代码
python
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool
from pydantic import BaseModel
from typing import List
# === 输出模型 ===
class Competitor(BaseModel):
name: str
description: str
strengths: List[str]
weaknesses: List[str]
pricing: str
class CompetitorReport(BaseModel):
title: str
summary: str
competitors: List[Competitor]
recommendation: str
# === 工具 ===
search = SerperDevTool()
# === Agents ===
researcher = Agent(
role="市场研究员",
goal="全面收集竞品信息，确保数据准确可靠",
backstory="""你是一位资深市场研究员，擅长从公开渠道收集竞品信息。
你注重数据的准确性和时效性，每条信息都会交叉验证。""",
tools=[search],
verbose=True
)
analyst = Agent(
role="竞争分析师",
goal="从收集的数据中提炼关键洞察和竞争策略",
backstory="""你是一位资深竞争分析师，精通 SWOT 分析、波特五力模型等分析框架。
你善于发现数据背后的规律和趋势。""",
verbose=True
)
strategist = Agent(
role="战略顾问",
goal="基于分析结果提供可执行的战略建议",
backstory="""你是一位企业战略顾问，拥有 20 年咨询经验。
你的建议以务实、可执行著称，每个建议都附带具体的行动步骤。""",
verbose=True
)
# === Tasks ===
research_task = Task(
description="""调研以下竞品的最新信息：
1. 产品定位和核心功能
2. 目标用户群体
3. 定价策略
4. 市场份额和用户评价
5. 最近的重大更新或战略调整
竞品列表：{competitors}
我们的产品：{our_product}""",
expected_output="每个竞品的详细信息报告，数据来源标注清晰",
agent=researcher
)
analysis_task = Task(
description="""基于调研数据，进行深度对比分析：
1. 各竞品的 SWOT 分析
2. 功能对比矩阵
3. 定价对比
4. 差异化优势识别
5. 市场空白点分析""",
expected_output="结构化的竞品对比分析报告",
agent=analyst,
context=[research_task]
)
strategy_task = Task(
description="""基于分析结果，为我们的产品提供战略建议：
1. 短期（3 个月）优化方向
2. 中期（6 个月）差异化策略
3. 长期（1 年）市场定位
4. 具体可执行的行动清单""",
expected_output="""战略建议报告，包含：
- 执行摘要
- 竞品对比分析
- 差异化策略
- 行动计划
- 风险提示""",
agent=strategist,
context=[analysis_task],
output_file="output/competitor_analysis.md"
)
# === Crew ===
crew = Crew(
agents=[researcher, analyst, strategist],
tasks=[research_task, analysis_task, strategy_task],
process=Process.sequential,
memory=True,
verbose=True
)
# === 执行 ===
result = crew.kickoff(
inputs={
"competitors": "Notion, Obsidian, Roam Research",
"our_product": "一款面向中国用户的 AI 知识管理工具"
}
)
print(result)2. 实战案例：代码审查流水线
python
from crewai import Agent, Task, Crew, Process
# === Agents ===
reviewer = Agent(
role="高级代码审查员",
goal="全面审查代码质量，发现潜在问题",
backstory="""你是一位有 15 年经验的代码审查专家，精通 Python、TypeScript 等语言。
你关注：代码风格、性能、安全性、可维护性、测试覆盖率。""",
verbose=True
)
security_expert = Agent(
role="安全专家",
goal="检查代码中的安全漏洞和风险",
backstory="""你是一位网络安全专家，专注于应用层安全。
你熟悉 OWASP Top 10，擅长发现 SQL 注入、XSS、CSRF 等漏洞。""",
verbose=True
)
optimizer = Agent(
role="性能优化专家",
goal="识别性能瓶颈并提供优化建议",
backstory="""你是一位性能优化专家，擅长算法优化、缓存策略、并发处理等。
你的优化建议以量化数据为支撑。""",
verbose=True
)
# === Tasks ===
code_review_task = Task(
description="""审查以下代码：
```python
{code}检查项：
- 代码风格和规范
- 错误处理
- 边界条件
- 可维护性""", expected_output="代码审查报告，按严重程度分类列出问题", agent=reviewer )
security_task = Task( description="对同一段代码进行安全审查，检查潜在的安全风险", expected_output="安全审查报告，列出漏洞和修复建议", agent=security_expert, context=[code_review_task] )
optimization_task = Task( description="分析代码性能，提供优化建议", expected_output="性能分析报告，包含优化方案和预期收益", agent=optimizer, context=[code_review_task] )
=== Crew ===
crew = Crew( agents=[reviewer, security_expert, optimizer], tasks=[code_review_task, security_task, optimization_task], process=Process.sequential, verbose=True )
result = crew.kickoff(inputs={ "code": """ def get_user(user_id): query = f"SELECT * FROM users WHERE id = {user_id}" result = db.execute(query) return result """ })
## 3. 生产环境最佳实践
### 3.1 错误处理
```python
import logging
from crewai import Crew
logger = logging.getLogger(__name__)
def run_crew_safely(crew, inputs=None, max_retries=3):
"""带重试和错误处理的 Crew 执行"""
for attempt in range(max_retries):
try:
result = crew.kickoff(inputs=inputs)
return result
except Exception as e:
logger.error(f"执行失败 (尝试 {attempt + 1}/{max_retries}): {e}")
if attempt == max_retries - 1:
raise
# 可选：指数退避
import time
time.sleep(2 ** attempt)3.2 成本控制
python
from crewai import Agent, LLM
# 按任务复杂度选择模型
def get_llm_for_task(complexity: str) -> LLM:
models = {
"simple": LLM(model="openai/gpt-4o-mini"), # 便宜
"medium": LLM(model="openai/gpt-4o"), # 中等
"complex": LLM(model="anthropic/claude-sonnet-4-20250514"), # 强
}
return models.get(complexity, models["medium"])
# 使用
simple_agent = Agent(
role="记录员",
goal="记录信息",
backstory="高效记录",
llm=get_llm_for_task("simple") # 简单任务用便宜模型
)
complex_agent = Agent(
role="架构师",
goal="设计系统架构",
backstory="资深架构师",
llm=get_llm_for_task("complex") # 复杂任务用强模型
)3.3 限流与重试
python
agent = Agent(
role="研究员",
goal="调研",
backstory="研究员",
max_rpm=10, # 每分钟最多 10 次 API 调用
max_iter=15, # 单次任务最多 15 次迭代
max_retry_limit=3 # 出错最多重试 3 次
)3.4 异步执行
python
import asyncio
async def main():
crew = Crew(
agents=[agent1, agent2],
tasks=[task1, task2],
verbose=True
)
# 非阻塞执行
result = await crew.kickoff_async()
return result
# 批量并行执行
async def batch_run(crew, inputs_list):
tasks = [crew.kickoff_async(inputs=inp) for inp in inputs_list]
return await asyncio.gather(*tasks)3.5 日志与监控
python
import logging
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('crew_execution.log'),
logging.StreamHandler()
]
)
# Crew 级别的回调
def step_callback(step_output):
logging.info(f"Step 完成: {str(step_output)[:200]}")
def task_callback(task_output):
logging.info(f"Task 完成，输出长度: {len(str(task_output))}")
crew = Crew(
agents=[...],
tasks=[...],
step_callback=step_callback,
task_callback=task_callback,
verbose=True
)4. 常见设计模式
4.1 研究 → 分析 → 报告
最经典的三段式流水线：
Researcher → Analyst → Writer适用场景：市场调研、技术评估、竞品分析。
4.2 多 Agent 辩论
多个 Agent 从不同角度审视同一问题：
python
optimist = Agent(role="乐观分析师", ...)
pessimist = Agent(role="悲观分析师", ...)
judge = Agent(role="裁判", ...)
# 正方
pro_task = Task(description="从正面分析...", agent=optimist)
# 反方
con_task = Task(description="从反面分析...", agent=pessimist)
# 综合
judge_task = Task(description="综合双方观点...", agent=judge, context=[pro_task, con_task])4.3 规划 → 执行 → 审核
Planner → Executor → Reviewer
↑ │
└────── 不通过 ────────┘适用场景：代码开发、文章写作、方案设计。
4.4 分类 → 路由 → 专项处理
Classifier → Router → Specialist A
→ Specialist B
→ Specialist C适用场景：客服工单、内容审核、任务分发。
5. 常见问题排查
Q1: Agent 输出质量不高
原因：role/goal/backstory 定义模糊。
解决：
python
# ❌ 模糊
Agent(role="助手", goal="帮忙", backstory="你是助手")
# ✅ 明确
Agent(
role="高级 Python 开发工程师",
goal="编写符合 PEP 8 规范、包含完整类型注解的生产级代码",
backstory="你有 10 年 Python 开发经验，精通 FastAPI、SQLAlchemy 等框架，"
"代码风格以 Google Python Style Guide 为标准"
)Q2: 执行时间过长
原因：max_iter 过大或 Agent 陷入循环。
解决：
python
agent = Agent(
...
max_iter=10, # 限制迭代次数
max_rpm=5, # 限制 API 调用频率
max_retry_limit=2 # 限制重试次数
)Q3: API 限流错误
解决：
python
agent = Agent(
...
max_rpm=5 # 降低请求频率
)Q4: Token 超限
原因：输入上下文过长。
解决：
- 精简 Task 的
description - 使用
context参数只传递必要的前置任务 - 选择更大的上下文窗口模型
Q5: 多 Agent 输出雷同
原因：角色定义太相似。
解决：
python
# ❌ 角色雷同
agent1 = Agent(role="分析师 A", ...)
agent2 = Agent(role="分析师 B", ...)
# ✅ 角色差异化
agent1 = Agent(role="技术分析师", goal="从技术角度分析可行性", ...)
agent2 = Agent(role="商业分析师", goal="从商业角度分析盈利模式", ...)6. CrewAI 训练与测试
6.1 训练 Crew
bash
# 通过 CLI 训练
crewai train
# 通过代码训练
crew.train(n_iterations=10, inputs={"topic": "AI"})训练通过多轮执行优化 Agent 的行为模式，使其输出更符合预期。
6.2 测试 Crew
bash
# 通过 CLI 测试
crewai test
# 通过代码测试
crew.test(n_iterations=3, inputs={"topic": "AI"})6.3 重放执行
bash
# 重放上次执行
crewai replay重放功能可以复现上次的执行过程，方便调试和分析。
总结
| 知识点 | 要点 |
|---|---|
| 角色设计 | 具体、专业、有区分度 |
| 模型选择 | 按任务复杂度选择不同 LLM |
| 错误处理 | try/except + 重试机制 |
| 成本控制 | 轻量任务用小模型，限流控制 |
| 异步执行 | kickoff_async() 用于生产环境 |
| 常见模式 | 研究→分析→报告、辩论、审核循环 |
| 训练测试 | crewai train / crewai test 优化行为 |