إتقان إدارة حالة Flow

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

class UnstructuredStateFlow(Flow):
    @start()
    def initialize_data(self):
        self.state["user_name"] = "Alex"
        self.state["preferences"] = {"theme": "dark", "language": "English"}
        self.state["items"] = []
        return "Initialized"

    @listen(initialize_data)
    def process_data(self, previous_result):
        user = self.state["user_name"]
        self.state["items"].append("item1")
        self.state["processed"] = True
        return "Processed"

flow = UnstructuredStateFlow()
result = flow.kickoff()

from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel, Field
from typing import List, Dict, Optional

class AppState(BaseModel):
    user_name: str = ""
    items: List[str] = []
    processed: bool = False
    completion_percentage: float = 0.0

class StructuredStateFlow(Flow[AppState]):
    @start()
    def initialize_data(self):
        self.state.user_name = "Taylor"
        return "Initialized"

    @listen(initialize_data)
    def process_data(self, previous_result):
        self.state.items.append("item1")
        self.state.processed = True
        self.state.completion_percentage = 50.0
        return "Processed"

flow = StructuredStateFlow()
result = flow.kickoff()

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

class CounterState(BaseModel):
    value: int = 0

@persist()
class PersistentCounterFlow(Flow[CounterState]):
    @start()
    def increment(self):
        self.state.value += 1
        return self.state.value

    @listen(increment)
    def double(self, value):
        self.state.value = value * 2
        return self.state.value

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

class PaymentState(BaseModel):
    amount: float = 0.0
    is_approved: bool = False
    retry_count: int = 0

class PaymentFlow(Flow[PaymentState]):
    @start()
    def process_payment(self):
        self.state.amount = 100.0
        self.state.is_approved = self.state.amount < 1000
        return "Payment processed"

    @router(process_payment)
    def check_approval(self, previous_result):
        if self.state.is_approved:
            return "approved"
        elif self.state.retry_count < 3:
            return "retry"
        else:
            return "rejected"

    @listen("approved")
    def handle_approval(self):
        return f"Payment of ${self.state.amount} approved!"

    @listen("retry")
    def handle_retry(self):
        self.state.retry_count += 1
        return "Retry initiated"

    @listen("rejected")
    def handle_rejection(self):
        return f"Payment of ${self.state.amount} rejected after {self.state.retry_count} retries."