Skip to content

# Task: RAG ingestion worker (events → Milvus + Neo4j)

## Goal

Design and scaffold a RAG ingestion worker that:

- Сonsumes domain events (messages, docs, files, RWA updates) from the existing event stream.
- Transforms them into normalized chunks/documents.
- Indexes them into Milvus (vector store) and Neo4j (graph store).
- Works idempotently and supports `reindex(team_id)`.

This worker complements the `rag-gateway` service (see `docs/cursor/rag_gateway_task.md`) by keeping its underlying stores up-to-date.

> IMPORTANT: This task is about architecture, data flow and scaffolding. Concrete model choices and full schemas can be refined later.

## Context

- Project root: `microdao-daarion/`.
- Planned/implemented RAG layer: see `docs/cursor/rag_gateway_task.md`.
- Existing docs:
- `docs/cursor/42_nats_event_streams_and_event_catalog.md` – event stream & catalog.
- `docs/cursor/34_internal_services_architecture.md` – internal services & topology.

We assume there is (or will be):

- An event bus (likely NATS) with domain events such as:
- `message.created`
- `doc.upsert`
- `file.uploaded`
- `rwa.energy.update`, `rwa.food.update`, etc.
- A Milvus cluster instance.
- A Neo4j instance.

The ingestion worker must not be called directly by agents. It is a back-office service that feeds RAG stores for the `rag-gateway`.

## High-level design

### 1. Service placement & structure

Create a new service (or extend RAG-gateway repo structure) under, for example:

- `services/rag-ingest-worker/`

Suggested files:

- `main.py` — entrypoint (CLI or long-running process).
- `config.py` — environment/config loader (event bus URL, Milvus/Neo4j URLs, batch sizes, etc.).
- `events/consumer.py` — NATS (or other) consumer logic.
- `pipeline/normalization.py` — turn events into normalized documents/chunks.
- `pipeline/embedding.py` — embedding model client/wrapper.
- `pipeline/index_milvus.py` — Milvus upsert logic.
- `pipeline/index_neo4j.py` — Neo4j graph updates.
- `api.py` — optional HTTP API for:
- `POST /ingest/one` – ingest single payload for debugging.
- `POST /ingest/reindex/{team_id}` – trigger reindex job.
- `GET /health` – health check.

### 2. Event sources

The worker should subscribe to a small set of core event types (names to be aligned with the actual Event Catalog):

- `message.created` — messages in chats/channels (Telegram, internal UI, etc.).
- `doc.upsert` — wiki/docs/specs updates.
- `file.uploaded` — files (PDF, images) that have parsed text.
- `rwa.*` — events related to energy/food/water assets (optional, for later).

Implementation details:

- Use NATS (or another broker) subscription patterns from `docs/cursor/42_nats_event_streams_and_event_catalog.md`.
- Each event should carry at least:
- `event_type`
- `team_id` / `dao_id`
- `user_id`
- `channel_id` / `project_id` (if applicable)
- `payload` with text/content and metadata.

## Normalized document/chunk model

Define a common internal model for what is sent to Milvus/Neo4j, e.g. `IngestChunk`:

Fields (minimum):

- `chunk_id` — deterministic ID (e.g. hash of (team_id, source_type, source_id, chunk_index)).
- `team_id` / `dao_id`.
- `project_id` (optional).
- `channel_id` (optional).
- `agent_id` (who generated it, if any).
- `source_type` — `"message" | "doc" | "file" | "wiki" | "rwa" | ...`.
- `source_id` — e.g. message ID, doc ID, file ID.
- `text` — the chunk content.
- `tags` — list of tags (topic, domain, etc.).
- `visibility` — `"public" | "confidential"`.
- `created_at` — timestamp.

Responsibilities:

- `pipeline/normalization.py`:
- For each event type, map event payload → one or more `IngestChunk` objects.
- Handle splitting of long texts into smaller chunks if needed.

## Embedding & Milvus indexing

### 1. Embedding

- Create an embedding component (`pipeline/embedding.py`) that:
- Accepts `IngestChunk` objects.
- Supports batch processing.

- Uses either:

  - Existing LLM proxy/embedding service (preferred), or
  - Direct model (e.g. local `bge-m3`, `gte-large`, etc.).

- Each chunk after embedding should have vector + metadata per schema in `rag_gateway_task`.

### 2. Milvus indexing

- `pipeline/index_milvus.py` should:
- Upsert chunks into Milvus.
- Ensure idempotency using `chunk_id` as primary key.

- Store metadata:

  - `team_id`, `project_id`, `channel_id`, `agent_id`,
  - `source_type`, `source_id`,
  - `visibility`, `tags`, `created_at`,
  - `embed_model` version.

