CrewAI 生产实践：构建虚拟 AI 团队

Crew = Agents + Tasks + Process

# pip install crewai crewai-tools

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

# 配置 LLM
llm = ChatOpenAI(model="gpt-5", temperature=0.7)

# 定义一个内容创作团队
strategist = Agent(
    role='内容策略师',
    goal='制定内容策略，确保内容符合品牌调性和目标受众需求',
    backstory="""
    你是一位拥有 10 年经验的内容策略师，曾在多家科技公司
    负责内容规划。你擅长分析受众画像、制定内容日历和优化
    内容分发策略。
    """,
    llm=llm,
    verbose=True,
    memory=True,  # 启用记忆
    allow_delegation=False,
)

writer = Agent(
    role='技术写作专家',
    goal='根据策略师的要求创作高质量的技术内容',
    backstory="""
    你是一位资深技术写作专家，擅长将复杂的技术概念
    转化为通俗易懂的文章。你精通 Markdown 格式，
    善于使用代码示例和图表来增强理解。
    """,
    llm=llm,
    verbose=True,
    memory=True,
)

editor = Agent(
    role='内容编辑',
    goal='审核并优化内容，确保质量和准确性',
    backstory="""
    你是一位严格但公正的内容编辑，关注语法准确性、
    逻辑连贯性和事实正确性。你会提出具体的修改建议，
    而不是泛泛而谈。
    """,
    llm=llm,
    verbose=True,
)

strategy_task = Task(
    description="""
    为一篇关于「AI Agent 在企业中的应用」的文章制定内容策略。
    
    要求：
    1. 目标受众：技术决策者（CTO、技术总监）
    2. 内容长度：3000-5000 字
    3. 风格：专业但不晦涩
    4. 需要覆盖的关键点：
       - AI Agent 的定义和价值
       - 主流框架对比
       - 企业落地案例
       - ROI 分析
    
    输出：详细的内容大纲和写作要点。
    """,
    expected_output="一份详细的内容大纲，包含章节结构和每章的要点",
    agent=strategist,
)

writing_task = Task(
    description="""
    根据策略师的大纲，撰写完整的技术文章。
    
    要求：
    - 使用 Markdown 格式
    - 包含代码示例
    - 包含对比表格
    - 适当的图表说明
    """,
    expected_output="一篇完整的 Markdown 格式技术文章",
    agent=writer,
    context=[strategy_task],  # 依赖前一个任务的输出
)

editing_task = Task(
    description="""
    审查文章并优化：
    1. 检查事实准确性
    2. 优化段落过渡
    3. 修正语法错误
    4. 确保风格一致性
    5. 输出最终版本
    """,
    expected_output="审查意见 + 最终优化版文章",
    agent=editor,
    context=[writing_task],
    output_file='output/final_article.md',  # 保存到文件
)

# 创建 Crew
content_crew = Crew(
    agents=[strategist, writer, editor],
    tasks=[strategy_task, writing_task, editing_task],
    process=Process.sequential,  # 顺序执行
    verbose=True,
    memory=True,               # 团队级记忆
    planning=True,             # 执行前先规划
    # 层次模式配置
    # process=Process.hierarchical,
    # manager_llm=llm,
)

# 执行
result = content_crew.kickoff()
print(result.raw)

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
import requests
import json

class WebSearchInput(BaseModel):
    query: str = Field(..., description="搜索关键词")
    max_results: int = Field(default=5, description="最大返回结果数")

class WebSearchTool(BaseTool):
    name: str = "web_search"
    description: str = "搜索互联网获取最新信息"
    args_schema: type[BaseModel] = WebSearchInput
    
    def _run(self, query: str, max_results: int = 5) -> str:
        # 使用搜索 API
        response = requests.get(
            "https://api.search.brave.com/res/v1/web/search",
            headers={"X-Subscription-Token": "YOUR_API_KEY"},
            params={"q": query, "count": max_results}
        )
        results = response.json().get("web", {}).get("results", [])
        
        formatted = []
        for r in results:
            formatted.append(f"标题: {r['title']}\n链接: {r['url']}\n摘要: {r['description']}\n")
        
        return "\n".join(formatted) if formatted else "未找到相关结果"

class DatabaseQueryInput(BaseModel):
    sql: str = Field(..., description="SQL 查询语句")

class DatabaseQueryTool(BaseTool):
    name: str = "database_query"
    description: str = "查询企业数据库获取业务数据"
    args_schema: type[BaseModel] = DatabaseQueryInput
    
    def _run(self, sql: str) -> str:
        import sqlite3
        conn = sqlite3.connect("data/company.db")
        cursor = conn.execute(sql)
        columns = [d[0] for d in cursor.description]
        rows = cursor.fetchall()
        conn.close()
        
        # 格式化为表格
        result = [", ".join(columns)]
        for row in rows[:20]:  # 限制返回行数
            result.append(", ".join(str(v) for v in row))
        return "\n".join(result)

