CrewAI 入门实战指南

我花了三个小时调试一个不断虚构API端点的CrewAI代理，这让我学到了至关重要的一点：CrewAI的强大之处也正是它最大的陷阱。无论你是从LangChain转过来，还是刚开始接触多代理系统，你很快就会意识到，编排多个AI代理不仅仅是串联提示词——它关乎管理依赖关系、上下文和故障模式。本指南将带你了解我在构建真实世界的CrewAI系统（用于自动化博客内容生成）时学到的东西，包括那些出错的代码以及我是如何修复的。

痛点：为什么不直接用单个代理？

你可能尝试过用单个大语言模型调用来生成博客文章。它确实能工作——直到你需要事实核查、搜索引擎优化和格式化。单个代理会忘记上下文、自相矛盾或生成无意义的内容。CrewAI通过让你定义专业化的代理来解决这个问题，这些代理像人类团队一样相互传递任务。但有个问题：CrewAI的简洁性掩盖了复杂性。如果你不仔细设计代理的角色和任务，就会遇到循环依赖、无限循环，或者代理拒绝交接工作的情况。

第一步：安装CrewAI（以及陷阱）

pip install crewai

我以为这会安装所有需要的东西。错了。 CrewAI依赖langchain和openai——但不是最新版本。如果你使用Python 3.12，会遇到pydantic冲突。以下是我使用的确切修复方法：

pip install crewai langchain==0.1.0 openai==1.6.1 pydantic==2.5.0

如果不固定这些版本，你会遇到ImportError: cannot import name 'BaseModel' from 'pydantic'。我为此浪费了30分钟。

第二步：定义你的代理（要么具体，要么受苦）

CrewAI中的代理被定义为具有role、goal和backstory的Python类。背景故事是可选的，但至关重要——它控制代理的语气和行为。以下是我开始时的写法：

from crewai import Agent

class Researcher(Agent):
    role = "研究员"
    goal = "查找关于AI在医疗保健领域的最新新闻"
    backstory = "你是一位严谨的研究员，会验证信息来源。"

这样写是可行的，但过于模糊。代理会生成通用回复。经过测试，我学会了添加约束条件：

class Researcher(Agent):
    role = "高级医疗AI研究员"
    goal = "查找3篇近期（2024年）关于AI在肿瘤学领域的同行评审论文"
    backstory = """你在医疗AI领域有10年经验。
    你总是引用具体的PMID或DOI链接。
    你从不编造来源。"""

注意引用来源的明确指令和年份约束。没有这些，我的代理编造了虚假论文。CrewAI不会验证事实——它信任大语言模型。

第三步：创建正确链接的任务

任务是最多人失败的地方。每个任务都有description、expected_output和agent。诀窍在于让任务依赖前一个任务的输出。我构建了一个双代理流水线：

from crewai import Task

research_task = Task(
    description="查找3篇近期AI医疗论文。输出包含标题和链接的列表。",
    expected_output="包含3篇论文的要点列表，包括标题、年份和URL",
    agent=researcher
)

writing_task = Task(
    description="""基于研究输出，撰写一篇500字的博客文章，
    总结研究发现。包含对所提供论文的引用。""",
    expected_output="一篇带有标题和引用来源的Markdown博客文章",
    agent=writer
)

这里有个bug：writing_task没有明确引用research_task的输出。CrewAI通过代理的内存隐式地传递上下文，但这并不可靠。我通过使用任务依赖来修复它：

writing_task = Task(
    description="""基于前一个任务的研究输出，
    撰写一篇500字的博客文章总结研究发现。
    研究输出为：{research_output}""",
    expected_output="一篇带有标题和引用来源的Markdown博客文章",
    agent=writer,
    context=[research_task]  # 明确依赖
)

context参数是一个任务列表，这些任务的输出会被注入到描述中。没有它，我的写作代理会虚构自己的研究内容。

第四步：运行团队（并处理失败）

现在创建一个Crew并运行它：

from crewai import Crew

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    verbose=True  # 调试时必不可少
)

result = crew.kickoff()
print(result)

当我第一次运行这个时，它成功了——但花了45秒，API调用费用为0.12美元。详细输出显示，研究员代理进行了3次独立的API调用来查找论文，然后写作代理又进行了2次调用来撰写文章。CrewAI的默认模式是顺序执行，意味着每个任务都要等待前一个任务完成。

缺陷： 如果第一个任务失败（例如API返回错误），整个团队就会崩溃。CrewAI没有内置的重试逻辑。我添加了一个简单的重试包装器：

import time

def safe_kickoff(crew, max_retries=3):
    for attempt in range(max_retries):
        try:
            return crew.kickoff()
        except Exception as e:
            print(f"第{attempt+1}次尝试失败：{e}")
            time.sleep(2 ** attempt)  # 指数退避
    raise Exception("团队在3次尝试后失败")

第五步：实际优化技巧

经过一周的测试，以下是我发现真正提高可靠性的方法：

限制代理内存：默认情况下，代理会记住整个对话。对于长任务，这会撑爆上下文窗口。对不需要历史记录的代理设置memory=False：
```
researcher = Researcher(memory=False)
```
使用allow_delegation=False：默认情况下，代理可以将任务委托给其他代理。这会造成循环。除非你在构建复杂层级结构，否则禁用它：
```
researcher = Researcher(allow_delegation=False)
```
缓存结果：CrewAI默认缓存大语言模型调用，但这是按代理进行的。如果你两次运行同一个团队，它会重复使用响应。这对调试很好，但在生产环境中很危险——你可能会提供过时数据。用以下方式禁用缓存：
```
crew = Crew(agents=[...], tasks=[...], cache=False)
```

监控token使用量：CrewAI不暴露token计数。我添加了一个简单的回调：

from langchain.callbacks import get_openai_callback
with get_openai_callback() as cb:
    result = crew.kickoff()
    print(f"总token数：{cb.total_tokens}，费用：${cb.total_cost}")

CrewAI没有告诉你的最大限制

在构建了一个用于内容生成的5代理系统后，我遇到了瓶颈：CrewAI没有内置的代理失败错误恢复机制。如果你的研究员代理返回无意义的内容，写作代理仍然会尝试使用它。唯一的修复方法是在任务描述中验证输出：

research_task = Task(
    description="""查找3篇论文。如果找不到3篇，输出'NO_RESULTS'
    并解释原因。不要编造论文。""",
    ...
)

然后在写作任务中检查这个哨兵值。这虽然有点hacky，但确实有效。

下一步：构建一个真实项目

不要从理论开始。克隆我的有问题的示例项目github.com/your-repo/crewai-blog-generator并修复故意设置的bug。README列出了我故意破坏的三件事：

写作任务缺少context参数
API失败没有重试逻辑
代理设置allow_delegation=True导致无限循环

修复这些问题，然后扩展系统，添加一个FactChecker代理来验证写作代理的引用。你在30分钟的调试中学到的东西会比阅读文档一小时还多。当你不可避免地搞砸某些事情时，记住：详细输出是你最好的朋友。将其设置为True，观察你的代理做出的每一个决定。

Getting started with CrewAI: a practical guide