- Consider using one Milvus collection with a partition key (`team_id`), or per-DAO collections — but keep code flexible.

## Neo4j graph updates

`pipeline/index_neo4j.py` should:

- For events that carry structural information (e.g. project uses resource, doc mentions topic):
- Create or update nodes: `User`, `MicroDAO`, `Project`, `Channel`, `Topic`, `Resource`, `File`, `RWAObject`, `Doc`.

- Create relationships such as:

  - `(:User)-[:MEMBER_OF]->(:MicroDAO)`
  - `(:Agent)-[:SERVES]->(:MicroDAO|:Project)`
  - `(:Doc)-[:MENTIONS]->(:Topic)`
  - `(:Project)-[:USES]->(:Resource)`

- All nodes/edges must include:

- `team_id` / `dao_id`

- `visibility` when it matters

- Operations should be upserts (MERGE) to avoid duplicates.

## Idempotency & reindex

### 1. Idempotent semantics

- Use deterministic `chunk_id` for Milvus records.
- Use Neo4j `MERGE` for nodes/edges based on natural keys (e.g. `(team_id, source_type, source_id, chunk_index)`).
- Replaying the same events should not corrupt or duplicate data.

### 2. Reindex API

- Provide a simple HTTP or CLI interface to:

- `POST /ingest/reindex/{team_id}` — schedule or start reindex for a team/DAO.

- Reindex strategy:

- Read documents/messages from source-of-truth (DB or event replay).

- Rebuild chunks and embeddings.
- Upsert into Milvus & Neo4j (idempotently).

Implementation details (can be left as TODOs if missing backends):

- If there is no easy historic source yet, stub the reindex endpoint with clear TODO and logging.

## Monitoring & logging

Add basic observability:

- Structured logs for:
- Each event type ingested.
- Number of chunks produced.
- Latency for embedding and indexing.
- (Optional) Metrics counters/gauges:
- `ingest_events_total`
- `ingest_chunks_total`
- `ingest_errors_total`

## Files to create/modify (suggested)

> Adjust exact paths if needed.

- `services/rag-ingest-worker/main.py`

- Parse config, connect to event bus, start consumers.

- `services/rag-ingest-worker/config.py`

- Environment variables: `EVENT_BUS_URL`, `MILVUS_URL`, `NEO4J_URL`, `EMBEDDING_SERVICE_URL`, etc.

- `services/rag-ingest-worker/events/consumer.py`

- NATS (or chosen bus) subscription logic.

- `services/rag-ingest-worker/pipeline/normalization.py`

- Functions `normalize_message_created(event)`, `normalize_doc_upsert(event)`, `normalize_file_uploaded(event)`.

- `services/rag-ingest-worker/pipeline/embedding.py`

- `embed_chunks(chunks: List[IngestChunk]) -> List[VectorChunk]`.

- `services/rag-ingest-worker/pipeline/index_milvus.py`

- `upsert_chunks_to_milvus(chunks: List[VectorChunk])`.

- `services/rag-ingest-worker/pipeline/index_neo4j.py`

- `update_graph_for_event(event, chunks: List[IngestChunk])`.

- Optional: `services/rag-ingest-worker/api.py`

- FastAPI app with:

  - `GET /health`
  - `POST /ingest/one`
  - `POST /ingest/reindex/{team_id}`

- Integration docs:

- Reference `docs/cursor/rag_gateway_task.md` and `docs/cursor/42_nats_event_streams_and_event_catalog.md` where appropriate.

## Acceptance criteria

1. A new `rag-ingest-worker` (or similarly named) module/service exists under `services/` with:
2. Clear directory structure (`events/`, `pipeline/`, `config.py`, `main.py`).

3. Stubs or initial implementations for consuming events and indexing to Milvus/Neo4j.

4. A normalized internal model (`IngestChunk` or equivalent) is defined and used across pipelines.

5. Milvus indexing code:

6. Uses idempotent upserts keyed by `chunk_id`.

7. Stores metadata compatible with the RAG-gateway schema.

8. Neo4j update code:

9. Uses MERGE for nodes/relationships.

10. Encodes `team_id`/`dao_id` and privacy where relevant.

11. Idempotency strategy and `reindex(team_id)` path are present in code (even if reindex is initially a stub with TODO).

12. Basic logging is present for ingestion operations.

13. This file (`docs/cursor/rag_ingestion_worker_task.md`) can be executed by Cursor as:

`bash cursor task < docs/cursor/rag_ingestion_worker_task.md`

and Cursor will use it as the single source of truth for implementing/refining the ingestion worker.