Ga naar de inhoud

Inleiding

Bekijken als Markdown Bewerken op GitHub Documentatie op 2cc7b2b

Queuety is een WordPress-plugin met een snelle taakwachtrij, een duurzame workflow-engine en een duurzame runtime voor statemachines. Workers claimen taken rechtstreeks uit MySQL en verwerken ze in een langlopend WP-CLI-proces. De runtime van PHP die WordPress en WP-CLI laadt, heeft mysqli of pdo_mysql nodig.

Waarom Queuety?

WordPress heeft geen echt achtergrondsysteem voor taken. wp_cron wordt alleen uitgevoerd bij paginabezoeken. De standaarduitvoerder van Action Scheduler draait op WP-Cron en loopback-HTTP-verzoeken. Elk verzoek verwerkt batches gedurende maximaal 30 seconden. Een API-aanroep van een LLM die 60 seconden duurt, wordt afgebroken door de time-out van 30 seconden van PHP. Er is geen manier om processen met meerdere stappen uit te voeren die na crashes blijven bestaan en vanaf hun vorige punt verdergaan.

Queuety lost dit op met drie basisfuncties:

  1. Taken voor achtergrondwerk dat u verstuurt zonder op het resultaat te wachten
  2. Workflows voor duurzame processen met meerdere stappen en blijvende status
  3. Statemachines voor langdurige, gebeurtenisgestuurde status van processen

Snel voorbeeld

Verstuur een taak met de moderne API:

PHP
use Queuety\Contracts\Job;
use Queuety\Dispatchable;

readonly class SendEmailJob implements Job {
    use Dispatchable;

    public function __construct(
        public string $to,
        public string $subject,
        public string $body,
    ) {}

    public function handle(): void {
        wp_mail( $this->to, $this->subject, $this->body );
    }
}

SendEmailJob::dispatch( 'user@example.com', 'Welcome', 'Hello from Queuety!' );

Of gebruik de klassieke aanpak met een handlernaam:

PHP
use Queuety\Queuety;

Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] );

Voer een duurzame workflow uit:

