`milvus://`¶

Vector database integration backed by the Milvus v2 REST API. The producer issues upsert, search, delete, and query against a collection. No external Go SDK is needed — the component speaks the HTTP REST surface directly. Reach for milvus:// when storing embeddings in a deployed Milvus instance and you want operation-per-URI granularity rather than gluing to a heavier client.

URI shape¶

milvus://<collection>/<operation>?<params>

Producer-only — CreateConsumer returns an error.

Token	Description
`<collection>`	Required — Milvus collection name.
`<operation>`	Optional — `upsert` (default), `search`, `delete`, `query`.

Parameters¶

Param	Type	Default	Producer/Consumer/Both	Description
`address`	string	`localhost:19530`	Producer	gRPC address of Milvus (used in log messages — does not change the REST URL).
`restURL`	string	`http://<address>`	Producer	Base URL of the Milvus REST endpoint. Trailing slash is stripped.
`token`	string	—	Producer	Bearer token sent as `Authorization: Bearer <token>`.
`dimension`	integer	`1536`	Producer	Embedding dimension (informational — Milvus validates the vector itself; not sent on the wire).
`topK`	integer	`5`	Producer	`limit` for `search`.

Operations¶

`upsert`¶

Body must be []map[string]any (or a JSON array). Each element is a Milvus row ({"id": ..., "embedding": [...], "metadata": {...}} — fields match the collection schema). Request: POST /v2/vectordb/entities/upsert.

`search`¶

Body must be []float64 (or a JSON array). The component wraps the vector into a single-row search request:

{"collectionName":"<col>","data":[[<vector>]],"annsField":"embedding","limit":<topK>,"outputFields":["*"]}

The response's data array is set as the new body and milvus.results.count records the count.

`delete`¶

Two modes:

By IDs — set milvus.ids header (comma-separated). The component builds filter: "id in [...]".
By filter — set the body to a filter expression string. Sent verbatim as filter.

`query`¶

Body must be a filter expression string (e.g. metadata['source'] == "docs"). Sent as filter, returns the response's data array as the new body.

Headers¶

Read by the producer¶

Header	Description
`milvus.ids`	`delete` operation — comma-separated IDs.

Written by the producer¶

Header	Description
`milvus.collection`	Collection name.
`milvus.operation`	Operation just executed.
`milvus.results.count`	`search` only — number of results returned.

Examples¶

End-to-end RAG embed + upsert + search:

workflows:
  - id: ingest
    trigger: { uri: direct://ingest }
    steps:
      - rag-chunk: { strategy: recursive }
      - rag-embed: { provider: openai, model: text-embedding-3-small }
      - transform:
          starlang: |
            %sl 2.0
            output application/json
            input payload
            ---
            # transform []*rag.Embedding into Milvus rows
            embeddings_to_milvus_rows(payload)
      - target: milvus://docs/upsert?address=milvus.internal:19530&token=${MILVUS_TOKEN}

  - id: search
    trigger: { uri: direct://search }
    steps:
      - rag-embed: { provider: openai, model: text-embedding-3-small }
      - target: milvus://docs/search?topK=10&token=${MILVUS_TOKEN}

Delete by filter:

- transform:
    starlang: '"metadata[''source''] == \"stale.pdf\""'
- target: milvus://docs/delete

Delete by IDs:

- set-header:
    name: milvus.ids
    starlang: 'join(", ", header("expired-ids"))'
- target: milvus://docs/delete

Behavior notes¶

HTTP REST, not gRPC — address is logged but the REST URL is what's actually used. Override with restURL=https://milvus.internal:443 if Milvus is fronted by HTTPS.
Response cap — the producer reads at most 100 MB per response (io.LimitReader). For larger result sets, paginate via filters / IDs.
Output fields — search and query request outputFields: ["*"]. To restrict, change the operation or supply your own request shape via a transform step.
Authentication — Milvus Cloud uses Authorization: Bearer <token>. Self-hosted Milvus typically needs no token.

Source¶

pkg/component/milvus/milvus.go

milvus://¶