Quickstart

{
  "role": "{topic} Senior Data Researcher",
  "goal": "Uncover cutting-edge developments in {topic}",
  "backstory": "You're a seasoned researcher who finds relevant information and presents it clearly.",
  "tools": ["SerperDevTool"],
  "settings": {
    "verbose": true
  }
}

{
  "name": "Research Crew",
  "agents": ["researcher"],
  "tasks": [
    {
      "name": "research_task",
      "description": "Conduct thorough research about {topic}. Use web search to find recent, credible information.",
      "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",
      "markdown": true
    }
  ],
  "process": "sequential",
  "verbose": true
}

# src/latest_ai_flow/crews/content_crew/content_crew.py
from pathlib import Path

from crewai.project import load_crew


def kickoff_content_crew(inputs: dict):
  crew, default_inputs = load_crew(Path(__file__).with_name("crew.jsonc"))
  return crew.kickoff(inputs={**default_inputs, **inputs})

# 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 kickoff_content_crew


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 = kickoff_content_crew(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()