RAG: Embed and Retrieve Documents

Two jobs. The ingest job fetches documents over HTTP, splits each into paragraph chunks that keep the URL they came from, embeds every chunk, and writes the vectors to Qdrant. The search job embeds each question with the same model and returns the closest chunks, each traced back to its source.

Ingest components: generate (one pass) → http (document index) → mapping → unarchive → mapping → http (each document) → mapping → mapping (paragraph chunks) → unarchive → branch with ollama_embeddings → mapping → http_client output to the Qdrant REST API.

Search components: file input → branch with ollama_embeddings → branch with the qdrant processor → mapping → file output.

Needs:

An Ollama server with the nomic-embed-text model, reachable from the node. The runs used Ollama 0.33.2 on the node's host.
A Qdrant server, reachable from the node on its REST port (6333) for writes and its gRPC port (6334) for search. The runs used Qdrant 1.19.1.
A collection whose vector size matches the model: 768, cosine distance.
A document index: a URL returning {"documents": ["https://…/a.md", "https://…/b.md"]}.

Where data goes: the node fetches the index and each document from their host. Chunk text goes to the Ollama server and vectors plus chunk text go to Qdrant. With both on the node's own host, as in the runs, nothing else leaves it.

Write vectors with http_client, not the qdrant output

On expanso-edge v2.1.21 the native qdrant output could not read message fields in its id mapping. Every write failed, and the job still reported completed with 0 points stored. The ingest job below builds the Qdrant upsert body in a mapping and sends it with http_client, which worked. The qdrant processor, used for search, works.

Set up the dependencies

On the host that will serve the model and the vector store:

ollama pull nomic-embed-text
docker run -d --name qdrant -p 6333:6333 -p 6334:6334 \
  qdrant/qdrant
curl -s -X PUT http://127.0.0.1:6333/collections/docs \
  -H 'Content-Type: application/json' \
  -d '{"vectors":{"size":768,"distance":"Cosine"}}'

The last command should print "result":true.

Ingest job

Each document is split on blank lines, and any paragraph that starts with # is dropped whole, including body text directly under a heading with no blank line between them. The first line becomes the title, and every chunk carries source_uri, doc, title, chunk_index, chunk_id and text. chunk_id is the full source URL plus the chunk index, and the point id is derived from it. Documents with the same file name at different URLs get separate points, and ingesting the same document again overwrites its points instead of adding duplicates.

rag-ingest.yaml
name: rag-ingest
type: pipeline
# One-shot job: stop after one failed execution instead of retrying.
restart_policy: never
selector:
  match_labels:
    pipeline_role: rag