PHP
Queuety::workflow( 'generate_report' )
    ->then( FetchDataHandler::class )
    ->then( CallLLMHandler::class )
    ->then( FormatOutputHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

Modelleer een langdurige sessie met een statemachine:

PHP
use Queuety\Enums\StateMachineStatus;

Queuety::machine( 'agent_session' )
    ->state( 'awaiting_user' )
    ->on( 'user_message', 'planning' )
    ->state( 'planning' )
    ->action( PlanSessionAction::class )
    ->on( 'planned', 'completed' )
    ->state( 'completed', StateMachineStatus::Completed )
    ->dispatch( [ 'thread_id' => 42 ] );

Start een worker:

Shell
wp queuety work

Begint u aan een systeem met agents, start dan met Agentorkestratie. U ziet er hoe planner- en uitvoeringsprocessen, controlepunten voor menselijke beoordeling en wachttijden tussen workflows samenwerken.

Is het systeem gebeurtenisgestuurd en langdurig actief, begin dan naast workflows met Statemachines.

Zie Voorbeelden voor integratievoorbeelden van begin tot eind. Deze bevatten nu patronen op basis van Neuron voor het wisselen tussen providers, streaming, herhaalde beoordelingen met geheugen en afhankelijke workflows.

Zie Actietriggers als u workflows wilt starten vanuit domeingebeurtenissen van WordPress, zoals save_post.

Zie Zelfstandig gebruik als u de op een database gebaseerde runtime buiten WordPress wilt gebruiken.

Belangrijkste onderscheidende kenmerken

  • Snelle uitvoering. Workers gebruiken directe toegang tot de MySQL-wachtrij en vermijden het cron-patroon per verzoek dat achtergrondwerk in WordPress vertraagt.
  • Duurzame workflows. Processen met meerdere stappen en persistente status die PHP-time-outs, crashes en retries overleven. Elke stapgrens is een atomaire MySQL-transactie.
  • Expliciete toestandsmachines. Modelleer chatsessies, goedkeuringen en andere langdurige levenscycli met benoemde statussen, beveiligde overgangen, wachtrijacties bij statusovergangen en duurzame gebeurtenistijdlijnen.
  • Moderne dispatch API. Zelfstandige jobklassen met de Dispatchable-trait, getypeerde constructorproperties als payload en $tries/$timeout/$backoff gedeclareerd in de klasse. Een readonly-klasse kan deze properties geen standaardwaarden geven; declareer ze daarom in een gewone klasse met readonly gepromoveerde constructorproperties.
  • Middleware-pipeline. Onion-style middleware voor snelheidsbeperking, throttling van uitzonderingen, unieke jobs en aangepaste logica. Per job gedeclareerd via een middleware()-methode.
  • Streamingstappen. De StreamingStep-interface levert chunks op die direct in de database worden opgeslagen. Bij een retry worden eerder opgeslagen chunks meegegeven, zodat de stream kan worden hervat waar die was gebleven. Ideaal voor LLM-antwoorden en grote downloads.
  • Uitbreidbare cachelaag. CacheFactory detecteert APCu automatisch of valt terug op een cache in het geheugen. Aangepaste backends implementeren Contracts\Cache. Intern gebruikt voor rate-limiter-zoekopdrachten en wachtrijstatus.
  • Batchverwerking en ketens. Dispatch groepen jobs met then/catch/finally-callbacks en voortgangsregistratie, of koppel jobs sequentieel aan elkaar, waarbij elke job afhankelijk is van de voltooiing van de vorige.
  • Duurzame vertragingen en wachttijden. delay() voegt vertragingen toe die bestand zijn tegen procesherstarts. Met wait_for_signal(), wait_for_signals(), wait_for_approval(), wait_for_input() en wait_for_workflow() kunnen workflows wachten op externe gebeurtenissen of andere workflows zonder PHP actief te houden.
  • Dynamische for-each. for_each() breidt tijdens runtime ontdekte vertakkingen uit op basis van de workflowstatus en ondersteunt voltooiingsmodi voor alles geslaagd, eerste geslaagd en quorum.
  • Asynchrone workflowoverdracht. start_workflows() zet tijdens runtime ontdekte items om in zelfstandige workflows op hoofdniveau waarop later kan worden gewacht.
  • WordPress-actietriggers. on_action() koppelt core- of plug-inactiehooks aan duurzame workflowdispatches, met gating en idempotentie.
  • Duurzame artefacten. Sla benoemde workflow-uitvoer buiten de hoofdstatus op en neem die mee in status, CLI-inspectie, export en replay.
  • Workflowbeveiligingen. version(), een deterministische definitiehash, idempotente dispatchsleutels en workflowbudgetten maken agentachtige uitvoeringen eenvoudiger te inspecteren en verkleinen de kans op duplicatie of onopgemerkt onbeperkt doorlopen.
  • Resourcebeheer. Gelijktijdigheidsgroepen, kostenunits, workflowbudgetten en toelating van workers helpen duur agentachtig werk binnen een vastgestelde grens te houden.
  • Compensatie van stappen. compensate_with() en compensate_on_failure() bieden rollback-hooks voor voltooide stappen wanneer workflows worden geannuleerd of mislukken.
  • Heartbeats. Langlopende stappen versturen heartbeats om te voorkomen dat het herstelproces voor verouderde jobs actief werk terugvordert.
  • Workflowtracering. Elke stapovergang wordt vastgelegd met invoer, uitvoer, status vóór en na, context, artefacten, chunks, fouten en debugging door de tijd heen via workflow_state_at().
  • Parallelle stappen. Voer stappen gelijktijdig uit en wacht tot ze allemaal zijn voltooid voordat u verdergaat.
  • Voorwaardelijke vertakkingen. Sla op basis van eerdere status over naar benoemde stappen met _next_step.
  • Workflows uitvoeren. Start onderliggende workflows die resultaten teruggeven aan de bovenliggende workflow.
  • Prioriteitswachtrijen. Vier niveaus (Low, Normal, High, Urgent) via typesafe PHP-enums.
  • Workers voor meerdere wachtrijen. Verwerk meerdere wachtrijen met strikte prioriteitsvolgorde in één worker.
  • Snelheidsbeperking. Beperk het aantal uitvoeringen per handler met een schuivend tijdvenster.
  • Terugkerende jobs. Planning op basis van intervallen en cron, met instelbaar beleid voor overlap (Allow, Skip, Buffer).
  • Workflows annuleren. Annuleer lopende workflows en activeer geregistreerde opruimhandlers.
  • Status opschonen. Verwijder automatisch oude stapuitvoer om grote workflowstatussen beheersbaar te houden.
  • Gelijktijdigheid van workers. --workers=N start meerdere processen met automatische herstart na een crash.
  • Volledige waarneembaarheid. Database-logging, statistieken per handler en gebeurteniswebhooks.
  • Testhulpmiddelen. QueueFake registreert gedispatchte jobs en batches voor beweringen in unittests.
  • PHP-attributen. #[QueuetyHandler('name')] voor registratie van handlers zonder configuratie.

Werking

Workers draaien binnen een langdurig WP-CLI-proces en halen jobs rechtstreeks op uit MySQL. Wachtrijrijen, workflowstatus, logs, batches, signalen en streamingchunks staan allemaal in MySQL, zodat herstarts van workers de orkestratiestatus niet verloren laten gaan.

Workflows verdelen langdurig werk in stappen. Elke stap slaat de uitvoer op in een gedeelde statusverzameling in de database. Als PHP midden in een stap stopt, probeert de worker die stap opnieuw met behoud van alle eerdere status. De stapgrens is één MySQL-transactie: de statusupdate, voltooiing van de job en het toevoegen van de volgende stap aan de wachtrij gebeuren allemaal atomair.

Code
Workflow: generate_report (3 steps)

Step 0: fetch_data     -> state: {user_data: {...}}
Step 1: call_llm       -> PHP dies -> retry -> state: {user_data: {...}, llm_response: "..."}
Step 2: format_output  -> state: {user_data: {...}, llm_response: "...", report_url: "/reports/42.pdf"}

Workflow: completed

Vereisten

  • PHP 8.2 of later
  • WordPress 6.4 of later
  • MySQL 8.0+ of MariaDB 10.6+, omdat workers jobs ophalen met SELECT ... FOR UPDATE SKIP LOCKED
  • mysqli of pdo_mysql in de PHP-runtime die WordPress en WP-CLI laadt
  • WP-CLI om speciale workers met wp queuety work te draaien
  • De pcntl- en posix-extensies PHP voor workerpools (--workers, --min-workers, --max-workers)

Zie Aan de slag voor de installatie.