CrewAI 2026：多Agent协作框架的生产部署经验

from crewai import Agent, Task, Crew, Process
from crewai.tools import BaseTool

# ❌ 错误：角色过于宽泛
bad_agent = Agent(
    role="通用助手",
    goal="帮助用户解决所有问题",
    backstory="一个全能的AI助手"
)

# ✅ 正确：角色聚焦且边界清晰
financial_analyst = Agent(
    role="上市公⻔财务分析师",
    goal="分析目标公司的财务报表，识别风险和机会，输出结构化评估报告",
    backstory="""你是一位拥有15年经验的上市公司财务分析师，
    专注于消费零售赛道。你擅长从财报数据中发现异常信号，
    曾多次提前预警财务造假。你只分析公开财报数据，
    不做股价预测或投资建议。""",
    tools=[FinancialDataTool(), SECQueryTool()],
    llm="gpt-4o",
    max_iter=5,  # 2026新特性：限制迭代次数防止无限循环
    verbose=True
)

# 研究报告生成Crew的任务分解
research_tasks = [
    Task(
        description="""分析{company}最近三年的财务报表：
        1. 营收增长率和毛利率趋势
        2. 应收账款周转天数变化
        3. 存货周转率变化
        4. 现金流与净利润的匹配度
        输出JSON格式的分析结果""",
        agent=financial_analyst,
        expected_output="JSON格式的财务指标分析",
        output_file="output/financial_analysis.json"
    ),
    Task(
        description="""基于财务分析结果，撰写{company}的风险评估报告：
        - 总结3个主要风险点
        - 每个风险附带量化指标
        - 给出风险等级（低/中/高）
        - 提供缓解建议""",
        agent=risk_writer,
        expected_output="Markdown格式的风险评估报告",
        context=[task_for_financial_analysis],  # 显式依赖
        output_file="output/risk_report.md"
    )
]

# docker-compose.yml - CrewAI生产部署
version: '3.9'
services:
  crewai-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - CREWAI_TELEMETRY=false
      - CREWAI_STORAGE_BACKEND=redis
      - REDIS_URL=redis://redis:6379
      - CREWAI_MAX_CONCURRENT_CREWS=10
    deploy:
      replicas: 3
      resources:
        limits:
          memory: 2G
          cpus: '2.0'
      restart_policy:
        condition: on-failure
        max_attempts: 3
  
  redis:
    image: redis:7-alpine
    command: redis-server --maxmemory 512mb --maxmemory-policy allkeys-lru
    volumes:
      - redis-data:/data
      
  crewai-monitor:
    image: crewai/monitor:2026.1
    ports:
      - "3000:3000"
    environment:
      - CREWAI_API_URL=http://crewai-api:8000
    depends_on:
      - crewai-api

volumes:
  redis-data:

import asyncio
from crewai import Crew
from crewai.memory import LongTermMemory

async def run_analysis_pipeline(companies: list[str]):
    """并行运行多个公司的分析Crew"""
    
    crews = []
    for company in companies:
        crew = Crew(
            agents=[financial_analyst, risk_writer, compliance_checker],
            tasks=build_tasks_for_company(company),
            process=Process.sequential,
            memory=True,
            long_term_memory=LongTermMemory(
                storage="redis",
                embedding_model="text-embedding-3-small"
            ),
            # 2026新特性
            max_execution_time=300,  # 单Crew最大执行时间
            retry_on_failure=True,
            max_retries=2
        )
        crews.append(crew)
    
    # 并行执行所有Crew，最多同时运行5个
    results = await asyncio.gather(
        *[crew.kickoff_async() for crew in crews],
        return_exceptions=True
    )
    
    # 处理结果和异常
    successful = []
    failed = []
    for company, result in zip(companies, results):
        if isinstance(result, Exception):
            failed.append({"company": company, "error": str(result)})
        else:
            successful.append({"company": company, "result": result})
    
    return {"successful": successful, "failed": failed}

from crewai.monitoring import CrewMonitor
import prometheus_client as prom

# Prometheus指标
crew_duration = prom.Histogram('crew_duration_seconds', 'Crew执行耗时', ['crew_name'])
agent_iterations = prom.Counter('agent_iterations_total', 'Agent迭代次数', ['agent_role'])
task_failures = prom.Counter('task_failures_total', '任务失败次数', ['task_type'])

# 自定义监控回调
class ProductionMonitor(CrewMonitor):
    def on_crew_start(self, crew, inputs):
        self.start_time = time.time()
        logger.info(f"Crew启动: {crew.name}, 输入: {inputs}")
    
    def on_agent_iteration(self, agent, iteration, thought):
        agent_iterations.labels(agent_role=agent.role).inc()
        if iteration > 3:
            logger.warning(f"Agent {agent.role} 已迭代{iteration}次，可能存在循环")
    
    def on_task_complete(self, task, output):
        duration = time.time() - self.start_time
        crew_duration.labels(crew_name=self.crew_name).observe(duration)
        logger.info(f"任务完成: {task.description[:50]}... 耗时{duration:.1f}s")
    
    def on_task_failure(self, task, error):
        task_failures.labels(task_type=type(task).__name__).inc()
        logger.error(f"任务失败: {error}")
        # 发送告警
        alert_manager.send(f"CrewAI任务失败: {task.description[:100]}")

# 注册监控
crew = Crew(
    agents=[...],
    tasks=[...],
    monitor=ProductionMonitor()
)

# 模型分层策略
analyst_agent = Agent(
    role="数据分析师",
    llm="gpt-4o",  # 复杂推理用大模型
    ...
)

writer_agent = Agent(
    role="内容撰写员", 
    llm="gpt-4o-mini",  # 写作用小模型
    ...
)

reviewer_agent = Agent(
    role="质量审核员",
    llm="gpt-4o",  # 质量判断用大模型
    ...
)

# 全局迭代限制
crew = Crew(
    agents=[agent_a, agent_b],
    tasks=[...],
    max_iter=10,  # 全局最大迭代
    # 2026新特性：循环检测
    loop_detection=True,
    loop_threshold=3  # 连续3次相同模式触发终止
)

from crewai.memory import MemoryManager

memory = MemoryManager(
    short_term_capacity=10,  # 保留最近10轮对话
    long_term_enabled=True,
    long_term_storage="redis",
    # 自动压缩旧记忆
    compression_enabled=True,
    compression_threshold=50  # 超过50条触发压缩
)