Scaling Ingestion Workers: The Data Factory

Ingesting 10 documents is easy. Ingesting 10 million documents requires a Factory. The extraction of entities and relationships is a "CPU-heavy" and "Network-heavy" task. If you run it on a single server, it will take months to finish. To scale, you must architect a Distributed Worker Pool.

In this lesson, we will look at how to build an Ingestion Pipeline that can handle thousands of documents per second. We will learn how to use Celery and Redis to manage a queue of "Jobs," how to handle Rate Limiting for your LLM APIs (OpenAI/Gemini), and how to implement Idempotent Graph Writing (ensuring that re-running a job doesn't create duplicate nodes).

---

1. The Distributed Worker Architecture

1. The Producer: A script that finds new files and adds their IDs to a Redis queue.
2. The Workers: 50+ Python containers that take an ID from the queue, download the file, and perform the AI extraction.
3. The Combiner: A single "Writer" service (or a well-indexed graph) that merges the new facts into the Knowledge Graph.

---

2. Managing the LLM Bottleneck

The biggest bottleneck in scaling isn't the graph database—it's the LLM Extraction Rate.

• If OpenAI allows you 1,000 requests per minute, but you have 10,000 documents, your workers will hit a "Rate Limit" error.
• The Solution: Use a Token Budgeter or a "Retry with Exponential Backoff" strategy in your workers. This ensures that the factory never "Explodes" when it hits the ceiling.

---

3. Idempotent Graph Writing: The 'MERGE' Rule

In a distributed system, a worker might fail and restart. If it does, it might try to write the same fact a second time.

• The Disaster: You end up with 50 nodes for "Sudeep."
• The Solution: Always use the Cypher MERGE command instead of CREATE.

MERGE (p:Person {email: $email}) SET p.name = $name This command says: "Find if this person exists. If yes, update them. If no, create them."

graph LR
    D[S3: Raw Docs] --> P[Producer]
    P --> Q[Redis Queue]
    Q --> W1[Worker A]
    Q --> W2[Worker B]
    Q --> W3[Worker C]
    W1 & W2 & W3 --> DB[(Neo4j Graph)]
    
    style Q fill:#f44336,color:#fff
    style DB fill:#4285F4,color:#fff
    note[Scaling is as simple as adding more Workers]

---

4. Implementation: A Scaling Worker with Celery

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task(autoretry_for=(Exception,), retry_kwargs={'max_retries': 5})
def ingest_document(doc_path):
    # 1. AI Extraction (The slow part)
    triplets = extract_knowledge(doc_path)
    
    # 2. Graph Writing (The atomic part)
    for t in triplets:
        write_triplet_to_graph(t)
        
    print(f"Document {doc_path} successfully ingested into graph.")

---

5. Summary and Exercises

Scaling is about Managing Flow.

• Queues decouple the data source from the processing speed.
• Workers provide the parallel "Brain Power" for extraction.
• Idempotency ensures the Knowledge Graph remains clean regardless of restarts.
• Rate Limiting is the most critical coordination task in the modern AI pipeline.

Exercises

1. Scale Math: If one worker can process 1 document per minute, how many workers do you need to ingest 60,000 documents in one day? (Hint: 24 hours * 60 minutes = 1440 mins).
2. Failure Scenario: A worker crashes halfway through a document. What happens to the triplets it already wrote to the graph? (Hint: Think about "Transactions").
3. Visualization: Draw a factory line with 5 machines. Show what happens if the "Conveyor Belt" (Queue) stops.

In the next lesson, we will look at the human side: The Front-End for Graph Discovery.