From 980a5e938c054962849b0b615a4669aa5c87e18b Mon Sep 17 00:00:00 2001 From: Lara Perinetti Date: Thu, 11 Jun 2026 17:31:47 +0200 Subject: [PATCH 1/4] Reapply "new examples with workflows" This reverts commit 6c3e66506b83c9852c1d1f606a8a6d6ab4fb89bd. --- template/.env.jinja | 2 + template/Makefile.jinja | 20 +- template/pyproject.toml.jinja | 11 +- template/src/examples/__init__.py | 7 + template/src/examples/workflows/README.md | 139 ++++++++++++ template/src/examples/workflows/__init__.py | 5 + .../src/examples/workflows/search/README.md | 73 +++++++ .../src/examples/workflows/search/__init__.py | 9 + .../examples/workflows/search/activities.py | 172 +++++++++++++++ .../src/examples/workflows/search/models.py | 22 ++ .../src/examples/workflows/search/workflow.py | 36 ++++ .../workflows/search_pipeline/README.md | 85 ++++++++ .../workflows/search_pipeline/__init__.py | 5 + .../workflows/search_pipeline/activities.py | 197 ++++++++++++++++++ .../workflows/search_pipeline/models.py | 22 ++ .../workflows/search_pipeline/workflow.py | 68 ++++++ template/src/examples/workflows/start.py | 92 ++++++++ template/src/examples/workflows/worker.py | 37 ++++ 18 files changed, 1000 insertions(+), 2 deletions(-) create mode 100644 template/src/examples/__init__.py create mode 100644 template/src/examples/workflows/README.md create mode 100644 template/src/examples/workflows/__init__.py create mode 100644 template/src/examples/workflows/search/README.md create mode 100644 template/src/examples/workflows/search/__init__.py create mode 100644 template/src/examples/workflows/search/activities.py create mode 100644 template/src/examples/workflows/search/models.py create mode 100644 template/src/examples/workflows/search/workflow.py create mode 100644 template/src/examples/workflows/search_pipeline/README.md create mode 100644 template/src/examples/workflows/search_pipeline/__init__.py create mode 100644 template/src/examples/workflows/search_pipeline/activities.py create mode 100644 template/src/examples/workflows/search_pipeline/models.py create mode 100644 template/src/examples/workflows/search_pipeline/workflow.py create mode 100644 template/src/examples/workflows/start.py create mode 100644 template/src/examples/workflows/worker.py diff --git a/template/.env.jinja b/template/.env.jinja index 5e39863..cf49985 100644 --- a/template/.env.jinja +++ b/template/.env.jinja @@ -1,6 +1,8 @@ MISTRAL_API_KEY={{ mistral_api_key }} MISTRAL_API_URL=https://api.mistral.ai COLLECTION_NAME={{ collection_name }} +# Required for workflow examples (make start-examples / make execute-ingestion) +DEPLOYMENT_NAME={{ _copier_conf.dst_path.name }} WORKSPACE_ROOT=. # Optional: change only if ports 18080 / 19072 are already in use VESPA_QUERY_PORT=18080 diff --git a/template/Makefile.jinja b/template/Makefile.jinja index 4064ec2..c1c4ae8 100644 --- a/template/Makefile.jinja +++ b/template/Makefile.jinja @@ -1,4 +1,4 @@ -.PHONY: installdeps setup-vespa start-vespa stop-vespa reset-vespa migrate-vespa verify-vespa ingest search bruno generate-vespa-lock +.PHONY: installdeps install-workflows setup-vespa start-vespa stop-vespa reset-vespa migrate-vespa verify-vespa ingest search bruno generate-vespa-lock start-examples execute-ingestion execute-pipeline-ingestion ifneq (,$(wildcard .env)) include .env @@ -65,3 +65,21 @@ generate-vespa-lock: uv run mistral-vespa generate \ --app-dir src/vespa_app \ --path ./vespa.lock + +## Install optional workflows dependency (required for examples/workflows/) +install-workflows: + uv sync --extra workflows + +## Start a worker that registers the example workflows (requires install-workflows) +start-examples: install-workflows + uv run python -m examples.workflows.worker + +## Execute the ingestion workflow via the Mistral Workflows API +## Usage: make execute-ingestion input='{"file_path": "sample_data/hello.txt", "collection_name": "mydocs"}' +execute-ingestion: install-workflows + uv run python -m examples.workflows.start --workflow document-ingestion $(if $(input),--input '$(input)',--input '{"file_path":"sample_data/hello.txt"}') + +## Execute the advanced pipeline-step ingestion workflow +## Usage: make execute-pipeline-ingestion input='{"file_path": "sample_data/hello.txt", "collection_name": "mydocs"}' +execute-pipeline-ingestion: install-workflows + uv run python -m examples.workflows.start --workflow document-ingestion-pipeline $(if $(input),--input '$(input)',--input '{"file_path":"sample_data/hello.txt"}') diff --git a/template/pyproject.toml.jinja b/template/pyproject.toml.jinja index 108ef9c..361ef8a 100644 --- a/template/pyproject.toml.jinja +++ b/template/pyproject.toml.jinja @@ -8,6 +8,12 @@ dependencies = [ "python-dotenv>=0.9.9", ] +[project.optional-dependencies] +workflows = [ + "mistralai-workflows[mistralai]>=3.0.0,<4", + "pydantic", +] + [dependency-groups] dev = [ "ruff>=0.11", @@ -16,6 +22,9 @@ dev = [ [tool.uv] exclude-newer = "7 days" +# Gemfury (UV_EXTRA_INDEX_URL) may list older versions of mistral-common / +# mistralai-search-toolkit than PyPI. Allow uv to pick the best match across indexes. +index-strategy = "unsafe-best-match" [tool.uv.exclude-newer-package] mistralai = false @@ -28,4 +37,4 @@ requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] -packages = ["src/entrypoints", "src/vespa_app"] +packages = ["src/entrypoints", "src/vespa_app", "src/examples"] diff --git a/template/src/examples/__init__.py b/template/src/examples/__init__.py new file mode 100644 index 0000000..eba47ca --- /dev/null +++ b/template/src/examples/__init__.py @@ -0,0 +1,7 @@ +"""Examples for the search-starter-app. + +Workflow-based examples live under ``examples/workflows/`` — see that folder's +README for setup and usage. + +Other examples that do not use the workflows framework live directly here. +""" diff --git a/template/src/examples/workflows/README.md b/template/src/examples/workflows/README.md new file mode 100644 index 0000000..6f90f01 --- /dev/null +++ b/template/src/examples/workflows/README.md @@ -0,0 +1,139 @@ +# Workflow Examples + +This folder contains examples that integrate the search-starter-app with the [Mistral Workflows](https://docs.mistral.ai/capabilities/workflows/) framework. + +Workflow-based examples live under `examples/workflows/`. Other examples that do not use the workflows framework (standalone scripts, direct API calls, etc.) live directly under `examples/`. + +## Available examples + +| Example | Workflow name | Makefile target | Description | +| --- | --- | --- | --- | +| [search/](search/) | `document-ingestion` | `make execute-ingestion` | Basic: 2 activities wrapping the full pipeline | +| [search_pipeline/](search_pipeline/) | `document-ingestion-pipeline` | `make execute-pipeline-ingestion` | Advanced: one activity per pipeline stage | + +Both examples ingest local files into Vespa using the same Search Toolkit components as `make ingest`. Choose the basic example to get started; use the pipeline example when you want per-step retries, timeouts, and a detailed timeline in the Mistral Console. + +## Prerequisites + +The workflows framework is an **optional** dependency — the core search project works without it. + +> Run these commands from your **generated project root** (the folder created by `copier copy`), not from the `search-starter-app` template repo itself. + +```bash +make install-workflows # uv sync --extra workflows +make setup-vespa +``` + +Ensure these are set in your `.env` file: + +- `MISTRAL_API_KEY` +- `DEPLOYMENT_NAME` — a stable identifier for this worker (defaults to your project name when generated via `copier copy`) + +## Workflow input + +Both workflows accept the same JSON input: + +```json +{ + "file_path": "sample_data/hello.txt", + "collection_name": "exampledocs" +} +``` + +| Field | Required | Default | Description | +| --- | --- | --- | --- | +| `file_path` | yes | — | Path to a file or directory to ingest | +| `collection_name` | no | `"exampledocs"` | Vespa collection name | + +## How to run + +> **Important:** execution commands call the Mistral Workflows API. The workflow must be registered first by a running worker. If you see `Workflow not found`, start the worker below and retry. + +**Terminal 1 — start the examples worker (leave this running):** + +```bash +make start-examples +``` + +Wait until the worker is listening for execution requests. It registers all workflows listed in `worker.py`. + +**Terminal 2 — trigger a workflow:** + +```bash +# Basic example (2 activities) +make execute-ingestion +make execute-ingestion input='{"file_path": "sample_data/hello.txt"}' + +# Advanced example — one activity per pipeline stage +make execute-pipeline-ingestion +make execute-pipeline-ingestion input='{"file_path": "sample_data", "collection_name": "mydocs"}' +``` + +You can also trigger workflows from the [Mistral Console](https://console.mistral.ai/build/workflows): select `document-ingestion` or `document-ingestion-pipeline`, click **Start Workflow**, and provide the input JSON. + +After ingestion, search directly (no workflow): + +```bash +make search query="hello world" +``` + +## Folder layout + +```text +examples/workflows/ +├── README.md # This file +├── worker.py # Registers and runs all workflow examples +├── start.py # CLI to trigger a workflow execution +├── search/ # Basic example (2 activities) +│ ├── models.py +│ ├── activities.py +│ ├── workflow.py +│ └── README.md +└── search_pipeline/ # Advanced example (7 activities) + ├── models.py + ├── activities.py + ├── workflow.py + └── README.md +``` + +## Design principles + +- **Workflows for ingestion** — document ingestion is long-running and benefits from durability, retries, and observability in the Mistral Console. +- **Search stays direct** — search queries use `make search` / `entrypoints/search.py` for low latency. Workflows are not needed for simple queries. +- **Activities own all I/O** — filesystem access, API calls, and Vespa writes live in `activities.py`. The workflow body in `workflow.py` only orchestrates. +- **Granular vs bundled activities** — the basic example bundles the pipeline into fewer activities; the pipeline example exposes each stage separately for production-style observability. The pipeline example passes serialized document data between steps and is best suited to **small files** (~2 MB activity payload limit); see [search_pipeline/README.md](search_pipeline/README.md#large-files-and-payload-limits). + +## Adding a new workflow example + +1. Create a subdirectory under `examples/workflows/` (e.g. `my_example/`). +2. Add `models.py`, `activities.py`, `workflow.py`, and a `README.md`. +3. Export the workflow class from the package `__init__.py`. +4. Register it in `EXAMPLE_WORKFLOWS` in `worker.py`. +5. Add a Makefile target that calls `examples.workflows.start --workflow `. + +Example structure: + +```text +examples/workflows/my_example/ +├── __init__.py +├── models.py +├── activities.py +├── workflow.py +└── README.md +``` + +## Troubleshooting + +| Issue | Fix | +| --- | --- | +| `Workflow not found` | Start `make start-examples` in a separate terminal, then retry | +| `DEPLOYMENT_NAME is required` | Add `DEPLOYMENT_NAME=` to `.env` | +| Worker fails on startup | Ensure `imports_passed_through()` wraps activity imports in `workflow.py` when activities use `mistralai.client` | +| `unserializable_payload_error` / payload too large | Advanced pipeline passes full document dicts between activities (~2 MB limit). Use `make ingest`, the basic `document-ingestion` workflow, or OffloadableField + blob storage for larger files — see [search_pipeline/README.md](search_pipeline/README.md#large-files-and-payload-limits) | +| Vespa errors during indexing | Run `make setup-vespa` before triggering ingestion | + +Enable verbose logging: + +```bash +LOG_LEVEL=DEBUG make start-examples +``` diff --git a/template/src/examples/workflows/__init__.py b/template/src/examples/workflows/__init__.py new file mode 100644 index 0000000..1baee12 --- /dev/null +++ b/template/src/examples/workflows/__init__.py @@ -0,0 +1,5 @@ +"""Workflow-based examples for the search-starter-app. + +Each subdirectory is a self-contained workflow example. Run the worker with: + make start-examples +""" diff --git a/template/src/examples/workflows/search/README.md b/template/src/examples/workflows/search/README.md new file mode 100644 index 0000000..4c4232f --- /dev/null +++ b/template/src/examples/workflows/search/README.md @@ -0,0 +1,73 @@ +# Document Ingestion Workflow Example + +This example wraps the search-starter-app ingestion pipeline in a [Mistral Workflow](https://docs.mistral.ai/capabilities/workflows/), adding durability, observability, and retry support. + +## What It Demonstrates + +| Primitive | Where | +|-----------|-------| +| `@workflows.workflow.define` | `workflow.py` — defines the workflow with a name visible in the Mistral Console | +| `@workflows.workflow.entrypoint` | `workflow.py` — pure orchestration, no I/O | +| `@workflows.activity` | `activities.py` — all file I/O, embedding calls, and Vespa writes | +| Pydantic input/output models | `models.py` — typed boundaries between workflow and activities | +| `workflows.run_worker()` | `worker.py` — registers workflows with the Temporal engine | + +## Architecture + +``` +IngestionWorkflow.run(IngestionInput) +└── ingest_documents activity + ├── FilesystemFileLoader [loads documents] + ├── PlainTextExtractor [for .txt, .md, .csv, .json files] + │ or MistralOCRExtractor [for PDFs and images] + ├── MarkdownTextSplitter [chunks the content] + ├── MistralEmbedder [generates vector embeddings] + └── VespaSearchIndex [stores chunks in Vespa] +``` + +## Prerequisites + +1. Install the workflows extra: + ```bash + make install-workflows + ``` + +2. Vespa must be running: + ```bash + make setup-vespa + ``` + +3. Set `MISTRAL_API_KEY` in your `.env` file. + +## Running the Example + +> The worker must be running before you trigger the workflow, otherwise the API returns `Workflow not found`. + +**Terminal 1 — start the worker (leave this running):** +```bash +make start-examples +``` + +**Terminal 2 — trigger ingestion:** +```bash +# Ingest a single file +make execute-ingestion input='{"file_path": "sample_data/hello.txt"}' + +# Ingest a directory +make execute-ingestion input='{"file_path": "sample_data"}' + +# Use a custom collection name +make execute-ingestion input='{"file_path": "sample_data/hello.txt", "collection_name": "mydocs"}' +``` + +You can also trigger the workflow from the [Mistral Console](https://console.mistral.ai/build/workflows) by selecting `document-ingestion` and providing the input JSON. + +## Key Difference from Direct Ingestion + +The existing `make ingest` command runs ingestion synchronously in the terminal. The workflow approach adds: + +- **Durability**: the worker can restart mid-ingestion and resume from the last completed step +- **Observability**: each execution is visible in the Mistral Console with a full timeline +- **Retries**: the activity retries automatically on transient errors (network blips, API rate limits) + +Search operations remain direct queries — they don't benefit from workflow orchestration since they complete in milliseconds. diff --git a/template/src/examples/workflows/search/__init__.py b/template/src/examples/workflows/search/__init__.py new file mode 100644 index 0000000..3d4c06d --- /dev/null +++ b/template/src/examples/workflows/search/__init__.py @@ -0,0 +1,9 @@ +"""Workflows × Search integration example. + +Demonstrates wrapping the search-starter-app ingestion pipeline in a +Mistral Workflow for durability, observability, and HITL support. +""" + +from .workflow import IngestionWorkflow + +__all__ = ["IngestionWorkflow"] diff --git a/template/src/examples/workflows/search/activities.py b/template/src/examples/workflows/search/activities.py new file mode 100644 index 0000000..c4bd97f --- /dev/null +++ b/template/src/examples/workflows/search/activities.py @@ -0,0 +1,172 @@ +"""Activities for the document ingestion workflow. + +All I/O lives here: file system access, embedding calls, and Vespa writes. +The workflow body in workflow.py calls these as durable, retryable units. + +Stateless pipeline components and API clients are provided via Depends() so +the SDK initialises them once at worker startup and reuses them across every +activity execution. The only per-call objects are VespaSearchIndex and the +Pipeline wrappers, because they are keyed to the collection_name input. +""" + +import os +from datetime import timedelta +from functools import cache +from pathlib import Path +from typing import Any + +import mistralai.workflows as workflows +from mistralai.client import Mistral +from mistralai.search.toolkit.embedders import MistralEmbedder +from mistralai.search.toolkit.ingestion.extractors import ( + MistralOCRExtractor, + PlainTextExtractor, +) +from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader +from mistralai.search.toolkit.ingestion.pipelines import Pipeline +from mistralai.search.toolkit.ingestion.text_splitters import ( + MarkdownTextSplitter, + MarkdownTextSplitterConfig, +) +from mistralai.search.toolkit.plugins.vespa import VespaClientConfig +from mistralai.workflows import Depends +from vespa_app import app, vespa_endpoint + +from .models import IngestionResult + +_TEXT_SUFFIXES = {".txt", ".md", ".markdown", ".csv", ".json"} + + +# --------------------------------------------------------------------------- +# Dependency providers — each called once at worker startup +# --------------------------------------------------------------------------- + + +@cache +def get_mistral_client() -> Mistral: + api_key = os.environ.get("MISTRAL_API_KEY", "") + if not api_key: + raise ValueError("MISTRAL_API_KEY is not set. Check your .env file.") + return Mistral( + api_key=api_key, + server_url=os.getenv("MISTRAL_API_URL", "https://api.mistral.ai"), + ) + + +def get_embedder() -> MistralEmbedder: + return MistralEmbedder(client=get_mistral_client()) + + +def get_ocr_extractor() -> MistralOCRExtractor: + return MistralOCRExtractor(client=get_mistral_client()) + + +def get_loader() -> FilesystemFileLoader: + return FilesystemFileLoader() + + +def get_text_splitter() -> MarkdownTextSplitter: + return MarkdownTextSplitter( + MarkdownTextSplitterConfig(chunk_size=4096, chunk_overlap=50) + ) + + +# --------------------------------------------------------------------------- +# Per-call pipeline factory +# --------------------------------------------------------------------------- + + +def _build_pipelines( + loader: FilesystemFileLoader, + text_splitter: MarkdownTextSplitter, + embedder: MistralEmbedder, + ocr_extractor: MistralOCRExtractor, + vector_store: Any, +) -> tuple[Pipeline, Pipeline]: + """Assemble plain-text and OCR pipelines from shared components. + + VespaSearchIndex (vector_store) is the only per-call argument because + it is keyed to the collection_name supplied at runtime. + """ + shared = dict( + loader=loader, + text_splitter=text_splitter, + embedder=embedder, + stores=vector_store, + ) + return ( + Pipeline(extractor=PlainTextExtractor(), **shared), + Pipeline(extractor=ocr_extractor, **shared), + ) + + +# --------------------------------------------------------------------------- +# Activities +# --------------------------------------------------------------------------- + + +@workflows.activity( + name="collect_document_paths", + start_to_close_timeout=timedelta(seconds=30), + retry_policy_max_attempts=1, +) +async def collect_document_paths(file_path: str) -> list[str]: + """Collect all file paths to ingest from a file or directory. + + Returns string paths (Path objects are not serialisable across the + activity boundary) already split into two lists: text files and OCR files. + """ + root = Path(file_path) + if root.is_file(): + documents = [root] + elif root.is_dir(): + documents = sorted(p for p in root.rglob("*") if p.is_file()) + else: + raise ValueError(f"Path not found: {file_path}") + + if not documents: + raise ValueError(f"No files found at: {file_path}") + + return [str(p) for p in documents] + + +@workflows.activity( + name="ingest_documents", + # Ingestion can be slow for large files or directories with many PDFs. + # 30 minutes is a generous upper bound; raise if processing very large corpora. + start_to_close_timeout=timedelta(minutes=30), + retry_policy_max_attempts=2, +) +async def ingest_documents( + paths: list[str], + collection_name: str, + embedder: MistralEmbedder = Depends(get_embedder), + ocr_extractor: MistralOCRExtractor = Depends(get_ocr_extractor), + loader: FilesystemFileLoader = Depends(get_loader), + text_splitter: MarkdownTextSplitter = Depends(get_text_splitter), +) -> IngestionResult: + """Run the ingestion pipelines on a pre-collected list of file paths.""" + vector_store = app.get_search_index( + VespaClientConfig(endpoint=vespa_endpoint()), + collection_name=collection_name, + ) + plain_text_pipeline, ocr_pipeline = _build_pipelines( + loader, text_splitter, embedder, ocr_extractor, vector_store + ) + + documents = [Path(p) for p in paths] + text_docs = [p for p in documents if p.suffix.lower() in _TEXT_SUFFIXES] + ocr_docs = [p for p in documents if p.suffix.lower() not in _TEXT_SUFFIXES] + + total_chunks = 0 + if text_docs: + total_chunks += await plain_text_pipeline.run(documents=text_docs, use_checkpoint=False) + if ocr_docs: + total_chunks += await ocr_pipeline.run(documents=ocr_docs, use_checkpoint=False) + + return IngestionResult( + status="success", + total_chunks=total_chunks, + file_count=len(documents), + collection_name=collection_name, + ) diff --git a/template/src/examples/workflows/search/models.py b/template/src/examples/workflows/search/models.py new file mode 100644 index 0000000..ceb2e00 --- /dev/null +++ b/template/src/examples/workflows/search/models.py @@ -0,0 +1,22 @@ +"""Pydantic models for the document ingestion workflow.""" + +from pydantic import BaseModel, Field + + +class IngestionInput(BaseModel): + """Input accepted by the document-ingestion workflow.""" + + file_path: str = Field(description="Local path to a file or directory to ingest") + collection_name: str = Field( + default="exampledocs", + description="Vespa collection to index documents into", + ) + + +class IngestionResult(BaseModel): + """Output returned by the document-ingestion workflow.""" + + status: str = Field(description="'success' or 'error'") + total_chunks: int = Field(description="Total number of chunks indexed") + file_count: int = Field(description="Number of files processed") + collection_name: str = Field(description="Vespa collection that was used") diff --git a/template/src/examples/workflows/search/workflow.py b/template/src/examples/workflows/search/workflow.py new file mode 100644 index 0000000..c0c22f9 --- /dev/null +++ b/template/src/examples/workflows/search/workflow.py @@ -0,0 +1,36 @@ +"""Document ingestion workflow -- orchestration only, no I/O here.""" + +import mistralai.workflows as workflows +from mistralai.workflows import workflow + +# Activities import the search toolkit and mistralai.client (→ httpx), which +# the Temporal sandbox blocks at workflow import time. imports_passed_through +# lets the workflow reference activity stubs without loading I/O deps in-process. +with workflow.unsafe.imports_passed_through(): + from .activities import collect_document_paths, ingest_documents + +from .models import IngestionInput, IngestionResult # noqa: E402 + + +@workflows.workflow.define( + name="document-ingestion", + workflow_display_name="Document Ingestion", + workflow_description=( + "Ingest documents from the local filesystem into the Vespa search index " + "using the Search Toolkit pipeline: load → extract → split → embed → index." + ), +) +class IngestionWorkflow: + """Workflow that wraps the search-starter-app ingestion pipeline. + + All I/O (file loading, LLM calls, Vespa writes) lives in activities.py so + the workflow body stays deterministic and replayable by the Temporal engine. + """ + + @workflows.workflow.entrypoint + async def run(self, params: IngestionInput) -> IngestionResult: + paths = await collect_document_paths(file_path=params.file_path) + return await ingest_documents( + paths=paths, + collection_name=params.collection_name, + ) diff --git a/template/src/examples/workflows/search_pipeline/README.md b/template/src/examples/workflows/search_pipeline/README.md new file mode 100644 index 0000000..791a370 --- /dev/null +++ b/template/src/examples/workflows/search_pipeline/README.md @@ -0,0 +1,85 @@ +# Document Ingestion Pipeline (Advanced) + +Advanced workflow example where **each Search Toolkit pipeline stage is its own activity**. Use this when you want per-step retries, timeouts, and a detailed timeline in the Mistral Console. + +## Comparison with the basic example + +| | [search/](../search/) (basic) | search_pipeline/ (this example) | +| --- | --- | --- | +| Workflow name | `document-ingestion` | `document-ingestion-pipeline` | +| Activities | 2 (`collect_document_paths`, `ingest_documents`) | 7 (one per pipeline stage) | +| Console timeline | Coarse-grained | Fine-grained: load → extract → split → embed → index | +| Best for | Getting started, minimal boilerplate | Production patterns, per-step retry policies | + +## Pipeline stages (activities) + +``` +For each file: + pipeline_load_document → FilesystemFileLoader + pipeline_extract_plain_text → PlainTextExtractor (.txt, .md, .csv, …) + or pipeline_extract_with_ocr → MistralOCRExtractor (PDFs, images) + pipeline_split_document → MarkdownTextSplitter + pipeline_embed_document → MistralEmbedder + pipeline_index_document → VespaSearchIndex +``` + +## Workflow input + +```json +{ + "file_path": "sample_data/hello.txt", + "collection_name": "exampledocs" +} +``` + +## Run + +**Terminal 1:** +```bash +make start-examples +``` + +**Terminal 2:** +```bash +make execute-pipeline-ingestion +make execute-pipeline-ingestion input='{"file_path": "sample_data", "collection_name": "mydocs"}' +``` + +Or from the [Mistral Console](https://console.mistral.ai/build/workflows): select `document-ingestion-pipeline`. + +## What to look for in the Console + +Each file should produce a sequence of activity events: + +1. **pipeline_load_document** — file read from disk +2. **pipeline_extract_plain_text** or **pipeline_extract_with_ocr** — text extraction +3. **pipeline_split_document** — chunking +4. **pipeline_embed_document** — embedding API call +5. **pipeline_index_document** — Vespa write + +If a step fails (e.g. transient OCR error), only that activity retries — completed steps are not re-run. + +## Large files and payload limits + +This example passes **serialized file and document data** between activities (see `_file_to_dict` / `_document_to_dict` in `activities.py`). Workflows enforce Temporal's **~2 MB limit** on each activity input and output. + +Payload size grows at every stage: + +| Stage | What crosses the activity boundary | +| --- | --- | +| load → extract | Raw file bytes (JSON-serialised) | +| extract → split | Full extracted text | +| split → embed | Text duplicated in every chunk | +| embed → index | Chunk text + embedding vectors | + +**This example is intended for small files** (e.g. `sample_data/hello.txt`) where the goal is per-step retries, timeouts, and a detailed timeline in the Mistral Console. + +### What to use instead for larger files + +| Scenario | Recommendation | +| --- | --- | +| Local files or directories, no per-step Console timeline needed | `make ingest` | +| Workflow durability with minimal payload size | [Basic example](../search/) (`document-ingestion`) — passes only file paths; the full pipeline runs inside one activity | +| Production ingestion with per-step activities and large documents | Pass **references** (paths, S3/GCS URIs, artifact IDs) between activities instead of inline payloads, or use **OffloadableField** with blob storage — see the [Handling Large Data](https://docs.mistral.ai/capabilities/workflows/guides/handling-large-data/) guide | + +We deliberately keep this starter example simple (inline dicts, no cloud storage setup). diff --git a/template/src/examples/workflows/search_pipeline/__init__.py b/template/src/examples/workflows/search_pipeline/__init__.py new file mode 100644 index 0000000..a0a4cf8 --- /dev/null +++ b/template/src/examples/workflows/search_pipeline/__init__.py @@ -0,0 +1,5 @@ +"""Advanced search ingestion workflow — one activity per pipeline step.""" + +from .workflow import PipelineIngestionWorkflow + +__all__ = ["PipelineIngestionWorkflow"] diff --git a/template/src/examples/workflows/search_pipeline/activities.py b/template/src/examples/workflows/search_pipeline/activities.py new file mode 100644 index 0000000..41b4f7e --- /dev/null +++ b/template/src/examples/workflows/search_pipeline/activities.py @@ -0,0 +1,197 @@ +"""Activities for the pipeline-step ingestion workflow. + +Each Search Toolkit pipeline stage is a separate durable activity so the Mistral +Console timeline shows load → extract → split → embed → index as distinct steps +with independent retry policies. +""" + +import os +from datetime import timedelta +from functools import cache +from pathlib import Path +from typing import Any + +import mistralai.workflows as workflows +from mistralai.client import Mistral +from mistralai.search.toolkit.context import IngestContext +from mistralai.search.toolkit.document import Document +from mistralai.search.toolkit.embedders import MistralEmbedder +from mistralai.search.toolkit.ingestion import File +from mistralai.search.toolkit.ingestion.extractors import ( + MistralOCRExtractor, + PlainTextExtractor, +) +from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader +from mistralai.search.toolkit.ingestion.text_splitters import ( + MarkdownTextSplitter, + MarkdownTextSplitterConfig, +) +from mistralai.search.toolkit.plugins.vespa import VespaClientConfig +from mistralai.workflows import Depends +from vespa_app import app, vespa_endpoint + + +@cache +def get_mistral_client() -> Mistral: + api_key = os.environ.get("MISTRAL_API_KEY", "") + if not api_key: + raise ValueError("MISTRAL_API_KEY is not set. Check your .env file.") + return Mistral( + api_key=api_key, + server_url=os.getenv("MISTRAL_API_URL", "https://api.mistral.ai"), + ) + + +def get_embedder() -> MistralEmbedder: + return MistralEmbedder(client=get_mistral_client()) + + +def get_ocr_extractor() -> MistralOCRExtractor: + return MistralOCRExtractor(client=get_mistral_client()) + + +def get_loader() -> FilesystemFileLoader: + return FilesystemFileLoader() + + +def get_text_splitter() -> MarkdownTextSplitter: + return MarkdownTextSplitter( + MarkdownTextSplitterConfig(chunk_size=4096, chunk_overlap=50) + ) + + +def get_plain_text_extractor() -> PlainTextExtractor: + return PlainTextExtractor() + + +def _file_to_dict(file: File) -> dict[str, Any]: + return file.model_dump(mode="json") + + +def _file_from_dict(data: dict[str, Any]) -> File: + return File.model_validate(data) + + +def _document_to_dict(document: Document) -> dict[str, Any]: + return document.model_dump(mode="json") + + +def _document_from_dict(data: dict[str, Any]) -> Document: + return Document.model_validate(data) + + +@workflows.activity( + name="pipeline_collect_document_paths", + start_to_close_timeout=timedelta(seconds=30), + retry_policy_max_attempts=1, +) +async def collect_document_paths(file_path: str) -> list[str]: + """List all files under a path (file or directory).""" + root = Path(file_path) + if root.is_file(): + documents = [root] + elif root.is_dir(): + documents = sorted(p for p in root.rglob("*") if p.is_file()) + else: + raise ValueError(f"Path not found: {file_path}") + + if not documents: + raise ValueError(f"No files found at: {file_path}") + + return [str(p) for p in documents] + + +@workflows.activity( + name="pipeline_load_document", + start_to_close_timeout=timedelta(minutes=1), + retry_policy_max_attempts=2, +) +async def load_document( + file_path: str, + loader: FilesystemFileLoader = Depends(get_loader), +) -> dict[str, Any]: + """Load a single file from disk into a serialisable File payload.""" + file = await loader.load_file(file_path) + return _file_to_dict(file) + + +@workflows.activity( + name="pipeline_extract_plain_text", + start_to_close_timeout=timedelta(minutes=2), + retry_policy_max_attempts=2, +) +async def extract_plain_text( + file_data: dict[str, Any], + extractor: PlainTextExtractor = Depends(get_plain_text_extractor), +) -> dict[str, Any]: + """Extract text from plain-text formats (.txt, .md, .csv, .json, …).""" + file = _file_from_dict(file_data) + document = await extractor.extract(file, context=IngestContext()) + return _document_to_dict(document) + + +@workflows.activity( + name="pipeline_extract_with_ocr", + start_to_close_timeout=timedelta(minutes=10), + retry_policy_max_attempts=3, +) +async def extract_with_ocr( + file_data: dict[str, Any], + ocr_extractor: MistralOCRExtractor = Depends(get_ocr_extractor), +) -> dict[str, Any]: + """Extract text from PDFs and images via Mistral OCR.""" + file = _file_from_dict(file_data) + document = await ocr_extractor.extract(file, context=IngestContext()) + return _document_to_dict(document) + + +@workflows.activity( + name="pipeline_split_document", + start_to_close_timeout=timedelta(minutes=2), + retry_policy_max_attempts=2, +) +async def split_document( + document_data: dict[str, Any], + text_splitter: MarkdownTextSplitter = Depends(get_text_splitter), +) -> dict[str, Any]: + """Split an extracted document into chunks.""" + document = _document_from_dict(document_data) + chunks = text_splitter.split_document(document, context=IngestContext()) + return _document_to_dict(document.model_copy(update={"chunks": chunks})) + + +@workflows.activity( + name="pipeline_embed_document", + start_to_close_timeout=timedelta(minutes=5), + retry_policy_max_attempts=3, +) +async def embed_document( + document_data: dict[str, Any], + embedder: MistralEmbedder = Depends(get_embedder), +) -> dict[str, Any]: + """Generate vector embeddings for all chunks in a document.""" + document = _document_from_dict(document_data) + if not document.chunks: + return document_data + + embedded_chunks = await embedder.embed_chunks(list(document.chunks)) + return _document_to_dict(document.model_copy(update={"chunks": embedded_chunks})) + + +@workflows.activity( + name="pipeline_index_document", + start_to_close_timeout=timedelta(minutes=5), + retry_policy_max_attempts=2, +) +async def index_document( + document_data: dict[str, Any], + collection_name: str, +) -> int: + """Write an embedded document to Vespa and return the number of chunks indexed.""" + document = _document_from_dict(document_data) + vector_store = app.get_search_index( + VespaClientConfig(endpoint=vespa_endpoint()), + collection_name=collection_name, + ) + await vector_store.index_document(document, context=IngestContext()) + return len(document.chunks) diff --git a/template/src/examples/workflows/search_pipeline/models.py b/template/src/examples/workflows/search_pipeline/models.py new file mode 100644 index 0000000..fce75a0 --- /dev/null +++ b/template/src/examples/workflows/search_pipeline/models.py @@ -0,0 +1,22 @@ +"""Pydantic models for the pipeline-step ingestion workflow.""" + +from pydantic import BaseModel, Field + + +class PipelineIngestionInput(BaseModel): + """Input accepted by the document-ingestion-pipeline workflow.""" + + file_path: str = Field(description="Local path to a file or directory to ingest") + collection_name: str = Field( + default="exampledocs", + description="Vespa collection to index documents into", + ) + + +class PipelineIngestionResult(BaseModel): + """Output returned after all documents are processed.""" + + status: str = Field(description="'success' or 'error'") + total_chunks: int = Field(description="Total number of chunks indexed") + file_count: int = Field(description="Number of files processed") + collection_name: str = Field(description="Vespa collection that was used") diff --git a/template/src/examples/workflows/search_pipeline/workflow.py b/template/src/examples/workflows/search_pipeline/workflow.py new file mode 100644 index 0000000..6851918 --- /dev/null +++ b/template/src/examples/workflows/search_pipeline/workflow.py @@ -0,0 +1,68 @@ +"""Advanced ingestion workflow — one activity per pipeline step. + +Demonstrates granular workflow orchestration over the Search Toolkit pipeline: + load → extract → split → embed → index + +Each stage is a separate @workflows.activity with its own timeout and retry +policy. The workflow fans out over files and branches deterministically on +file suffix to pick PlainTextExtractor vs MistralOCRExtractor. +""" + +import mistralai.workflows as workflows +from mistralai.workflows import workflow + +with workflow.unsafe.imports_passed_through(): + from .activities import ( + collect_document_paths, + embed_document, + extract_plain_text, + extract_with_ocr, + index_document, + load_document, + split_document, + ) + +from .models import PipelineIngestionInput, PipelineIngestionResult # noqa: E402 + +_TEXT_SUFFIXES = frozenset({".txt", ".md", ".markdown", ".csv", ".json"}) + + +@workflows.workflow.define( + name="document-ingestion-pipeline", + workflow_display_name="Document Ingestion (Pipeline Steps)", + workflow_description=( + "Advanced ingestion example: each Search Toolkit stage runs as its own " + "activity (load → extract → split → embed → index) with per-step retries " + "and observability in the Mistral Console." + ), +) +class PipelineIngestionWorkflow: + """Orchestrate ingestion with one durable activity per pipeline stage.""" + + @workflows.workflow.entrypoint + async def run(self, params: PipelineIngestionInput) -> PipelineIngestionResult: + paths = await collect_document_paths(file_path=params.file_path) + + total_chunks = 0 + for path in paths: + suffix = path.rsplit(".", 1)[-1].lower() if "." in path else "" + file_data = await load_document(file_path=path) + + if f".{suffix}" in _TEXT_SUFFIXES: + document_data = await extract_plain_text(file_data=file_data) + else: + document_data = await extract_with_ocr(file_data=file_data) + + document_data = await split_document(document_data=document_data) + document_data = await embed_document(document_data=document_data) + total_chunks += await index_document( + document_data=document_data, + collection_name=params.collection_name, + ) + + return PipelineIngestionResult( + status="success", + total_chunks=total_chunks, + file_count=len(paths), + collection_name=params.collection_name, + ) diff --git a/template/src/examples/workflows/start.py b/template/src/examples/workflows/start.py new file mode 100644 index 0000000..6d6605e --- /dev/null +++ b/template/src/examples/workflows/start.py @@ -0,0 +1,92 @@ +"""Trigger a workflow execution from the command line. + +Requires the optional workflows extra: + uv sync --extra workflows +""" +# ruff: noqa: E402 + +import argparse +import asyncio +import json +import os + +from dotenv import load_dotenv + +load_dotenv(override=True) + +from mistralai.extra.workflows import WorkflowEncodingConfig, configure_workflow_encoding +from mistralai.workflows.client import get_mistral_client + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Trigger a workflow execution.") + parser.add_argument( + "--workflow", + default="document-ingestion", + help="Workflow name (document-ingestion or document-ingestion-pipeline)", + ) + parser.add_argument( + "--input", + default=r"{}", + help=r'Input data as a JSON string (e.g. \'{"file_path": "sample_data/hello.txt"}\')', + ) + return parser.parse_args() + + +async def main() -> None: + args = parse_args() + + try: + raw_input = json.loads(args.input) + except json.JSONDecodeError as exc: + raise SystemExit( + f"Error: invalid JSON for --input: {exc.args[0]}\n" + f" Received: {args.input!r}\n" + f' Example: --input \'{{"file_path": "sample_data/hello.txt"}}\'' + ) from exc + + raw_input = raw_input or {} + if not isinstance(raw_input, dict): + raise SystemExit( + f"Error: --input must be a JSON object, got {type(raw_input).__name__}" + ) + + api_key = os.environ.get("MISTRAL_API_KEY", "") + if not api_key: + raise SystemExit("Error: MISTRAL_API_KEY is not set. Check your .env file.") + + deployment_name = os.environ.get("DEPLOYMENT_NAME") + if not deployment_name: + raise SystemExit( + "Error: DEPLOYMENT_NAME is not set. Add it to your .env file, e.g.:\n" + " DEPLOYMENT_NAME=my-search-project" + ) + + client = get_mistral_client( + api_key=api_key, + server_url=os.environ.get("MISTRAL_API_URL", "https://api.mistral.ai"), + ) + + await configure_workflow_encoding(WorkflowEncodingConfig(), client=client) + + try: + result = await client.workflows.execute_workflow_and_wait_async( + workflow_identifier=args.workflow, + input=raw_input, + deployment_name=deployment_name, + ) + except Exception as exc: + if "Workflow not found" in str(exc) or "404" in str(exc): + raise SystemExit( + f"Error: workflow '{args.workflow}' not found.\n" + "Start the examples worker first (separate terminal):\n" + " make start-examples\n" + "Then retry this command." + ) from exc + raise + + print(f"Result: {result}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/template/src/examples/workflows/worker.py b/template/src/examples/workflows/worker.py new file mode 100644 index 0000000..686cc1f --- /dev/null +++ b/template/src/examples/workflows/worker.py @@ -0,0 +1,37 @@ +"""Start a worker that registers all example workflows for search-starter-app.""" +# ruff: noqa: E402 + +import asyncio +import os +import sys + +from dotenv import load_dotenv + +load_dotenv(override=True) + +import mistralai.workflows as workflows + +from examples.workflows.search import IngestionWorkflow +from examples.workflows.search_pipeline import PipelineIngestionWorkflow + +EXAMPLE_WORKFLOWS = [IngestionWorkflow, PipelineIngestionWorkflow] + + +async def main() -> None: + if not os.environ.get("DEPLOYMENT_NAME"): + print( + "Error: DEPLOYMENT_NAME is not set. Add it to your .env file, e.g.:\n" + " DEPLOYMENT_NAME=my-search-project", + file=sys.stderr, + ) + raise SystemExit(1) + + names = [wf.__name__ for wf in EXAMPLE_WORKFLOWS] + print( + f"Starting worker with {len(EXAMPLE_WORKFLOWS)} example(s): {', '.join(names)}" + ) + await workflows.run_worker(EXAMPLE_WORKFLOWS) + + +if __name__ == "__main__": + asyncio.run(main()) From 76ee1a37282ec19dd44fd95251425a9b3cc0edaa Mon Sep 17 00:00:00 2001 From: Lara Perinetti Date: Thu, 11 Jun 2026 18:26:39 +0200 Subject: [PATCH 2/4] Remove multi-activity pipeline example; document scaling guidance. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drop search_pipeline/ per review — passing serialized documents between activities bloats Temporal storage. Keep a single ingest activity with small I/O and document anti-patterns in the READMEs. Co-authored-by: Cursor --- template/Makefile.jinja | 7 +- template/src/examples/workflows/README.md | 54 ++--- .../src/examples/workflows/search/README.md | 48 ++--- .../workflows/search_pipeline/README.md | 85 -------- .../workflows/search_pipeline/__init__.py | 5 - .../workflows/search_pipeline/activities.py | 197 ------------------ .../workflows/search_pipeline/models.py | 22 -- .../workflows/search_pipeline/workflow.py | 68 ------ template/src/examples/workflows/start.py | 2 +- template/src/examples/workflows/worker.py | 3 +- 10 files changed, 48 insertions(+), 443 deletions(-) delete mode 100644 template/src/examples/workflows/search_pipeline/README.md delete mode 100644 template/src/examples/workflows/search_pipeline/__init__.py delete mode 100644 template/src/examples/workflows/search_pipeline/activities.py delete mode 100644 template/src/examples/workflows/search_pipeline/models.py delete mode 100644 template/src/examples/workflows/search_pipeline/workflow.py diff --git a/template/Makefile.jinja b/template/Makefile.jinja index c1c4ae8..dac49be 100644 --- a/template/Makefile.jinja +++ b/template/Makefile.jinja @@ -1,4 +1,4 @@ -.PHONY: installdeps install-workflows setup-vespa start-vespa stop-vespa reset-vespa migrate-vespa verify-vespa ingest search bruno generate-vespa-lock start-examples execute-ingestion execute-pipeline-ingestion +.PHONY: installdeps install-workflows setup-vespa start-vespa stop-vespa reset-vespa migrate-vespa verify-vespa ingest search bruno generate-vespa-lock start-examples execute-ingestion ifneq (,$(wildcard .env)) include .env @@ -78,8 +78,3 @@ start-examples: install-workflows ## Usage: make execute-ingestion input='{"file_path": "sample_data/hello.txt", "collection_name": "mydocs"}' execute-ingestion: install-workflows uv run python -m examples.workflows.start --workflow document-ingestion $(if $(input),--input '$(input)',--input '{"file_path":"sample_data/hello.txt"}') - -## Execute the advanced pipeline-step ingestion workflow -## Usage: make execute-pipeline-ingestion input='{"file_path": "sample_data/hello.txt", "collection_name": "mydocs"}' -execute-pipeline-ingestion: install-workflows - uv run python -m examples.workflows.start --workflow document-ingestion-pipeline $(if $(input),--input '$(input)',--input '{"file_path":"sample_data/hello.txt"}') diff --git a/template/src/examples/workflows/README.md b/template/src/examples/workflows/README.md index 6f90f01..aae6344 100644 --- a/template/src/examples/workflows/README.md +++ b/template/src/examples/workflows/README.md @@ -4,14 +4,13 @@ This folder contains examples that integrate the search-starter-app with the [Mi Workflow-based examples live under `examples/workflows/`. Other examples that do not use the workflows framework (standalone scripts, direct API calls, etc.) live directly under `examples/`. -## Available examples +## Available example | Example | Workflow name | Makefile target | Description | | --- | --- | --- | --- | -| [search/](search/) | `document-ingestion` | `make execute-ingestion` | Basic: 2 activities wrapping the full pipeline | -| [search_pipeline/](search_pipeline/) | `document-ingestion-pipeline` | `make execute-pipeline-ingestion` | Advanced: one activity per pipeline stage | +| [search/](search/) | `document-ingestion` | `make execute-ingestion` | Ingestion workflow with small Temporal I/O | -Both examples ingest local files into Vespa using the same Search Toolkit components as `make ingest`. Choose the basic example to get started; use the pipeline example when you want per-step retries, timeouts, and a detailed timeline in the Mistral Console. +The example ingests local files into Vespa using the same Search Toolkit components as `make ingest`, wrapped in a durable workflow. ## Prerequisites @@ -31,8 +30,6 @@ Ensure these are set in your `.env` file: ## Workflow input -Both workflows accept the same JSON input: - ```json { "file_path": "sample_data/hello.txt", @@ -55,21 +52,17 @@ Both workflows accept the same JSON input: make start-examples ``` -Wait until the worker is listening for execution requests. It registers all workflows listed in `worker.py`. +Wait until the worker is listening for execution requests. -**Terminal 2 — trigger a workflow:** +**Terminal 2 — trigger the workflow:** ```bash -# Basic example (2 activities) make execute-ingestion make execute-ingestion input='{"file_path": "sample_data/hello.txt"}' - -# Advanced example — one activity per pipeline stage -make execute-pipeline-ingestion -make execute-pipeline-ingestion input='{"file_path": "sample_data", "collection_name": "mydocs"}' +make execute-ingestion input='{"file_path": "sample_data", "collection_name": "mydocs"}' ``` -You can also trigger workflows from the [Mistral Console](https://console.mistral.ai/build/workflows): select `document-ingestion` or `document-ingestion-pipeline`, click **Start Workflow**, and provide the input JSON. +You can also trigger the workflow from the [Mistral Console](https://console.mistral.ai/build/workflows): select `document-ingestion`, click **Start Workflow**, and provide the input JSON. After ingestion, search directly (no workflow): @@ -84,12 +77,7 @@ examples/workflows/ ├── README.md # This file ├── worker.py # Registers and runs all workflow examples ├── start.py # CLI to trigger a workflow execution -├── search/ # Basic example (2 activities) -│ ├── models.py -│ ├── activities.py -│ ├── workflow.py -│ └── README.md -└── search_pipeline/ # Advanced example (7 activities) +└── search/ # Document ingestion workflow ├── models.py ├── activities.py ├── workflow.py @@ -100,8 +88,20 @@ examples/workflows/ - **Workflows for ingestion** — document ingestion is long-running and benefits from durability, retries, and observability in the Mistral Console. - **Search stays direct** — search queries use `make search` / `entrypoints/search.py` for low latency. Workflows are not needed for simple queries. +- **Small activity I/O** — the workflow passes file paths and collection names only. The full Search Toolkit pipeline runs inside a single `ingest_documents` activity; the activity returns a small result (`total_chunks`, `file_count`, …). - **Activities own all I/O** — filesystem access, API calls, and Vespa writes live in `activities.py`. The workflow body in `workflow.py` only orchestrates. -- **Granular vs bundled activities** — the basic example bundles the pipeline into fewer activities; the pipeline example exposes each stage separately for production-style observability. The pipeline example passes serialized document data between steps and is best suited to **small files** (~2 MB activity payload limit); see [search_pipeline/README.md](search_pipeline/README.md#large-files-and-payload-limits). + +## Designing ingestion workflows at scale + +Temporal stores every activity input and output in Postgres. **Do not split ingestion into one activity per pipeline stage** if each step passes document bytes, text, chunks, or embeddings — that duplicates large payloads in workflow history and does not scale. + +| Step | Cost | Where results should live | +| --- | --- | --- | +| Load from remote source | Medium | Object storage (S3) keyed by source URI + version | +| Extract plain text, split | Cheap | Recompute inside the activity on retry — do not cache in Temporal | +| OCR, embeddings | Expensive | External cache (Redis) keyed by `hash(file_bytes)`, not the raw content — implement outside this starter example | + +**Anti-pattern:** one `@workflows.activity` per pipeline stage with serialized `File` / `Document` dicts passed between them. That pattern only works for tiny demo files and fills Temporal storage quickly. ## Adding a new workflow example @@ -111,17 +111,6 @@ examples/workflows/ 4. Register it in `EXAMPLE_WORKFLOWS` in `worker.py`. 5. Add a Makefile target that calls `examples.workflows.start --workflow `. -Example structure: - -```text -examples/workflows/my_example/ -├── __init__.py -├── models.py -├── activities.py -├── workflow.py -└── README.md -``` - ## Troubleshooting | Issue | Fix | @@ -129,7 +118,6 @@ examples/workflows/my_example/ | `Workflow not found` | Start `make start-examples` in a separate terminal, then retry | | `DEPLOYMENT_NAME is required` | Add `DEPLOYMENT_NAME=` to `.env` | | Worker fails on startup | Ensure `imports_passed_through()` wraps activity imports in `workflow.py` when activities use `mistralai.client` | -| `unserializable_payload_error` / payload too large | Advanced pipeline passes full document dicts between activities (~2 MB limit). Use `make ingest`, the basic `document-ingestion` workflow, or OffloadableField + blob storage for larger files — see [search_pipeline/README.md](search_pipeline/README.md#large-files-and-payload-limits) | | Vespa errors during indexing | Run `make setup-vespa` before triggering ingestion | Enable verbose logging: diff --git a/template/src/examples/workflows/search/README.md b/template/src/examples/workflows/search/README.md index 4c4232f..489565c 100644 --- a/template/src/examples/workflows/search/README.md +++ b/template/src/examples/workflows/search/README.md @@ -2,29 +2,30 @@ This example wraps the search-starter-app ingestion pipeline in a [Mistral Workflow](https://docs.mistral.ai/capabilities/workflows/), adding durability, observability, and retry support. +It follows the recommended pattern for ingestion at scale: **small activity I/O**, with the full Search Toolkit pipeline running inside a single activity. + ## What It Demonstrates | Primitive | Where | -|-----------|-------| -| `@workflows.workflow.define` | `workflow.py` — defines the workflow with a name visible in the Mistral Console | -| `@workflows.workflow.entrypoint` | `workflow.py` — pure orchestration, no I/O | +| --- | --- | +| `@workflows.workflow.define` | `workflow.py` — workflow registered in the Mistral Console | +| `@workflows.workflow.entrypoint` | `workflow.py` — orchestration only; passes paths, not document content | | `@workflows.activity` | `activities.py` — all file I/O, embedding calls, and Vespa writes | -| Pydantic input/output models | `models.py` — typed boundaries between workflow and activities | -| `workflows.run_worker()` | `worker.py` — registers workflows with the Temporal engine | +| Pydantic input/output models | `models.py` — typed boundaries; workflow output is a small summary | +| `workflows.run_worker()` | `worker.py` — registers the workflow with Temporal | ## Architecture ``` IngestionWorkflow.run(IngestionInput) -└── ingest_documents activity - ├── FilesystemFileLoader [loads documents] - ├── PlainTextExtractor [for .txt, .md, .csv, .json files] - │ or MistralOCRExtractor [for PDFs and images] - ├── MarkdownTextSplitter [chunks the content] - ├── MistralEmbedder [generates vector embeddings] - └── VespaSearchIndex [stores chunks in Vespa] +├── collect_document_paths activity → list[str] paths only +└── ingest_documents activity → IngestionResult (counts + status) + └── Search Toolkit Pipeline: + load → extract → split → embed → index ``` +The workflow never sees file bytes, extracted text, chunks, or embeddings — only paths and the final chunk count. + ## Prerequisites 1. Install the workflows extra: @@ -37,7 +38,7 @@ IngestionWorkflow.run(IngestionInput) make setup-vespa ``` -3. Set `MISTRAL_API_KEY` in your `.env` file. +3. Set `MISTRAL_API_KEY` and `DEPLOYMENT_NAME` in your `.env` file. ## Running the Example @@ -50,24 +51,23 @@ make start-examples **Terminal 2 — trigger ingestion:** ```bash -# Ingest a single file make execute-ingestion input='{"file_path": "sample_data/hello.txt"}' - -# Ingest a directory make execute-ingestion input='{"file_path": "sample_data"}' - -# Use a custom collection name make execute-ingestion input='{"file_path": "sample_data/hello.txt", "collection_name": "mydocs"}' ``` -You can also trigger the workflow from the [Mistral Console](https://console.mistral.ai/build/workflows) by selecting `document-ingestion` and providing the input JSON. +You can also trigger the workflow from the [Mistral Console](https://console.mistral.ai/build/workflows) by selecting `document-ingestion`. + +## What not to do + +Do **not** split ingestion into one activity per pipeline stage while passing serialized documents between them. Temporal persists every activity input/output in Postgres; that pattern duplicates large payloads and does not scale. See the parent [workflows README](../README.md#designing-ingestion-workflows-at-scale). ## Key Difference from Direct Ingestion -The existing `make ingest` command runs ingestion synchronously in the terminal. The workflow approach adds: +`make ingest` runs the same Search Toolkit pipeline synchronously. The workflow adds: -- **Durability**: the worker can restart mid-ingestion and resume from the last completed step -- **Observability**: each execution is visible in the Mistral Console with a full timeline -- **Retries**: the activity retries automatically on transient errors (network blips, API rate limits) +- **Durability** — the worker can restart mid-ingestion and resume from the last completed activity +- **Observability** — executions appear in the Mistral Console +- **Retries** — the activity retries on transient errors without bloating Temporal with per-step document payloads -Search operations remain direct queries — they don't benefit from workflow orchestration since they complete in milliseconds. +Search queries remain direct (`make search`) — they do not need workflow orchestration. diff --git a/template/src/examples/workflows/search_pipeline/README.md b/template/src/examples/workflows/search_pipeline/README.md deleted file mode 100644 index 791a370..0000000 --- a/template/src/examples/workflows/search_pipeline/README.md +++ /dev/null @@ -1,85 +0,0 @@ -# Document Ingestion Pipeline (Advanced) - -Advanced workflow example where **each Search Toolkit pipeline stage is its own activity**. Use this when you want per-step retries, timeouts, and a detailed timeline in the Mistral Console. - -## Comparison with the basic example - -| | [search/](../search/) (basic) | search_pipeline/ (this example) | -| --- | --- | --- | -| Workflow name | `document-ingestion` | `document-ingestion-pipeline` | -| Activities | 2 (`collect_document_paths`, `ingest_documents`) | 7 (one per pipeline stage) | -| Console timeline | Coarse-grained | Fine-grained: load → extract → split → embed → index | -| Best for | Getting started, minimal boilerplate | Production patterns, per-step retry policies | - -## Pipeline stages (activities) - -``` -For each file: - pipeline_load_document → FilesystemFileLoader - pipeline_extract_plain_text → PlainTextExtractor (.txt, .md, .csv, …) - or pipeline_extract_with_ocr → MistralOCRExtractor (PDFs, images) - pipeline_split_document → MarkdownTextSplitter - pipeline_embed_document → MistralEmbedder - pipeline_index_document → VespaSearchIndex -``` - -## Workflow input - -```json -{ - "file_path": "sample_data/hello.txt", - "collection_name": "exampledocs" -} -``` - -## Run - -**Terminal 1:** -```bash -make start-examples -``` - -**Terminal 2:** -```bash -make execute-pipeline-ingestion -make execute-pipeline-ingestion input='{"file_path": "sample_data", "collection_name": "mydocs"}' -``` - -Or from the [Mistral Console](https://console.mistral.ai/build/workflows): select `document-ingestion-pipeline`. - -## What to look for in the Console - -Each file should produce a sequence of activity events: - -1. **pipeline_load_document** — file read from disk -2. **pipeline_extract_plain_text** or **pipeline_extract_with_ocr** — text extraction -3. **pipeline_split_document** — chunking -4. **pipeline_embed_document** — embedding API call -5. **pipeline_index_document** — Vespa write - -If a step fails (e.g. transient OCR error), only that activity retries — completed steps are not re-run. - -## Large files and payload limits - -This example passes **serialized file and document data** between activities (see `_file_to_dict` / `_document_to_dict` in `activities.py`). Workflows enforce Temporal's **~2 MB limit** on each activity input and output. - -Payload size grows at every stage: - -| Stage | What crosses the activity boundary | -| --- | --- | -| load → extract | Raw file bytes (JSON-serialised) | -| extract → split | Full extracted text | -| split → embed | Text duplicated in every chunk | -| embed → index | Chunk text + embedding vectors | - -**This example is intended for small files** (e.g. `sample_data/hello.txt`) where the goal is per-step retries, timeouts, and a detailed timeline in the Mistral Console. - -### What to use instead for larger files - -| Scenario | Recommendation | -| --- | --- | -| Local files or directories, no per-step Console timeline needed | `make ingest` | -| Workflow durability with minimal payload size | [Basic example](../search/) (`document-ingestion`) — passes only file paths; the full pipeline runs inside one activity | -| Production ingestion with per-step activities and large documents | Pass **references** (paths, S3/GCS URIs, artifact IDs) between activities instead of inline payloads, or use **OffloadableField** with blob storage — see the [Handling Large Data](https://docs.mistral.ai/capabilities/workflows/guides/handling-large-data/) guide | - -We deliberately keep this starter example simple (inline dicts, no cloud storage setup). diff --git a/template/src/examples/workflows/search_pipeline/__init__.py b/template/src/examples/workflows/search_pipeline/__init__.py deleted file mode 100644 index a0a4cf8..0000000 --- a/template/src/examples/workflows/search_pipeline/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -"""Advanced search ingestion workflow — one activity per pipeline step.""" - -from .workflow import PipelineIngestionWorkflow - -__all__ = ["PipelineIngestionWorkflow"] diff --git a/template/src/examples/workflows/search_pipeline/activities.py b/template/src/examples/workflows/search_pipeline/activities.py deleted file mode 100644 index 41b4f7e..0000000 --- a/template/src/examples/workflows/search_pipeline/activities.py +++ /dev/null @@ -1,197 +0,0 @@ -"""Activities for the pipeline-step ingestion workflow. - -Each Search Toolkit pipeline stage is a separate durable activity so the Mistral -Console timeline shows load → extract → split → embed → index as distinct steps -with independent retry policies. -""" - -import os -from datetime import timedelta -from functools import cache -from pathlib import Path -from typing import Any - -import mistralai.workflows as workflows -from mistralai.client import Mistral -from mistralai.search.toolkit.context import IngestContext -from mistralai.search.toolkit.document import Document -from mistralai.search.toolkit.embedders import MistralEmbedder -from mistralai.search.toolkit.ingestion import File -from mistralai.search.toolkit.ingestion.extractors import ( - MistralOCRExtractor, - PlainTextExtractor, -) -from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader -from mistralai.search.toolkit.ingestion.text_splitters import ( - MarkdownTextSplitter, - MarkdownTextSplitterConfig, -) -from mistralai.search.toolkit.plugins.vespa import VespaClientConfig -from mistralai.workflows import Depends -from vespa_app import app, vespa_endpoint - - -@cache -def get_mistral_client() -> Mistral: - api_key = os.environ.get("MISTRAL_API_KEY", "") - if not api_key: - raise ValueError("MISTRAL_API_KEY is not set. Check your .env file.") - return Mistral( - api_key=api_key, - server_url=os.getenv("MISTRAL_API_URL", "https://api.mistral.ai"), - ) - - -def get_embedder() -> MistralEmbedder: - return MistralEmbedder(client=get_mistral_client()) - - -def get_ocr_extractor() -> MistralOCRExtractor: - return MistralOCRExtractor(client=get_mistral_client()) - - -def get_loader() -> FilesystemFileLoader: - return FilesystemFileLoader() - - -def get_text_splitter() -> MarkdownTextSplitter: - return MarkdownTextSplitter( - MarkdownTextSplitterConfig(chunk_size=4096, chunk_overlap=50) - ) - - -def get_plain_text_extractor() -> PlainTextExtractor: - return PlainTextExtractor() - - -def _file_to_dict(file: File) -> dict[str, Any]: - return file.model_dump(mode="json") - - -def _file_from_dict(data: dict[str, Any]) -> File: - return File.model_validate(data) - - -def _document_to_dict(document: Document) -> dict[str, Any]: - return document.model_dump(mode="json") - - -def _document_from_dict(data: dict[str, Any]) -> Document: - return Document.model_validate(data) - - -@workflows.activity( - name="pipeline_collect_document_paths", - start_to_close_timeout=timedelta(seconds=30), - retry_policy_max_attempts=1, -) -async def collect_document_paths(file_path: str) -> list[str]: - """List all files under a path (file or directory).""" - root = Path(file_path) - if root.is_file(): - documents = [root] - elif root.is_dir(): - documents = sorted(p for p in root.rglob("*") if p.is_file()) - else: - raise ValueError(f"Path not found: {file_path}") - - if not documents: - raise ValueError(f"No files found at: {file_path}") - - return [str(p) for p in documents] - - -@workflows.activity( - name="pipeline_load_document", - start_to_close_timeout=timedelta(minutes=1), - retry_policy_max_attempts=2, -) -async def load_document( - file_path: str, - loader: FilesystemFileLoader = Depends(get_loader), -) -> dict[str, Any]: - """Load a single file from disk into a serialisable File payload.""" - file = await loader.load_file(file_path) - return _file_to_dict(file) - - -@workflows.activity( - name="pipeline_extract_plain_text", - start_to_close_timeout=timedelta(minutes=2), - retry_policy_max_attempts=2, -) -async def extract_plain_text( - file_data: dict[str, Any], - extractor: PlainTextExtractor = Depends(get_plain_text_extractor), -) -> dict[str, Any]: - """Extract text from plain-text formats (.txt, .md, .csv, .json, …).""" - file = _file_from_dict(file_data) - document = await extractor.extract(file, context=IngestContext()) - return _document_to_dict(document) - - -@workflows.activity( - name="pipeline_extract_with_ocr", - start_to_close_timeout=timedelta(minutes=10), - retry_policy_max_attempts=3, -) -async def extract_with_ocr( - file_data: dict[str, Any], - ocr_extractor: MistralOCRExtractor = Depends(get_ocr_extractor), -) -> dict[str, Any]: - """Extract text from PDFs and images via Mistral OCR.""" - file = _file_from_dict(file_data) - document = await ocr_extractor.extract(file, context=IngestContext()) - return _document_to_dict(document) - - -@workflows.activity( - name="pipeline_split_document", - start_to_close_timeout=timedelta(minutes=2), - retry_policy_max_attempts=2, -) -async def split_document( - document_data: dict[str, Any], - text_splitter: MarkdownTextSplitter = Depends(get_text_splitter), -) -> dict[str, Any]: - """Split an extracted document into chunks.""" - document = _document_from_dict(document_data) - chunks = text_splitter.split_document(document, context=IngestContext()) - return _document_to_dict(document.model_copy(update={"chunks": chunks})) - - -@workflows.activity( - name="pipeline_embed_document", - start_to_close_timeout=timedelta(minutes=5), - retry_policy_max_attempts=3, -) -async def embed_document( - document_data: dict[str, Any], - embedder: MistralEmbedder = Depends(get_embedder), -) -> dict[str, Any]: - """Generate vector embeddings for all chunks in a document.""" - document = _document_from_dict(document_data) - if not document.chunks: - return document_data - - embedded_chunks = await embedder.embed_chunks(list(document.chunks)) - return _document_to_dict(document.model_copy(update={"chunks": embedded_chunks})) - - -@workflows.activity( - name="pipeline_index_document", - start_to_close_timeout=timedelta(minutes=5), - retry_policy_max_attempts=2, -) -async def index_document( - document_data: dict[str, Any], - collection_name: str, -) -> int: - """Write an embedded document to Vespa and return the number of chunks indexed.""" - document = _document_from_dict(document_data) - vector_store = app.get_search_index( - VespaClientConfig(endpoint=vespa_endpoint()), - collection_name=collection_name, - ) - await vector_store.index_document(document, context=IngestContext()) - return len(document.chunks) diff --git a/template/src/examples/workflows/search_pipeline/models.py b/template/src/examples/workflows/search_pipeline/models.py deleted file mode 100644 index fce75a0..0000000 --- a/template/src/examples/workflows/search_pipeline/models.py +++ /dev/null @@ -1,22 +0,0 @@ -"""Pydantic models for the pipeline-step ingestion workflow.""" - -from pydantic import BaseModel, Field - - -class PipelineIngestionInput(BaseModel): - """Input accepted by the document-ingestion-pipeline workflow.""" - - file_path: str = Field(description="Local path to a file or directory to ingest") - collection_name: str = Field( - default="exampledocs", - description="Vespa collection to index documents into", - ) - - -class PipelineIngestionResult(BaseModel): - """Output returned after all documents are processed.""" - - status: str = Field(description="'success' or 'error'") - total_chunks: int = Field(description="Total number of chunks indexed") - file_count: int = Field(description="Number of files processed") - collection_name: str = Field(description="Vespa collection that was used") diff --git a/template/src/examples/workflows/search_pipeline/workflow.py b/template/src/examples/workflows/search_pipeline/workflow.py deleted file mode 100644 index 6851918..0000000 --- a/template/src/examples/workflows/search_pipeline/workflow.py +++ /dev/null @@ -1,68 +0,0 @@ -"""Advanced ingestion workflow — one activity per pipeline step. - -Demonstrates granular workflow orchestration over the Search Toolkit pipeline: - load → extract → split → embed → index - -Each stage is a separate @workflows.activity with its own timeout and retry -policy. The workflow fans out over files and branches deterministically on -file suffix to pick PlainTextExtractor vs MistralOCRExtractor. -""" - -import mistralai.workflows as workflows -from mistralai.workflows import workflow - -with workflow.unsafe.imports_passed_through(): - from .activities import ( - collect_document_paths, - embed_document, - extract_plain_text, - extract_with_ocr, - index_document, - load_document, - split_document, - ) - -from .models import PipelineIngestionInput, PipelineIngestionResult # noqa: E402 - -_TEXT_SUFFIXES = frozenset({".txt", ".md", ".markdown", ".csv", ".json"}) - - -@workflows.workflow.define( - name="document-ingestion-pipeline", - workflow_display_name="Document Ingestion (Pipeline Steps)", - workflow_description=( - "Advanced ingestion example: each Search Toolkit stage runs as its own " - "activity (load → extract → split → embed → index) with per-step retries " - "and observability in the Mistral Console." - ), -) -class PipelineIngestionWorkflow: - """Orchestrate ingestion with one durable activity per pipeline stage.""" - - @workflows.workflow.entrypoint - async def run(self, params: PipelineIngestionInput) -> PipelineIngestionResult: - paths = await collect_document_paths(file_path=params.file_path) - - total_chunks = 0 - for path in paths: - suffix = path.rsplit(".", 1)[-1].lower() if "." in path else "" - file_data = await load_document(file_path=path) - - if f".{suffix}" in _TEXT_SUFFIXES: - document_data = await extract_plain_text(file_data=file_data) - else: - document_data = await extract_with_ocr(file_data=file_data) - - document_data = await split_document(document_data=document_data) - document_data = await embed_document(document_data=document_data) - total_chunks += await index_document( - document_data=document_data, - collection_name=params.collection_name, - ) - - return PipelineIngestionResult( - status="success", - total_chunks=total_chunks, - file_count=len(paths), - collection_name=params.collection_name, - ) diff --git a/template/src/examples/workflows/start.py b/template/src/examples/workflows/start.py index 6d6605e..7397fc4 100644 --- a/template/src/examples/workflows/start.py +++ b/template/src/examples/workflows/start.py @@ -23,7 +23,7 @@ def parse_args() -> argparse.Namespace: parser.add_argument( "--workflow", default="document-ingestion", - help="Workflow name (document-ingestion or document-ingestion-pipeline)", + help="Workflow name (default: document-ingestion)", ) parser.add_argument( "--input", diff --git a/template/src/examples/workflows/worker.py b/template/src/examples/workflows/worker.py index 686cc1f..60c2461 100644 --- a/template/src/examples/workflows/worker.py +++ b/template/src/examples/workflows/worker.py @@ -12,9 +12,8 @@ import mistralai.workflows as workflows from examples.workflows.search import IngestionWorkflow -from examples.workflows.search_pipeline import PipelineIngestionWorkflow -EXAMPLE_WORKFLOWS = [IngestionWorkflow, PipelineIngestionWorkflow] +EXAMPLE_WORKFLOWS = [IngestionWorkflow] async def main() -> None: From d36c6a29cae4b81d8f5f28b088a477d306f970ec Mon Sep 17 00:00:00 2001 From: Lara Perinetti Date: Thu, 11 Jun 2026 18:31:50 +0200 Subject: [PATCH 3/4] Document per-stage retry pattern via Handling Large Data guide. Point readers to OffloadableField + blob storage when splitting ingestion activities instead of passing document payloads through Temporal. Co-authored-by: Cursor --- template/src/examples/workflows/README.md | 8 ++++++++ template/src/examples/workflows/search/README.md | 2 ++ 2 files changed, 10 insertions(+) diff --git a/template/src/examples/workflows/README.md b/template/src/examples/workflows/README.md index aae6344..8a4de62 100644 --- a/template/src/examples/workflows/README.md +++ b/template/src/examples/workflows/README.md @@ -103,6 +103,14 @@ Temporal stores every activity input and output in Postgres. **Do not split inge **Anti-pattern:** one `@workflows.activity` per pipeline stage with serialized `File` / `Document` dicts passed between them. That pattern only works for tiny demo files and fills Temporal storage quickly. +### Per-stage retries + +This starter example runs the full pipeline inside a single activity so Temporal only stores small inputs and outputs (paths in, chunk counts out). + +If you need to **split the pipeline into separate activities** — for example, independent retry policies on OCR or embedding — you must still keep activity arguments and return values small. Pass references (file paths, S3 URIs, content hashes) between steps, and store intermediate or expensive-step results in external storage rather than in workflow history. + +If you want to split the pipeline to enable retries for some stages, follow the [Handling Large Data](https://docs.mistral.ai/capabilities/workflows/guides/handling-large-data/) guide (`OffloadableField` + blob storage for activity payloads). + ## Adding a new workflow example 1. Create a subdirectory under `examples/workflows/` (e.g. `my_example/`). diff --git a/template/src/examples/workflows/search/README.md b/template/src/examples/workflows/search/README.md index 489565c..26a17dd 100644 --- a/template/src/examples/workflows/search/README.md +++ b/template/src/examples/workflows/search/README.md @@ -62,6 +62,8 @@ You can also trigger the workflow from the [Mistral Console](https://console.mis Do **not** split ingestion into one activity per pipeline stage while passing serialized documents between them. Temporal persists every activity input/output in Postgres; that pattern duplicates large payloads and does not scale. See the parent [workflows README](../README.md#designing-ingestion-workflows-at-scale). +If you want to split the pipeline to enable retries for some stages, follow the [Handling Large Data](https://docs.mistral.ai/capabilities/workflows/guides/handling-large-data/) guide instead of passing document content between activities. + ## Key Difference from Direct Ingestion `make ingest` runs the same Search Toolkit pipeline synchronously. The workflow adds: From e999b60c47fd36a0a6851d77ba82ec8d2f6b015c Mon Sep 17 00:00:00 2001 From: Lara Perinetti Date: Wed, 24 Jun 2026 14:32:53 +0200 Subject: [PATCH 4/4] Remove broken Handling Large Data doc link. Co-Authored-By: Claude Sonnet 4.6 --- template/src/examples/workflows/README.md | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/template/src/examples/workflows/README.md b/template/src/examples/workflows/README.md index 8a4de62..e2c6712 100644 --- a/template/src/examples/workflows/README.md +++ b/template/src/examples/workflows/README.md @@ -87,7 +87,7 @@ examples/workflows/ ## Design principles - **Workflows for ingestion** — document ingestion is long-running and benefits from durability, retries, and observability in the Mistral Console. -- **Search stays direct** — search queries use `make search` / `entrypoints/search.py` for low latency. Workflows are not needed for simple queries. +- **Search stays direct** — search queries use `make search` / `entrypoints/search.py` for low latency. Workflows are not needed for queries. - **Small activity I/O** — the workflow passes file paths and collection names only. The full Search Toolkit pipeline runs inside a single `ingest_documents` activity; the activity returns a small result (`total_chunks`, `file_count`, …). - **Activities own all I/O** — filesystem access, API calls, and Vespa writes live in `activities.py`. The workflow body in `workflow.py` only orchestrates. @@ -109,8 +109,6 @@ This starter example runs the full pipeline inside a single activity so Temporal If you need to **split the pipeline into separate activities** — for example, independent retry policies on OCR or embedding — you must still keep activity arguments and return values small. Pass references (file paths, S3 URIs, content hashes) between steps, and store intermediate or expensive-step results in external storage rather than in workflow history. -If you want to split the pipeline to enable retries for some stages, follow the [Handling Large Data](https://docs.mistral.ai/capabilities/workflows/guides/handling-large-data/) guide (`OffloadableField` + blob storage for activity payloads). - ## Adding a new workflow example 1. Create a subdirectory under `examples/workflows/` (e.g. `my_example/`).