Ασύγχρονες αναθέσεις

Χρησιμοποιήστε το start_workflows() όταν μια ροή εργασίας χρειάζεται να αναθέσει τις εργασίες που εντόπισε σε ανεξάρτητες ροές εργασίας ανώτατου επιπέδου, αντί να τις κρατήσει όλες μέσα σε μία εκτέλεση.

Αν μοντελοποιείτε ρητά εκτελέσεις agent, τα start_agents() και wait_for_agents() είναι σημασιολογικά ισοδύναμα ονόματα με προεπιλογές προσαρμοσμένες σε agent.

Για πιο ολοκληρωμένα παραδείγματα, δείτε την Ενορχήστρωση agent.

Αυτό είναι χρήσιμο για ενορχήστρωση τύπου agent:

• Μια ροή εργασίας σχεδιαστή εντοπίζει εργασίες κατά την εκτέλεση
• Κάθε εργασία γίνεται δική της μόνιμη ροή εργασίας
• Η γονική ροή αποθηκεύει τα αναγνωριστικά αυτών των ροών στην κατάσταση
• Η γονική ροή μπορεί αργότερα να περιμένει τα all, any, quorum ή μια επώνυμη ομάδα που έχει ξεκινήσει

Η μέθοδος ->start_workflows()

use Queuety\Enums\WaitMode;
use Queuety\Queuety;

$agent_task = Queuety::workflow( 'agent_task' )
    ->then( ResearchTopicStep::class )
    ->then( SummarizeTopicStep::class );

Queuety::workflow( 'brief_research' )
    ->then( PlanTopicsStep::class )
    ->start_workflows(
        items_key: 'topics',
        workflow_builder: $agent_task,
        result_key: 'child_workflow_ids',
        group_key: 'researchers',
    )
    ->wait_for_workflow_group( 'researchers', WaitMode::All, result_key: 'child_results' )
    ->then( SynthesizeBriefStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );

Σε αυτό το παράδειγμα:

1. Το PlanTopicsStep γράφει έναν πίνακα με ωφέλιμα φορτία θεμάτων στο $state['topics']
2. Το start_workflows() αποστέλλει μία ροή εργασίας agent_task ανά θέμα
3. Τα αναγνωριστικά των ροών εργασίας που ξεκίνησαν γράφονται στο $state['child_workflow_ids']
4. Το wait_for_workflow_group() σταματά μέχρι η ομάδα που ξεκίνησε να ικανοποιήσει την επιλεγμένη συνθήκη ολοκλήρωσης
5. Το SynthesizeBriefStep λαμβάνει τις καταστάσεις των ολοκληρωμένων θυγατρικών ροών στην $state['child_results']

Ωφέλιμα φορτία θυγατρικών ροών

Κάθε στοιχείο του items_key γίνεται το αρχικό ωφέλιμο φορτίο μίας ροής εργασίας που ξεκινά.

Αν το στοιχείο είναι πίνακας, τα δημόσια κλειδιά του συγχωνεύονται με το ωφέλιμο φορτίο της θυγατρικής ροής:

[
    [ 'topic' => 'pricing', 'source' => 'competitors' ],
    [ 'topic' => 'reviews', 'source' => 'customers' ],
]

Αν το στοιχείο είναι βαθμωτή τιμή, το Queuety την αποθηκεύει στο payload_key:

->start_workflows(
    items_key: 'topic_names',
    workflow_builder: $agent_task,
    payload_key: 'topic',
)

Κληρονόμηση της γονικής κατάστασης

Από προεπιλογή, το start_workflows() συγχωνεύει τη δημόσια κατάσταση της γονικής ροής εργασίας με το ωφέλιμο φορτίο κάθε θυγατρικής ροής.

Έτσι, κοινόχρηστο περιεχόμενο όπως τα brief_id, account_id ή campaign είναι αυτόματα διαθέσιμο σε κάθε ροή εργασίας που ξεκινά.

