> ## Documentation Index
> Fetch the complete documentation index at: https://polos.dev/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Parallel Workflows

> Run workflows in parallel and aggregate results

Run multiple workflows concurrently and aggregate results.

## Parallel execution with batch\_invoke\_and\_wait

<CodeGroup>
  ```python Python theme={null}
  from polos import workflow, WorkflowContext
  from polos.types.types import BatchWorkflowInput

  @workflow(id="parallel_review")
  async def parallel_review(ctx: WorkflowContext, payload):
      reviewers = ["alice", "bob", "charlie"]

      # Create batch of review requests
      requests = [
          BatchWorkflowInput(
              id="single_review",
              payload={"reviewer": r, "document_id": payload["doc_id"]},
          )
          for r in reviewers
      ]

      # Run all in parallel
      results = await ctx.step.batch_invoke_and_wait("reviews", requests)

      # Aggregate
      approved = sum(1 for r in results if r.result.get("approved"))
      return {"approved": approved, "total": len(results)}
  ```

  ```typescript TypeScript theme={null}
  import { defineWorkflow } from '@polos/sdk';
  import type { BatchWorkflowInput } from '@polos/sdk';

  const parallelReview = defineWorkflow<Record<string, unknown>, unknown, AggregatedReview>(
    { id: 'parallel_review' },
    async (ctx, payload) => {
      const reviewers = (payload['reviewers'] as string[]) ?? ['alice', 'bob', 'charlie'];

      // Create batch of review requests
      const reviewRequests: BatchWorkflowInput[] = reviewers.map((reviewer) => ({
        workflow: 'single_review',
        payload: { reviewerId: reviewer, documentId: payload['documentId'] },
      }));

      // Run all in parallel
      const results = await ctx.step.batchInvokeAndWait<ReviewResult>(
        'parallel_reviews',
        reviewRequests,
      );

      // Aggregate
      const approvedCount = results.filter((r) => r.approved).length;
      return { approved: approvedCount, total: results.length };
    },
  );
  ```
</CodeGroup>

## Fire-and-forget with batch\_invoke

<CodeGroup>
  ```python Python theme={null}
  @workflow(id="launch_background_tasks")
  async def launch_tasks(ctx: WorkflowContext, payload):
      requests = [
          BatchWorkflowInput(id="background_task", payload={"task_id": i})
          for i in range(10)
      ]

      # Launch without waiting
      handles = await ctx.step.batch_invoke("launch", requests)

      return {"launched": len(handles), "ids": [h.id for h in handles]}
  ```

  ```typescript TypeScript theme={null}
  const fireAndForgetBatch = defineWorkflow<FireAndForgetPayload, unknown, Record<string, unknown>>(
    { id: 'fire_and_forget_batch' },
    async (ctx, payload) => {
      const taskRequests: BatchWorkflowInput[] = payload.tasks.map((task) => ({
        workflow: 'background_task',
        payload: { taskId: task.id, data: task.data },
      }));

      // Launch without waiting
      const handles = await ctx.step.batchInvoke(
        'launch_background_tasks',
        taskRequests,
      );

      return { launched: handles.length, executionIds: handles.map((h) => h.executionId) };
    },
  );
  ```
</CodeGroup>

## Fan-out/fan-in pattern

<CodeGroup>
  ```python Python theme={null}
  @workflow(id="process_chunks")
  async def process_chunks(ctx: WorkflowContext, payload):
      data = payload["data"]
      chunk_size = 10

      # Split into chunks
      chunks = [data[i:i+chunk_size] for i in range(0, len(data), chunk_size)]

      # Process in parallel
      requests = [
          BatchWorkflowInput(id="process_chunk", payload={"chunk": c, "index": i})
          for i, c in enumerate(chunks)
      ]
      results = await ctx.step.batch_invoke_and_wait("chunks", requests)

      # Combine results
      all_processed = []
      for r in results:
          all_processed.extend(r.result.get("processed", []))
      return {"total": len(all_processed)}
  ```

  ```typescript TypeScript theme={null}
  const dataChunkProcessor = defineWorkflow<ChunkProcessorPayload, unknown, ChunkProcessorResult>(
    { id: 'data_chunk_processor' },
    async (ctx, payload) => {
      const data = payload.data;
      const chunkSize = payload.chunkSize;

      // Split into chunks
      const chunks: unknown[][] = [];
      for (let i = 0; i < data.length; i += chunkSize) {
        chunks.push(data.slice(i, i + chunkSize));
      }

      // Process in parallel
      const chunkRequests: BatchWorkflowInput[] = chunks.map((chunk, i) => ({
        workflow: 'process_chunk',
        payload: { chunk, chunkIndex: i },
      }));
      const results = await ctx.step.batchInvokeAndWait<ChunkResult>(
        'parallel_chunks',
        chunkRequests,
      );

      // Combine results
      const allProcessed: unknown[] = [];
      for (const result of results) {
        allProcessed.push(...result.processed);
      }
      return { totalItems: data.length, chunksProcessed: results.length, processedItems: allProcessed.length, results: allProcessed };
    },
  );
  ```
</CodeGroup>

## Run it

<CodeGroup>
  ```bash Python theme={null}
  git clone https://github.com/polos-dev/polos.git
  cd polos/python-examples/13-parallel-review
  cp .env.example .env  # Add your POLOS_PROJECT_ID and API key
  uv sync
  python main.py
  ```

  ```bash TypeScript theme={null}
  git clone https://github.com/polos-dev/polos.git
  cd polos/typescript-examples/13-parallel-review
  cp .env.example .env  # Add your POLOS_PROJECT_ID and API key
  npm install
  npx tsx main.ts
  ```
</CodeGroup>

Open [http://localhost:5173](http://localhost:5173) to view your agents and workflows, run them from the UI, and see execution traces.

[Python example on GitHub](https://github.com/polos-dev/polos/tree/main/python-examples/13-parallel-review) | [TypeScript example on GitHub](https://github.com/polos-dev/polos/tree/main/typescript-examples/13-parallel-review)