config:
  input:
    generate:
      # One fetch pass. For continuous ingestion set an interval and
      # drop count, and remove `restart_policy: never` above: one
      # transient fetch error would stop a long-running job for good.
      # The default policy restarts it on failure.
      count: 1
      interval: ""
      mapping: root = {}
  pipeline:
    processors:
      # 1. Fetch the document index:
      #    {"documents": ["https://.../a.md", ...]}
      - http:
          url: "https://DOCS_HOST/docs/index.json"
          verb: GET
          retries: 3
      - mapping: |
          root = if errored() {
            throw("index fetch failed: " + error().or("unknown"))
          } else { this.documents }
      - unarchive:
          format: json_array
      # 2. Fetch each document, remembering where it came from.
      - mapping: |
          meta source_uri = this
          root = ""
      - http:
          url: '${! meta("source_uri") }'
          verb: GET
          retries: 3
      - mapping: |
          root = if errored() {
            throw("document fetch failed: " + error().or("unknown"))
          } else { content() }
      # 3. Chunk by paragraph; every chunk carries its source URI.
      - mapping: |
          let src = @source_uri
          let doc = $src.filepath_split().index(1)
          let body = content().string()
          let title = $body.split("\n").index(0).trim_prefix("# ")
          let paras = $body.split("\n\n").map_each(
            p -> p.trim()
          ).filter(p -> p != "" && !p.has_prefix("#"))
          root = $paras.enumerated().map_each(e -> {
            "source_uri": $src,
            "doc": $doc,
            "title": $title,
            "chunk_index": e.index,
            "chunk_id": $src + "#" + e.index.string(),
            "text": e.value
          })
      - unarchive:
          format: json_array
      - branch:
          # nomic-embed-text expects a task prefix on the input text.
          request_map: 'root = "search_document: " + this.text'
          processors:
            - ollama_embeddings:
                model: "nomic-embed-text"
                server_address: "http://127.0.0.1:11434"
          result_map: root.vector = this
      # Deterministic UUID from chunk_id: re-ingesting overwrites,
      # never duplicates.
      - mapping: |
          let h = this.chunk_id.hash("sha256").encode("hex")
          root.points = [{
            "id": "%s-%s-%s-%s-%s".format(
              $h.slice(0, 8), $h.slice(8, 12), $h.slice(12, 16),
              $h.slice(16, 20), $h.slice(20, 32)
            ),
            "vector": this.vector,
            "payload": this.without("vector")
          }]
  output:
    http_client:
      url: "http://127.0.0.1:6333/collections/docs/points?wait=true"
      verb: PUT
      headers:
        Content-Type: application/json
      retries: 3
      retry_period: 500ms
      max_in_flight: 4

The selector sends the job to a node labelled pipeline_role: rag; set that label on your node, or change the selector to a label it already carries. With no matching node, the job is stored and never runs.

Replace https://DOCS_HOST/docs/index.json with your document index. If Ollama or Qdrant is not on the node's own host, change server_address and the http_client URL.

A fetch that fails, for the index or any document, throws, so the job fails instead of completing with nothing stored.

Search job

The questions file is on the node that runs the job, one JSON object per line with a question field. The job also copies an optional expected_doc field through, which is useful for testing.

rag-query.yaml
name: rag-query
type: pipeline
# One-shot job: stop after one failed execution instead of retrying.
restart_policy: never
selector:
  match_labels:
    pipeline_role: rag
config:
  input:
    file:
      paths: ["/var/tmp/expanso-rag/questions.jsonl"]
      scanner:
        lines: {}
  pipeline:
    processors:
      - branch:
          request_map: 'root = "search_query: " + this.question'
          processors:
            - ollama_embeddings:
                model: "nomic-embed-text"
                server_address: "http://127.0.0.1:11434"
          result_map: root.vector = this
      - branch:
          processors:
            - qdrant:
                grpc_host: "127.0.0.1:6334"
                collection_name: "docs"
                vector_mapping: root = this.vector
                # payload_fields defaults to [] (no payload); ask
                # for the source-tracing fields explicitly.
                payload_fields: [source_uri, doc, chunk_id,
                  chunk_index, text]
                payload_filter: include
                limit: 3
          result_map: root.hits = this
      # The qdrant processor returns proto3-JSON hits
      # ({"stringValue": ...}); flatten them into plain fields for
      # whatever consumes the answers.
      - mapping: |
          root.question = this.question
          root.expected_doc = this.expected_doc
          root.hits = this.hits.map_each(h -> {
            "id": h.id.uuid,
            "score": h.score,
            "source_uri": h.payload.source_uri.stringValue.or(null),
            "doc": h.payload.doc.stringValue,
            "chunk_id": h.payload.chunk_id.stringValue,
            "chunk_index":
              h.payload.chunk_index.integerValue.number(),
            "text": h.payload.text.stringValue
          })
  output:
    file:
      path: "/var/tmp/expanso-rag/out/answers.jsonl"
      codec: lines

The qdrant processor returns no payload unless payload_fields lists the fields, and returns hits in a typed form such as {"stringValue": "..."}. The last mapping flattens them.

Run and check it

Deploy the ingest job, then count the stored points:

