Flows

from crewai.flow.flow import Flow, listen, start
from dotenv import load_dotenv
from litellm import completion


class ExampleFlow(Flow):
    model = "gpt-4o-mini"

    @start()
    def generate_city(self):
        print("Flow 시작")
        # 각 Flow 상태는 자동으로 고유 ID를 가짐
        print(f"Flow 상태 ID: {self.state['id']}")

        response = completion(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": "세계에서 무작위 도시 이름 하나를 반환하세요.",
                },
            ],
        )

        random_city = response["choices"][0]["message"]["content"]
        # 도시를 상태에 저장
        self.state["city"] = random_city
        print(f"무작위 도시: {random_city}")

        return random_city

    @listen(generate_city)
    def generate_fun_fact(self, random_city):
        response = completion(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": f"{random_city}에 대한 재미있는 사실을 알려주세요.",
                },
            ],
        )

        fun_fact = response["choices"][0]["message"]["content"]
        # 재미있는 사실을 상태에 저장
        self.state["fun_fact"] = fun_fact
        return fun_fact



flow = ExampleFlow()
result = flow.kickoff()

print(f"생성된 재미있는 사실: {result}")

from crewai.flow.flow import Flow, listen, start

class UnstructuredExampleFlow(Flow):

    @start()
    def first_method(self):
        # 상태에 자동으로 'id' 필드가 포함된다
        print(f"State ID: {self.state['id']}")
        self.state['counter'] = 0
        self.state['message'] = "Hello from structured flow"

    @listen(first_method)
    def second_method(self):
        self.state['counter'] += 1
        self.state['message'] += " - updated"

    @listen(second_method)
    def third_method(self):
        self.state['counter'] += 1
        self.state['message'] += " - updated again"

        print(f"State after third_method: {self.state}")


flow = UnstructuredExampleFlow()
flow.kickoff()

from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel


class ExampleState(BaseModel):
    # 참고: 모든 상태에 'id' 필드가 자동으로 추가됨
    counter: int = 0
    message: str = ""


class StructuredExampleFlow(Flow[ExampleState]):

    @start()
    def first_method(self):
        # 필요 시 자동 생성된 ID에 접근
        print(f"State ID: {self.state.id}")
        self.state.message = "Hello from structured flow"

    @listen(first_method)
    def second_method(self):
        self.state.counter += 1
        self.state.message += " - updated"

    @listen(second_method)
    def third_method(self):
        self.state.counter += 1
        self.state.message += " - updated again"

        print(f"State after third_method: {self.state}")


flow = StructuredExampleFlow()
flow.kickoff()

@persist  # 기본적으로 SQLiteFlowPersistence 사용
class MyFlow(Flow[MyState]):
    @start()
    def initialize_flow(self):
        # 이 메서드는 자동으로 상태가 유지됨
        self.state.counter = 1
        print("플로우 초기화 완료. 상태 ID:", self.state.id)

    @listen(initialize_flow)
    def next_step(self):
        # 상태(self.state.id 포함)가 자동으로 다시 로드됨
        self.state.counter += 1
        print("플로우 상태가 유지됨. 카운터:", self.state.counter)

class AnotherFlow(Flow[dict]):
    @persist  # 이 메서드의 상태만 지속
    @start()
    def begin(self):
        if "runs" not in self.state:
            self.state["runs"] = 0
        self.state["runs"] += 1
        print("메서드 단위로 지속된 실행 횟수:", self.state["runs"])

crewai create flow name_of_flow

#!/usr/bin/env python
from random import randint

from pydantic import BaseModel
from crewai.flow.flow import Flow, listen, start
from .crews.poem_crew.poem_crew import PoemCrew

class PoemState(BaseModel):
    sentence_count: int = 1
    poem: str = ""

class PoemFlow(Flow[PoemState]):

    @start()
    def generate_sentence_count(self):
        print("문장 수 생성 중")
        self.state.sentence_count = randint(1, 5)

    @listen(generate_sentence_count)
    def generate_poem(self):
        print("시 생성 중")
        result = PoemCrew().crew().kickoff(inputs={"sentence_count": self.state.sentence_count})

        print("시 생성 완료", result.raw)
        self.state.poem = result.raw

    @listen(generate_poem)
    def save_poem(self):
        print("시 저장 중")
        with open("poem.txt", "w") as f:
            f.write(self.state.poem)

def kickoff():
    poem_flow = PoemFlow()
    poem_flow.kickoff()


def plot():
    poem_flow = PoemFlow()
    poem_flow.plot()

if __name__ == "__main__":
    kickoff()

crewai install

source .venv/bin/activate

crewai flow kickoff

uv run kickoff

# 플로우 인스턴스가 있다고 가정
flow.plot("my_flow_plot")

crewai flow plot