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

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

# 定义具有专业工具的Agent
@tool
def financial_data_query(query: str) -> str:
    """查询金融数据库，支持股票、基金、债券数据"""
    # 实际实现...
    return data

@tool
def risk_calculator(portfolio: str) -> dict:
    """计算投资组合风险指标"""
    return {"var_95": 0.12, "max_drawdown": 0.25}

analyst = Agent(
    role="高级金融分析师",
    goal="提供准确的市场分析和投资建议",
    backstory="拥有15年华尔街经验的CFA持证人，擅长跨市场分析",
    tools=[financial_data_query, risk_calculator],
    llm="gpt-4o",
    max_iter=5,          # 2026新增：限制单Agent迭代次数
    max_rpm=30,          # 限流：每分钟最多30次LLM调用
    memory=True,         # 启用Agent级记忆
    verbose=True
)

risk_manager = Agent(
    role="风险管理官",
    goal="确保所有投资建议符合风险控制标准",
    backstory="前银监会风控专家，对系统性风险有敏锐直觉",
    llm="claude-sonnet-4",
    max_iter=3,
    memory=True
)

# 定义任务链
analysis_task = Task(
    description="分析当前市场环境，评估{asset}的投资价值",
    expected_output="包含基本面、技术面、风险评估的完整报告",
    agent=analyst,
    max_execution_time=120  # 2026新增：任务级超时
)

review_task = Task(
    description="审查分析报告，确认风险指标在可接受范围内",
    expected_output="风险审批意见，包含通过/拒绝及理由",
    agent=risk_manager,
    context=[analysis_task],  # 依赖前置任务输出
    max_execution_time=60
)

# 组建团队
crew = Crew(
    agents=[analyst, risk_manager],
    tasks=[analysis_task, review_task],
    process=Process.sequential,  # 2026新增：Process.hierarchical 支持层级管理
    memory=True,
    embedder={                  # 2026新增：可配置的嵌入模型
        "provider": "openai",
        "config": {"model": "text-embedding-3-small"}
    },
    manager_llm="gpt-4o",      # 层级模式下的管理者LLM
    planning=True,              # 启用任务规划阶段
    output_log_file="crew_log.json"
)

result = crew.kickoff(inputs={"asset": "贵州茅台(600519)"})

# 仲裁Agent配置
arbitrator = Agent(
    role="决策仲裁人",
    goal="在Agent间出现分歧时做出最终决策",
    backstory="资深决策科学专家，擅长在不确定性下做判断",
    llm="gpt-4o",
    allow_delegation=False  # 仲裁Agent不能再委托任务
)

crew = Crew(
    agents=[agent1, agent2, agent3],
    tasks=[task1, task2, task3],
    process=Process.sequential,
    context_compression=True,      # 自动压缩前置上下文
    context_window_strategy="sliding",  # 滑动窗口策略
    max_context_tokens=8000        # 限制上下文token数
)

# 配置记忆衰减策略
crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True,
    memory_config={
        "provider": "chroma",
        "decay_strategy": "time_weighted",  # 时间加权衰减
        "decay_factor": 0.95,               # 衰减因子
        "cleanup_interval": 100,            # 每100次执行清理一次
        "max_memories_per_agent": 500       # 每Agent最多保留500条记忆
    }
)