Kenhuang Academy
  • English

      • Quick Start

      Two tracks: a Python-only Crew (learn the objects) and a Flow + Crew (the official production entry). CLI scaffolding is in Installation; the snippets below run as scripts.

      Set .env: OPENAI_API_KEY. Web-search examples also need SERPER_API_KEY (serper.dev).

      Minimal 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)

      What happens: kickoff runs the task list with Process.sequential. The agent reasons from role / goal / backstory. {topic} comes from inputs. The return value is a CrewOutput; use result.raw.

      Add a search tool

      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,
)

      Requires pip install 'crewai[tools]' and SERPER_API_KEY. More in Tools.

      Minimal Flow: @start and @listen

      Official Flow decorators include @start() (entry points) and @listen(...) (run after another method completes). Production apps keep state on the Flow and put autonomous work in a Crew step.

      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] uses Pydantic for structured state; a dict also works. kickoff() returns the last completed method’s return value. flow.plot() writes an HTML diagram.

      CLI equivalent: crewai create flow latest-ai-flow, configure agents in crew.jsonc, load with from crewai.project import load_crew, then crew.kickoff(...) inside @listen. See the official Quickstart.

      Talk to one agent

      With no task list, use 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)

      Checklist

      1. Install crewai (or uv tool install crewai) and set OPENAI_API_KEY
      2. Run the minimal Crew.kickoff
      3. Add SerperDevTool if you have a Serper key
      4. Wrap the same Crew in a @start / @listen Flow
      5. Continue with Agents and Crews and Processes

      Next steps

      评论