퀵스타트

{
  "role": "{topic} 시니어 데이터 리서처",
  "goal": "{topic} 분야의 최신 동향을 파악한다",
  "backstory": "당신은 가장 관련성 높은 정보를 찾아 명확하게 전달하는 연구원입니다.",
  "tools": ["SerperDevTool"],
  "settings": {
    "verbose": true
  }
}

{
  "name": "Research Crew",
  "agents": ["researcher"],
  "tasks": [
    {
      "name": "research_task",
      "description": "{topic}에 대해 철저히 조사하세요. 웹 검색으로 최신이고 신뢰할 수 있는 정보를 찾으세요.",
      "expected_output": "마크다운 보고서로, 주요 트렌드·주목할 도구나 기업·시사점 등으로 섹션을 나누세요. 분량은 약 800~1200단어. 문서 전체를 코드 펜스로 감싸지 마세요.",
      "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"주제: {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 실행 완료.")

  @listen(run_research)
  def summarize(self):
    print("보고서 경로: output/report.md")


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


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


if __name__ == "__main__":
  kickoff()