垦荒学园
  • 中文

      • 快速上手

      两条路并行：纯 Python 跑一个 Crew（理解对象），以及 Flow + Crew（官方生产入口）。CLI 脚手架见 安装；本章代码可直接放进脚本。

      先配置 .env：OPENAI_API_KEY。网页搜索示例还需要 SERPER_API_KEY（serper.dev）。

      最小 Crew（Python）

      from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Research Analyst",
    goal="Find accurate, current facts about the given topic",
    backstory="You are a careful researcher who cites clear evidence.",
    verbose=True,
)

task = Task(
    description="List 5 concise facts about {topic}. Prefer recent, checkable claims.",
    expected_output="A markdown bullet list of exactly 5 facts.",
    agent=researcher,
)

crew = Crew(
    agents=[researcher],
    tasks=[task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "CrewAI Flows"})
print(result.raw)

      发生了什么： kickoff 按 Process.sequential 执行任务列表；Agent 根据 role / goal / backstory 推理；{topic} 由 inputs 注入。返回值是 CrewOutput，常用 result.raw。

      加上搜索工具

      from crewai_tools import SerperDevTool

researcher = Agent(
    role="Web Researcher",
    goal="Uncover current developments using web search",
    backstory="You search first, then summarize with sources.",
    tools=[SerperDevTool()],
    verbose=True,
)

      需 pip install 'crewai[tools]' 与 SERPER_API_KEY。更多工具见 Tools。

      最小 Flow：@start 与 @listen

      官方确认的 Flow 装饰器包括 @start()（入口）和 @listen(...)（监听上一步完成）。生产应用把状态放在 Flow 上，把自主协作放进 Crew 步骤。

      from pydantic import BaseModel
from crewai import Agent, Task, Crew, Process
from crewai.flow.flow import Flow, listen, start

class ResearchState(BaseModel):
    topic: str = "AI Agents"
    report: str = ""

def build_crew() -> Crew:
    agent = Agent(
        role="{topic} researcher",
        goal="Write a short briefing on {topic}",
        backstory="You write clearly and avoid hype.",
    )
    task = Task(
        description="Write a 200-word briefing about {topic}.",
        expected_output="Plain markdown, no fenced document wrapper.",
        agent=agent,
    )
    return Crew(agents=[agent], tasks=[task], process=Process.sequential)

class BriefingFlow(Flow[ResearchState]):
    @start()
    def prepare_topic(self):
        self.state.topic = "AI Agents"
        print(f"Topic: {self.state.topic}")

    @listen(prepare_topic)
    def run_research(self):
        result = build_crew().kickoff(inputs={"topic": self.state.topic})
        self.state.report = result.raw
        return result.raw

flow = BriefingFlow()
print(flow.kickoff())

      Flow[ResearchState] 用 Pydantic 做结构化状态；也可用 dict。kickoff() 的返回值是最后一个完成的方法的返回值。flow.plot() 可生成流程图 HTML。

      CLI 等价：crewai create flow latest-ai-flow，在 crew.jsonc 里配 Agent，用 from crewai.project import load_crew 加载后在 @listen 里 crew.kickoff(...)。详见官方 Quickstart。

      直接对话单个 Agent

      没有任务列表时可用 Agent.kickoff()：

      from crewai import Agent

agent = Agent(
    role="Tutor",
    goal="Explain CrewAI concepts briefly",
    backstory="You teach with short examples.",
)
out = agent.kickoff("What is the difference between a Flow and a Crew?")
print(out.raw)

      第一个任务清单

      1. 安装 crewai（或 uv tool install crewai）并设置 OPENAI_API_KEY
      2. 跑通最小 Crew.kickoff
      3. 有 Key 再加 SerperDevTool
      4. 把同一 Crew 放进 @start / @listen Flow
      5. 再学 Agent 与 Crew 与 Process

      下一步

      评论