快速入门

# src/latest_ai_flow/crews/content_crew/config/agents.yaml
researcher:
  role: >
    {topic} Senior Data Researcher
  goal: >
    Uncover cutting-edge developments in {topic}
  backstory: >
    You're a seasoned researcher with a knack for uncovering the latest
    developments in {topic}. You find the most relevant information and
    present it clearly.

# src/latest_ai_flow/crews/content_crew/config/tasks.yaml
research_task:
  description: >
    Conduct thorough research about {topic}. Use web search to find current,
    credible information. The current year is 2026.
  expected_output: >
    A markdown report with clear sections: key trends, notable tools or companies,
    and implications. Aim for 800–1200 words. No fenced code blocks around the whole document.
  agent: researcher
  output_file: output/report.md

# src/latest_ai_flow/crews/content_crew/content_crew.py
from typing import List

from crewai import Agent, Crew, Process, Task
from crewai.agents.agent_builder.base_agent import BaseAgent
from crewai.project import CrewBase, agent, crew, task
from crewai_tools import SerperDevTool


@CrewBase
class ResearchCrew:
  """Single-agent research crew used inside the Flow."""

  agents: List[BaseAgent]
  tasks: List[Task]

  agents_config = "config/agents.yaml"
  tasks_config = "config/tasks.yaml"

  @agent
  def researcher(self) -> Agent:
    return Agent(
      config=self.agents_config["researcher"],  # type: ignore[index]
      verbose=True,
      tools=[SerperDevTool()],
    )

  @task
  def research_task(self) -> Task:
    return Task(
      config=self.tasks_config["research_task"],  # type: ignore[index]
    )

  @crew
  def crew(self) -> Crew:
    return Crew(
      agents=self.agents,
      tasks=self.tasks,
      process=Process.sequential,
      verbose=True,
    )

# src/latest_ai_flow/main.py
from pydantic import BaseModel

from crewai.flow import Flow, listen, start

from latest_ai_flow.crews.content_crew.content_crew import ResearchCrew


class ResearchFlowState(BaseModel):
  topic: str = ""
  report: str = ""


class LatestAiFlow(Flow[ResearchFlowState]):
  @start()
  def prepare_topic(self, crewai_trigger_payload: dict | None = None):
    if crewai_trigger_payload:
      self.state.topic = crewai_trigger_payload.get("topic", "AI Agents")
    else:
      self.state.topic = "AI Agents"
    print(f"Topic: {self.state.topic}")

  @listen(prepare_topic)
  def run_research(self):
    result = ResearchCrew().crew().kickoff(inputs={"topic": self.state.topic})
    self.state.report = result.raw
    print("Research crew finished.")

  @listen(run_research)
  def summarize(self):
    print("Report path: output/report.md")


def kickoff():
  LatestAiFlow().kickoff()


def plot():
  LatestAiFlow().plot()


if __name__ == "__main__":
  kickoff()