# 将工具分配给 Agent
researcher = Agent(
    role='市场研究员',
    goal='收集和分析市场数据',
    backstory='你是市场研究专家，擅长数据收集和分析。',
    llm=llm,
    tools=[WebSearchTool(), DatabaseQueryTool()],
)

# 层次模式：管理者 Agent 负责任务分配
from crewai import Process

manager = Agent(
    role='项目经理',
    goal='协调团队成员，确保任务按时高质量完成',
    backstory='你是一位经验丰富的项目经理，擅长资源分配和进度管理。',
    llm=llm,
    allow_delegation=True,  # 允许委派任务
)

hierarchical_crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.hierarchical,
    manager_agent=manager,  # 指定管理者
    manager_llm=llm,
    verbose=True,
)

from crewai import Crew
from crewai.utilities import Logger

class ProductionCrew:
    def __init__(self, config: dict):
        self.config = config
        self.max_retries = config.get("max_retries", 3)
        self.logger = Logger(verbose=True)
    
    def run_with_retry(self, crew: Crew, inputs: dict) -> dict:
        """带重试机制的 Crew 执行"""
        for attempt in range(self.max_retries):
            try:
                result = crew.kickoff(inputs=inputs)
                return {
                    "status": "success",
                    "result": result.raw,
                    "token_usage": result.token_usage,
                }
            except Exception as e:
                self.logger.log(f"尝试 {attempt+1} 失败: {e}")
                if attempt < self.max_retries - 1:
                    # 调整重试策略
                    crew.llm.temperature = min(crew.llm.temperature + 0.1, 1.0)
                    continue
                return {
                    "status": "failed",
                    "error": str(e),
                    "attempt": attempt + 1,
                }

# 使用 LangSmith 或自定义回调进行监控
from langchain.callbacks import get_openai_callback

def monitored_crew_run(crew: Crew, task: str):
    """带监控的 Crew 执行"""
    
    # Token 使用追踪
    with get_openai_callback() as cb:
        result = crew.kickoff(task)
    
    metrics = {
        "total_tokens": cb.total_tokens,
        "prompt_tokens": cb.prompt_tokens,
        "completion_tokens": cb.completion_tokens,
        "total_cost": cb.total_cost,
        "execution_time": result.execution_time,
    }
    
    # 发送到监控系统
    send_metrics_to_dashboard(metrics)
    
    # 输出执行报告
    print(f"""
    ═══════════════════════════════════
    Crew 执行报告
    ═══════════════════════════════════
    任务: {task[:50]}...
    状态: 成功
    Token 使用: {metrics['total_tokens']:,}
    成本: ${metrics['total_cost']:.4f}
    耗时: {metrics['execution_time']:.1f}s
    ═══════════════════════════════════
    """)
    
    return result, metrics

from crewai.utilities import CacheHandler
import hashlib
import json
import os

class FileCacheHandler(CacheHandler):
    """文件系统缓存"""
    
    def __init__(self, cache_dir: str = "./.crew_cache"):
        super().__init__()
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)
    
    def _hash_key(self, agent_name: str, input_text: str) -> str:
        return hashlib.md5(f"{agent_name}:{input_text}".encode()).hexdigest()
    
    def get(self, agent_name: str, input_text: str):
        key = self._hash_key(agent_name, input_text)
        path = os.path.join(self.cache_dir, f"{key}.json")
        if os.path.exists(path):
            with open(path) as f:
                return json.load(f)
        return None
    
    def set(self, agent_name: str, input_text: str, value):
        key = self._hash_key(agent_name, input_text)
        path = os.path.join(self.cache_dir, f"{key}.json")
        with open(path, "w") as f:
            json.dump(value, f, ensure_ascii=False)

# 使用缓存
crew = Crew(
    agents=[...],
    tasks=[...],
    cache_handler=FileCacheHandler(),
    # 启用嵌入缓存
    embedder={
        "provider": "openai",
        "config": {
            "model": "text-embedding-3-small",
        }
    },
)

# 模型分层策略
models = {
    "planning": "gpt-5",           # 规划用强模型
    "execution": "gpt-4o-mini",     # 执行用快速模型
    "review": "gpt-5",             # 审查用强模型
    "summary": "gpt-4o-mini",      # 总结用快速模型
}

# 为不同 Agent 分配不同模型
strategist = Agent(role='策略师', llm=ChatOpenAI(model=models["planning"]))
writer = Agent(role='写作者', llm=ChatOpenAI(model=models["execution"]))
editor = Agent(role='编辑', llm=ChatOpenAI(model=models["review"]))