Guia Rápido

{
  "role": "Pesquisador(a) Sênior de Dados em {topic}",
  "goal": "Descobrir os desenvolvimentos mais recentes em {topic}",
  "backstory": "Você é um pesquisador experiente que encontra as informações mais relevantes e apresenta tudo com clareza.",
  "tools": ["SerperDevTool"],
  "settings": {
    "verbose": true
  }
}

{
  "name": "Research Crew",
  "agents": ["researcher"],
  "tasks": [
    {
      "name": "research_task",
      "description": "Faça uma pesquisa aprofundada sobre {topic}. Use busca na web para obter informações recentes e confiáveis.",
      "expected_output": "Um relatório em markdown com seções claras: tendências principais, ferramentas ou empresas relevantes e implicações. Entre 800 e 1200 palavras. Sem cercas de código em volta do documento inteiro.",
      "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"Tópico: {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("Crew de pesquisa concluído.")

  @listen(run_research)
  def summarize(self):
    print("Relatório em: output/report.md")


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


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


if __name__ == "__main__":
  kickoff()