expanso-cli job deploy rag-ingest.yaml
curl -s http://127.0.0.1:6333/collections/docs \
  | grep -o '"points_count":[0-9]*'

The count equals the number of chunks across your documents. Deploying the ingest job again leaves it unchanged. Then, on the node, write a question and run the search:

mkdir -p /var/tmp/expanso-rag/out
printf '%s\n' '{"question":"When should the battery be swapped?"}' \
  > /var/tmp/expanso-rag/questions.jsonl
expanso-cli job deploy rag-query.yaml
cat /var/tmp/expanso-rag/out/answers.jsonl

Each line of answers.jsonl holds the question and its top 3 hits, each with score, source_uri, doc, chunk_id, chunk_index and text.

What the run proved

The runs used four synthetic Markdown documents served over HTTP, such as a field-sensor battery guide and a data-center cooling runbook.

Each ingest run fetched the index once and each of the 4 documents once.
The 4 documents became 16 chunks. Every stored payload matched its source URL, document, chunk index and text.
Every vector had 768 dimensions. A stored vector and a direct Ollama embedding of the same text had cosine similarity 1.000000.
A second ingest run left the count at 16.
For each of 4 questions, the top hit came from the expected document:

Question	Top hit	Score
How often do the hoist wire ropes on the port cranes need a visual check?	`harbor-crane-maintenance.md#1`	0.817
When do we switch on sprinklers to keep the apple buds from freezing?	`orchard-irrigation-policy.md#2`	0.752
At what reading should the sensor battery be swapped out?	`field-sensor-battery-guide.md#1`	0.806
What inlet temperature sets off the critical alarm for the server room?	`data-center-cooling-runbook.md#1`	0.823

Both jobs were deployed through Expanso Cloud to a labelled node running expanso-edge v2.1.21. Ollama, Qdrant and the document server ran on the same host as the node. Those runs built chunk_id from the file name, which is why the table shows ids such as harbor-crane-maintenance.md#1, and did not set restart_policy.

Two corrections were added later and proved on a local-mode node only, with the same Ollama model and Qdrant version:

Source-URL chunk ids. With two documents both named guide.md at different URLs plus a third document, the file-name id stored 3 points because the two guides overwrote each other. The source-URL id stored all 5 chunks, a second ingest left the count at 5, and a question about each guide returned a chunk from the right URL first.
restart_policy: never. The ingest job with it completed with 5 points. The same job pointed at a bad index failed after exactly one execution.

Both jobs exactly as published, source-URL chunk ids and restart_policy: never included, were then run on a local-mode node against the four synthetic documents. Ingest stored 16 chunks, each chunk_id the source URL plus # plus the chunk index. The query job answered all 4 questions with the top hit from the expected document. Each job completed after 1 execution.

Limits

Only ollama_embeddings with nomic-embed-text and Qdrant were run. The catalog also has aws_bedrock_embeddings, gcp_vertex_ai_embeddings and openai_embeddings, and chat processors such as ollama_chat; none of them was part of these runs.
Chunks are paragraphs. There is no token-aware or overlapping splitter, and a very long paragraph becomes one long chunk.
A paragraph that starts with # is dropped whole, so text written directly under a heading is never embedded and nothing reports it. Put a blank line after each heading, or change the chunk filter (a change that was not run).
The ingest job does one pass. For continuous ingestion, the job's comment says to set an interval and drop count; that was not run. Such a long-running job should not keep restart_policy: never, or a single transient fetch error stops it for good. Choose a policy suited to a long-running job; the default restarts it on failure.
nomic-embed-text expects the search_document: and search_query: prefixes the jobs add. Another model may expect different input.
The published jobs, with source-URL chunk ids and restart_policy: never, were proved on a local-mode node only; the Cloud runs used the earlier jobs. See What the run proved.

Build by Job: the task-to-components matrix
ollama_embeddings processor and qdrant processor

Set up the dependencies​

Ingest job​

Search job​

Run and check it​

What the run proved​

Limits​

Related​