A production-quality, asynchronous document ingestion pipeline built with Temporal.io, Unstructured.io, OpenAI Embeddings, and Milvus vector database.
Documents are fetched from a URL, parsed into text chunks, embedded using text-embedding-3-large, and stored in Milvus — all orchestrated as a durable, retryable Temporal workflow.
- Architecture Overview
- Prerequisites
- Project Structure
- Infrastructure Setup
- Python Environment Setup
- Configuration
- Running the Worker
- Triggering the Workflow
- Monitoring
- Design Explanation
- Error Handling Strategy
- Milvus Schema Design
- Asyncio Concurrency Model
- Assumptions & Limitations
trigger.py
│
│ start_workflow(file_id, file_url)
▼
┌──────────────────────────────────────────────────────────────────┐
│ Temporal Server │
│ │
│ DocumentWorkflow │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ 1. Validate extension │ │
│ │ 2. fetch_document(url) → FetchResult │ │
│ │ 3. parse_document(bytes, ext) → ParseResult │ │
│ │ 4. generate_embedding × N ← asyncio.gather() │ │
│ │ 5. store_chunk × N ← asyncio.gather() │ │
│ └──────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────┘
│ │ │
▼ ▼ ▼
httpx download OpenAI API Milvus standalone
(async) text-embedding-3-large (pymilvus)
| Tool | Version | Notes |
|---|---|---|
| Docker & Docker Compose | 20.10+ | For Temporal + Milvus infrastructure |
| Python | 3.10+ | Worker and trigger scripts |
| pip | 23+ | Dependency management |
unstructured[all-docs] calls system libraries for PDF/Office parsing:
macOS (Homebrew)
brew install poppler tesseract libmagicUbuntu / Debian
sudo apt-get install -y poppler-utils tesseract-ocr libmagic1Windows
Use WSL2 (recommended) or install:
- Poppler for Windows
- Tesseract for Windows
- Ensure both are on your
PATH
temporal-rag-pipeline/
├── docker-compose.yml # Full infrastructure stack
├── temporal-config/
│ └── development-sql.yaml # Temporal dynamic config
├── requirements.txt # Pinned Python dependencies
├── .env.example # Environment variable template
├── worker.py # Temporal worker entry-point
├── trigger.py # CLI to submit a workflow run
├── workflows/
│ └── document_workflow.py # DocumentWorkflow definition
├── activities/
│ ├── fetch_activity.py # Download document from URL
│ ├── parse_activity.py # Parse bytes → text chunks
│ ├── embed_activity.py # Generate OpenAI embeddings
│ └── store_activity.py # Write to Milvus
├── milvus/
│ └── schema.py # Collection schema + init helpers
└── utils/
├── logger.py # Structured logging setup
└── file_validator.py # Extension validation helpers
docker-compose up -dThis starts:
- PostgreSQL — Temporal persistence backend
- Elasticsearch — Temporal visibility store (enables workflow search in UI)
- Temporal Server — workflow engine (port
7233) - Temporal UI — web dashboard (port
8080) - etcd — Milvus metadata store
- MinIO — Milvus object storage
- Milvus Standalone — vector database (port
19530)
docker-compose psAll services should show healthy or running. Temporal may take ~30–60 seconds
for its auto-setup to complete on first run.
Open http://localhost:8080 in your browser. You should see the Temporal Web UI with an empty workflow list.
docker-compose downdocker-compose down -vpython -m venv .venv
# macOS / Linux
source .venv/bin/activate
# Windows (PowerShell)
.venv\Scripts\Activate.ps1pip install -r requirements.txtNote:
unstructured[all-docs]is a large package that pulls in many optional parsers. The first install can take several minutes.
cp .env.example .envOPENAI_API_KEY=sk-your-real-key-here
MILVUS_HOST=localhost
MILVUS_PORT=19530
TEMPORAL_HOST=localhost
TEMPORAL_PORT=7233
MILVUS_COLLECTION_NAME=document_chunks
TEMPORAL_TASK_QUEUE=rag-pipeline
MAX_CHUNK_SIZE=1000The only value you must change is OPENAI_API_KEY. All others default to
the local docker-compose setup.
python worker.pyOn startup, the worker will:
- Load
.env - Connect to the Temporal server
- Ensure the Milvus
document_chunkscollection exists (created if absent) - Register
DocumentWorkflow+ all four activities on therag-pipelinetask queue - Begin polling for workflow and activity tasks
Expected output:
2024-05-01 12:00:00 INFO __main__ — Connecting to Temporal at localhost:7233…
2024-05-01 12:00:00 INFO __main__ — Connected to Temporal.
2024-05-01 12:00:01 INFO __main__ — Initialising Milvus collection 'document_chunks'…
2024-05-01 12:00:01 INFO milvus.schema — Collection 'document_chunks' created with HNSW index…
2024-05-01 12:00:01 INFO __main__ — Worker is running and polling for tasks…
Keep the worker running in a terminal window while you trigger workflows.
Open a second terminal (with the same .venv activated):
python trigger.py \
--file-id "attention-paper-001" \
--file-url "https://arxiv.org/pdf/1706.03762"# Word document
python trigger.py \
--file-id "report-q4-2024" \
--file-url "https://file-examples.com/storage/fe7c3cb03766b4278975d05/2017/02/file-sample_100kB.docx"
# Excel spreadsheet
python trigger.py \
--file-id "financials-2024" \
--file-url "https://file-examples.com/storage/fe7c3cb03766b4278975d05/2017/02/file_example_XLS_10.xls"
# Specify an explicit workflow ID (useful for idempotency)
python trigger.py \
--file-id "my-doc" \
--file-url "https://example.com/document.pdf" \
--workflow-id "ingest-my-doc-v1"────────────────────────────────────────────────────────────
Triggering DocumentWorkflow
Workflow ID : rag-attention-paper-001-a3f2b1c0
File ID : attention-paper-001
File URL : https://arxiv.org/pdf/1706.03762
Task Queue : rag-pipeline
────────────────────────────────────────────────────────────
════════════════════════════════════════════════════════════
✓ Workflow completed successfully!
════════════════════════════════════════════════════════════
File ID : attention-paper-001
Extension : .pdf
Chunks : 47
Message : Successfully ingested 47 chunks from .pdf document (file_id=attention-paper-001).
════════════════════════════════════════════════════════════
python trigger.py --file-id "img" --file-url "https://example.com/photo.jpg"
# Exit code: 1
# Error Type: UnsupportedFileType
# Message: Unsupported file type '.jpg'. Supported types are: .doc, .docx, .pdf, .xls, .xlsx.Navigate to http://localhost:8080 to:
- View running and completed workflows
- Inspect workflow event history (every activity start/complete/fail)
- Manually terminate or reset workflows
- Search workflows by status, workflow ID, or type
Use a Python REPL or script to confirm data was stored:
from pymilvus import MilvusClient
client = MilvusClient(uri="http://localhost:19530")
# Count all chunks for a specific file
results = client.query(
collection_name="document_chunks",
filter='file_id == "attention-paper-001"',
output_fields=["file_id", "chunk_index", "chunk_text"],
limit=5,
)
for row in results:
print(f"[{row['chunk_index']}] {row['chunk_text'][:100]}…")
# Total chunk count in the collection
count = client.query(
collection_name="document_chunks",
filter="chunk_index >= 0",
output_fields=["id"],
limit=10000,
)
print(f"Total chunks stored: {len(count)}")# Similarity search — retrieve the 3 most relevant chunks for a query
from openai import OpenAI
oai = OpenAI()
query_vec = oai.embeddings.create(
model="text-embedding-3-large",
input="What is the attention mechanism?"
).data[0].embedding
hits = client.search(
collection_name="document_chunks",
data=[query_vec],
limit=3,
output_fields=["file_id", "chunk_index", "chunk_text"],
)
for hit in hits[0]:
print(f"score={hit['distance']:.4f} | {hit['entity']['chunk_text'][:120]}")The pipeline is decomposed into four single-responsibility Activities and one Workflow that orchestrates them:
| Component | Responsibility | Key library |
|---|---|---|
fetch_document |
HTTP download, type detection | httpx.AsyncClient |
parse_document |
Bytes → ordered text chunks | unstructured.partition |
generate_embedding |
Text → float vector | openai.AsyncOpenAI |
store_chunk |
Vector + text → Milvus | pymilvus.MilvusClient |
DocumentWorkflow |
Orchestration + fan-out | temporalio |
This decomposition means each failure mode is isolated. A transient OpenAI rate limit does not cause the expensive parsing step to re-run — Temporal retries only the failed activity. This is one of the core advantages of the Temporal model over a monolithic script.
Workflow starts
│
├─ [fetch_document] ──────────────────────────────► FetchResult
│
├─ [parse_document] ──────────────────────────────► ParseResult (N chunks)
│
├─ [generate_embedding] × N ──┐ (all fired concurrently via asyncio.gather)
│ ├── chunk 0 ──► EmbedResult
│ ├── chunk 1 ──► EmbedResult
│ └── chunk N ──► EmbedResult
│ ▲ total time ≈ 1 API call RTT, not N × RTT
│
└─ [store_chunk] × N ─────────┐ (all fired concurrently via asyncio.gather)
├── chunk 0 ──► stored
├── chunk 1 ──► stored
└── chunk N ──► stored
▲ total time ≈ 1 Milvus write RTT
- Durability — If the worker crashes mid-run, Temporal reschedules pending activities automatically. No lost progress.
- Retries — Each activity has a retry policy with exponential back-off. A transient OpenAI 429 is retried without any application-level polling.
- Visibility — Every step is recorded in the Temporal event history, making debugging straightforward: you can see exactly which chunk failed and why.
- Decoupling — The trigger script and the worker are completely independent processes. The trigger submits work to the Temporal task queue and can exit before the workflow finishes (or wait for the result, as we do here).
| Activity | Max attempts | Initial interval | Non-retryable errors |
|---|---|---|---|
fetch_document |
5 | 5 s | FileNotFound, AccessDenied, UnsupportedFileType, FileTooLarge |
parse_document |
3 | 3 s | EmptyDocument, EmptyChunks |
generate_embedding |
8 | 10 s | AuthenticationError, BadRequest, EmptyChunk |
store_chunk |
5 | 3 s | — |
OpenAI rate limits (429 Too Many Requests) are handled at two levels:
- tenacity inside
generate_embeddingapplies jittered exponential back-off (up to 4 attempts, max 60 s wait) before the activity itself is considered failed. - If tenacity exhausts its retries, the activity fails and Temporal's retry policy takes over with its own back-off schedule (up to 8 total attempts).
This two-level approach prevents the Temporal retry budget from being consumed by rapid-fire 429s while still eventually giving up if the rate limit persists.
File type validation happens at two points:
- In the workflow (before any activity is called) — fails immediately without spending any resources.
- In the fetch activity — catches files where the extension is absent in the URL but the Content-Type header reveals an unsupported type.
Both raise ApplicationError with non_retryable=True and type="UnsupportedFileType"
so Temporal marks the workflow as failed rather than retrying it.
Collection: document_chunks
─────────────────────────────────────────────────────────────────
Field Type Notes
─────────────────────────────────────────────────────────────────
id INT64 (auto PK) Milvus-generated unique row ID
file_id VARCHAR(256) Links chunk to source document
chunk_index INT64 Preserves intra-document ordering
chunk_text VARCHAR(65535) Raw text (for retrieval + display)
embedding FLOAT_VECTOR(3072) Dense vector from text-embedding-3-large
─────────────────────────────────────────────────────────────────
Index: HNSW on `embedding`
metric_type: COSINE
M: 16 (graph degree — higher = better recall, more RAM)
efConstruction: 200 (build-time search depth — higher = better quality)
Why HNSW over IVF_FLAT?
HNSW does not require a training phase (no fit() call needed) and delivers
excellent recall at low latency for collections up to ~10 M vectors. For an
ingestion pipeline where the collection grows incrementally, HNSW is the
pragmatic default choice.
Why store chunk_text in Milvus?
Storing the raw text alongside the vector means a single Milvus query returns
everything needed to construct a RAG prompt — no secondary database lookup.
The trade-off is storage size, which is acceptable for typical document collections.
A naive implementation might call requests.get(), partition(), the OpenAI REST
API, and MilvusClient.insert() sequentially, each blocking the Python thread
until the network round-trip completes. In a Temporal worker, blocking a thread
also blocks the worker's heartbeat loop, which can cause Temporal to assume the
worker has crashed and reschedule the activity.
The Temporal Python SDK runs on an asyncio event loop. Activities declared as
async def run as coroutines on that loop. When a coroutine reaches an await
expression on a network call:
- The event loop suspends that coroutine.
- The event loop checks if any other coroutine has work to do (e.g. a response has arrived for a different HTTP call).
- When the network response arrives, the original coroutine resumes.
No thread is blocked. A single worker process can handle tens of concurrent in-flight HTTP requests to OpenAI using a single OS thread.
# Sequential (slow): each call waits for the previous one
for chunk in chunks:
result = await workflow.execute_activity("generate_embedding", chunk)
# Concurrent (fast): all calls are in-flight simultaneously
results = await asyncio.gather(*[
workflow.execute_activity("generate_embedding", chunk)
for chunk in chunks
])For a 50-chunk document with a 300 ms OpenAI RTT:
- Sequential: ~50 × 300 ms = 15 seconds
- Concurrent: ~1 × 300 ms = ~300 ms (plus scheduling overhead)
The same pattern applies to the Milvus store fan-out.
unstructured.partition() is CPU-bound and synchronous. Running it directly in
an async function would block the event loop for the entire parsing duration,
preventing heartbeats and starving other coroutines. We solve this with:
elements = await loop.run_in_executor(None, _partition_bytes, file_bytes, ext)run_in_executor(None, ...) submits the work to Python's default
ThreadPoolExecutor, freeing the event loop to handle other tasks while the CPU
work runs on a separate thread.
- OpenAI API key required — You must supply a valid key with access to
text-embedding-3-large. The model is not free; expect ~$0.00013 per 1K tokens. - File size limit — Files larger than 50 MB are rejected by the fetch activity
to prevent OOM conditions. Adjust
MAX_DOWNLOAD_BYTESinfetch_activity.pyif needed. - Milvus standalone — The docker-compose setup uses Milvus standalone (single node). For production, use a clustered Milvus or a managed service like Zilliz.
- No deduplication — Re-triggering the workflow with the same
file_idwill create duplicate chunks in Milvus. A production system should either delete existing rows for thefile_idbefore re-ingesting, or use an upsert strategy. - Temporal namespace — The default
defaultnamespace is used. For production, create a dedicated namespace per environment. - Environment variables — Secrets are loaded from
.envviapython-dotenv. In production, use a secrets manager (AWS Secrets Manager, HashiCorp Vault, etc.) and inject variables as environment variables rather than using a.envfile.