Η πηγή items_key και το result_key εξαιρούνται από την κληρονόμηση, ώστε κάθε θυγατρική ροή να μη λαμβάνει ολόκληρη τη λίστα εργασιών ή τον τελικό πίνακα αναγνωριστικών ροών εργασίας.

Ορίστε το inherit_state: false όταν κάθε θυγατρική ροή πρέπει να λαμβάνει μόνο το ωφέλιμο φορτίο του στοιχείου της.

Γιατί όχι το run_workflow();

Το run_workflow() δημιουργεί σχέση γονικής και θυγατρικής ροής, όπου η γονική ροή παραμένει σε αναμονή σε αυτό το βήμα μέχρι να ολοκληρωθεί η θυγατρική.

Το start_workflows() είναι πιο χαλαρό:

• Οι ροές εργασίας που ξεκίνησαν είναι ανεξάρτητες ροές ανώτατου επιπέδου
• Κάθε ροή εργασίας που ξεκίνησε μπορεί να επιθεωρηθεί, να επαναληφθεί ή να εξαχθεί ανεξάρτητα
• Η γονική ροή αποφασίζει αργότερα αν και πώς θα περιμένει αυτές τις ροές

Έτσι, το start_workflows() ταιριάζει καλύτερα στην ασύγχρονη ανάθεση εργασιών agent, ενώ το run_workflow() παραμένει καταλληλότερο για ένθετη σύνθεση μέσα σε ένα δέντρο ενορχήστρωσης.

Αναμονή σε μεταγενέστερο στάδιο

Το start_workflows() έχει σχεδιαστεί ώστε να συνδυάζεται με τις Εξαρτήσεις ροών εργασίας:

use Queuety\Enums\WaitMode;

Queuety::workflow( 'review_board' )
    ->start_workflows( 'review_tasks', $review_task_workflow, 'review_workflow_ids' )
    ->wait_for_workflows( 'review_workflow_ids', WaitMode::Any, 'winner' )
    ->then( FinalizeDecisionStep::class )
    ->dispatch( $payload );

• Το WaitMode::All συλλέγει όλες τις καταστάσεις των ολοκληρωμένων θυγατρικών ροών
• Το WaitMode::Any συνεχίζει με την πρώτη επιτυχημένη θυγατρική ροή εργασίας

Έτσι μπορείτε να ενορχηστρώνετε σχεδιαστή και εκτελεστή χωρίς να χρειάζεστε πλήρη μηχανή DAG που να μπορεί να επεξεργάζεται κατά τον χρόνο εκτέλεσης.

Ψευδώνυμα φιλικά προς agent

Το start_agents() είναι ψευδώνυμο του start_workflows() με προεπιλογές προσαρμοσμένες σε ροές σχεδιαστή/εκτελεστή:

• Το result_key έχει ως προεπιλογή το agent_workflow_ids
• Το payload_key έχει ως προεπιλογή το agent_task
• Το name έχει ως προεπιλογή το start_agents

Το wait_for_agents() είναι ψευδώνυμο του wait_for_workflows() με προεπιλογές προσαρμοσμένες στη συλλογή αποτελεσμάτων agent:

• Το workflows έχει ως προεπιλογή το agent_workflow_ids
• Το result_key έχει ως προεπιλογή το agent_results
• Το name έχει ως προεπιλογή το wait_for_agents

Το wait_for_agent_group() είναι ψευδώνυμο του wait_for_workflow_group() με προεπιλογές προσαρμοσμένες σε δέσμες agent.

use Queuety\Enums\WaitMode;

$agent_run = Queuety::workflow( 'agent_run' )
    ->then( ResearchTopicStep::class )
    ->then( SummarizeTopicStep::class );

Queuety::workflow( 'brief_research' )
    ->then( PlanTopicsStep::class )
    ->start_agents( 'agent_tasks', $agent_run, group_key: 'researchers' )
    ->wait_for_agent_group( 'researchers', WaitMode::Quorum, 2 )
    ->then( SynthesizeBriefStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );