CrewAI 多智能体协作

CrewAI 是最流行的"角色分工"式多智能体框架之一：把任务拆给多个各司其职的 Agent（如研究员、写手），由 Crew 编排执行顺序，最后一个任务的结果就是整支"团队"的产出。它主打开箱即用——你不用自己写 ReAct 循环，定义好角色和任务即可跑，适合内容生产、调研、批量处理等流水线场景。

安装与配置

Python 3.10+，安装后设置模型密钥：

pip install crewai            # 校对版本 1.x（2025 起从 0.x 大版本演进，接口仍在快速迭代）
$env:OPENAI_API_KEY = "sk-..."   # 默认用 OpenAI；换用 deepseek 见文末

三个核心概念

Agent（角色）：由 role（角色名）、goal（目标）、backstory（背景人设）三段式定义，模型据此扮演专家；可挂 tools/llm；
Task（任务）：description 描述做什么、expected_output 描述交付物长什么样，可绑定某个 agent；上一任务输出可作为下一任务上下文；
Crew（团队）：把 agents + tasks 组装起来，用 Process.sequential（顺序执行）或 Process.hierarchical（经理调度）驱动，kickoff() 开跑。

最小"研究员 + 写手" Crew

先让"研究员"写资料，再把资料交给"写手"出稿——顺序执行、天然串联：

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="技术研究员",
    goal="查找并归纳某主题的核心要点，输出有条理的素材",
    backstory="你是一名严谨的科技编辑，擅长把英文资料翻译整理成中文要点。",
)
writer = Agent(
    role="中文技术写手",
    goal="基于素材写一段 200 字左右、通俗易懂的中文介绍",
    backstory="你擅长把专业内容讲得让初学者也能看懂。",
)

research_task = Task(
    description="调研「向量数据库」是什么、为什么需要它，输出 3~5 个要点素材",
    expected_output="中文要点列表，每条一句话，包含术语解释",
    agent=researcher,
)
write_task = Task(
    description="根据研究员提供的素材，写一段面向初学者的中文介绍",
    expected_output="约 200 字的一段话，语气轻松、有开头有结尾",
    agent=writer,
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,   # 顺序执行：先研究、后写作
    verbose=True,                 # 控制台打印每个 agent 的思考与动作
)

result = crew.kickoff()
print("=== 最终产出 ===")
print(result)     # 输出：一段约 200 字的中文介绍（内容依模型而定，示意略）

运行时会看到 verbose 输出：[研究员] Task: 调研… → 模型思考 → 产出要点 → [写手] Task: 根据素材… → 最终段落。

怎么读 kickoff 的返回

kickoff() 返回 CrewOutput 对象，常用属性：result.raw（最终文本）、result.tasks_output（每个任务的输出列表）、result.usage（token 用量，算成本用）：

print(result.raw)                       # 最终产出文本
for t in result.tasks_output:
    print(t.agent, "->", t.raw[:60])    # 输出：技术研究员 -> 1. 向量数据库是……（示意）

不同小版本的属性名偶有调整，跑前 print(type(result)) 或看官方 docs 的 CrewOutput 说明即可。

任务参数与结果传递

结果传递：sequential 模式下，前一个任务的输出会自动作为下一个任务的上下文，所以"研究员→写手"天然串联，无需手动传参；
参数化：任务描述里写 {占位符}，启动时用 kickoff(inputs=...) 传值，同一个 Crew 可复用给不同主题：

research_task = Task(
    description="调研「{topic}」是什么、为什么需要，输出 3~5 个中文要点",
    expected_output="中文要点列表",
    agent=researcher,
)

crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task],
            process=Process.sequential)
result = crew.kickoff(inputs={"topic": "向量数据库"})   # 换主题只改这里

跑挂排查三板斧：verbose=True 看每一步输出、检查 OPENAI_API_KEY/模型是否配好、把 expected_output 写具体（它决定任务"做到什么程度算完成"）。

hierarchical：经理调度模式

任务不预先绑定人，而是交给"经理 Agent"现场派活。经理通常用更强模型，成本更高：

manager = Agent(
    role="项目经理",
    goal="把用户问题拆解后分派给合适的专家，并汇总最终答案",
    backstory="你擅长调度多个专家协作。",
)
t1 = Task(description="调研「向量数据库」是什么，输出中文要点", expected_output="要点列表")
t2 = Task(description="基于要点写一段面向初学者的中文介绍", expected_output="约 200 字短文")
crew2 = Crew(
    agents=[researcher, writer],
    tasks=[t1, t2],                        # hierarchical 下任务不预绑 agent
    process=Process.hierarchical,          # 经理先规划，再逐人派活
    manager_agent=manager,                 # 也可用 manager_llm 指定经理模型
)
print(crew2.kickoff().raw)                 # 输出：经理汇总后的答案（内容依模型而定）

让 Crew 使用 deepseek 等模型

默认读 OPENAI_API_KEY 走 OpenAI；CrewAI 通过"provider/模型名"前缀识别其他厂商（底层经 litellm 路由）。按官方 llm-connections 文档配置，例如：

import os
from crewai import Agent, LLM

llm = LLM(
    model="deepseek/deepseek-chat",      # provider/model 前缀写法
    base_url="https://api.deepseek.com", # 有的厂商需要显式给网关地址
    api_key=os.getenv("DEEPSEEK_API_KEY"),
)
researcher_deepseek = Agent(
    role="技术研究员",
    goal="调研并输出中文要点",
    backstory="资深科技编辑。",
    llm=llm,                             # 显式指定模型
)

注意：工具调用、结构化输出等能力是否可用取决于所选模型与兼容端点；官方仓库 example 与 docs 更新频繁，CrewAI 1.x 之后又引入了 Flow（装饰器式事件编排），本系列以最稳定的 Agent/Task/Crew 经典用法为主，进阶用法去 docs.crewai.com 查阅。小结：CrewAI 用 role/goal/backstory 定义角色、Task 定义交付物、Crew 按 sequential 或 hierarchical 编排，kickoff() 一次跑完整条流水线，读 result.raw / result.tasks_output 取产出；模型可用 LLM(model="厂商/模型") 切换，细节以官方文档为准。

笔记加载中…

标签：#Agent实战

返回知识库