RAG exploration

Directory.Packages.props

@@ -18,6 +18,9 @@
     <PackageVersion Include="Microsoft.Extensions.Diagnostics.HealthChecks" Version="10.0.11" />
     <PackageVersion Include="Nuke.Common" Version="10.1.0" />
     <PackageVersion Include="Polly.Core" Version="8.7.0" />
+    <!-- Pgvector sink package -->
+    <PackageVersion Include="Microsoft.Extensions.AI.Abstractions" Version="10.9.0" />
+    <PackageVersion Include="Pgvector" Version="0.3.2" />
     <!-- Meilisearch sink package -->
     <PackageVersion Include="Meilisearch" Version="0.20.0" />
     <!-- Kafka sink package -->

Wallaby.slnx

@@ -15,6 +15,7 @@
     <Project Path="src/Wallaby.Sinks.Kafka/Wallaby.Sinks.Kafka.csproj" />
     <Project Path="src/Wallaby.Sinks.Meilisearch/Wallaby.Sinks.Meilisearch.csproj" />
     <Project Path="src/Wallaby.Sinks.OpenSearch/Wallaby.Sinks.OpenSearch.csproj" />
+    <Project Path="src/Wallaby.Sinks.Pgvector/Wallaby.Sinks.Pgvector.csproj" />
     <Project Path="src/Wallaby.Testing/Wallaby.Testing.csproj" />
     <Project Path="src/Wallaby/Wallaby.csproj" />
   </Folder>
@@ -23,6 +24,7 @@
     <Project Path="tests/Wallaby.Benchmarks/Wallaby.Benchmarks.csproj" />
     <Project Path="tests/Wallaby.Client.Tests/Wallaby.Client.Tests.csproj" />
     <Project Path="tests/Wallaby.Providers.EntityFrameworkCore.Tests/Wallaby.Providers.EntityFrameworkCore.Tests.csproj" />
+    <Project Path="tests/Wallaby.Sinks.Pgvector.Tests/Wallaby.Sinks.Pgvector.Tests.csproj" />
     <Project Path="tests/Wallaby.Providers.Marten.Tests/Wallaby.Providers.Marten.Tests.csproj" />
     <Project Path="tests/Wallaby.Sinks.Elasticsearch.Tests/Wallaby.Sinks.Elasticsearch.Tests.csproj" />
     <Project Path="tests/Wallaby.Sinks.Http.Tests/Wallaby.Sinks.Http.Tests.csproj" />

docs/.vitepress/config.mts

@@ -51,6 +51,7 @@ export default defineConfig({
           { text: 'Getting Started', link: '/getting-started' },
           { text: 'Mappings', link: '/mappings' },
           { text: 'Backfill', link: '/backfill' },
+          { text: 'RAG & Embeddings', link: '/rag' },
           { text: 'External Slots', link: '/external-slots' },
           { text: 'Configuration', link: '/configuration' },
           { text: 'Testing', link: '/testing' },
@@ -82,6 +83,7 @@ export default defineConfig({
           { text: 'Kafka', link: '/sinks/kafka' },
           { text: 'Elasticsearch', link: '/sinks/elasticsearch' },
           { text: 'OpenSearch', link: '/sinks/opensearch' },
+          { text: 'Pgvector', link: '/sinks/pgvector' },
           { text: 'Custom', link: '/sinks/custom' },
         ]
       },

docs/backfill.md

@@ -21,7 +21,8 @@ sink.Map<Product>()
 ```
 
 Each entity is versioned and backfilled independently, so reindexing one doesn't disturb others or the
-live stream.
+live stream. Version bumps are also how [embedding-model migrations](/rag#model-migrations) re-embed
+a corpus: encode the model in the version string and bump it with `purgeOnChange: true`.
 
 When an entity is mapped to **several sinks**, backfill state is still per table: bumping *any*
 mapping's version re-snapshots the table, and the snapshot flows through every sink mapped to it.

docs/mappings.md

@@ -52,6 +52,9 @@ interface as a class - [`IWallabyEfTransform<T>`](/providers/entity-framework-co
 (EF Core) or [`IWallabyMartenTransform<T>`](/providers/marten/#class-based-transforms) (Marten) - and
 register it with `UsingTransform<TEntity, TTransform>()`. This class is registered & resolved from the container.
 
+Transforms can also enrich documents with vector embeddings for semantic search - see
+[RAG & Embeddings](/rag).
+
 ## Mapping classes
 
 Inline mappings grow the `AddWallaby` callback and can make your `Program.cs` unwieldy. Move each mapping into a

docs/rag.md

@@ -0,0 +1,126 @@
+---
+description: "Keeping embeddings and RAG corpora continuously in sync with Postgres using CDC: destination-side embedding, the pgvector sink, and model migrations."
+---
+
+# RAG & Embeddings
+
+A RAG corpus or semantic search index is only as good as its freshness, and keeping vectors in sync
+with a database is exactly the shape of problem Wallaby already solves: changes stream through your
+transform, deletes propagate, [backfill](/backfill) seeds and re-seeds destinations, and
+`WithBackfillVersion` gives embedding-model migrations a one-line answer.
+
+The guiding principle: **let the destination own embedding**. When the party that stores the vector
+also computes it, no vectors transit the pipeline, there is no cache to build or invalidate, and
+the destination can skip re-embedding text it has already seen. Every path below follows that
+shape; computing vectors inside a transform is the fallback, not the default.
+
+## Search sinks: the destination embeds
+
+Each search destination has a native way to embed the text Wallaby syncs:
+
+- **Meilisearch** - declare a [server-side embedder](/sinks/meilisearch#embedders-vector-search)
+  (`OpenAi`, `HuggingFace`, `Ollama`, or `Rest`) with a `documentTemplate`; the index becomes
+  hybrid-searchable with zero embedding code:
+
+```csharp
+cdc.AddMeilisearchSink("meili", m =>
+{
+    m.Host = "http://localhost:7700";
+    m.ConfigureIndex("products", s =>
+    {
+        s.SearchableAttributes = ["name", "description"];
+        s.Embedders = new Dictionary<string, Embedder>
+        {
+            ["default"] = new Embedder
+            {
+                Source = EmbedderSource.OpenAi,
+                Model = "text-embedding-3-small",
+                ApiKey = openAiKey,
+                DocumentTemplate = "{{doc.name}}: {{doc.description}}",
+            },
+        };
+    });
+});
+```
+
+- **Elasticsearch** - map the field as
+  [`semantic_text`](/sinks/elasticsearch#vector-search) backed by an inference endpoint; the
+  cluster chunks and embeds at index time. The inference API needs an appropriate Elastic
+  subscription, and the default ELSER endpoint needs ML nodes.
+- **OpenSearch** - attach a [neural-search ingest pipeline](/sinks/opensearch#vector-search)
+  (a `text_embedding` processor over a deployed model) to the index.
+
+In all three, Wallaby delivers plain text and every insert, update, delete, and backfill keeps the
+index converged. They differ on re-embedding cost: Meilisearch caches embeddings and calls the
+embedder only for documents whose rendered `documentTemplate` changed, while Elasticsearch and
+OpenSearch run inference on every indexed document - fine for the live change stream (Wallaby only
+delivers rows that changed), but a [backfill](/backfill) re-embeds the whole corpus.
+
+## Postgres as the vector store: the pgvector sink
+
+For "my RAG corpus is just Postgres", the [pgvector sink](/sinks/pgvector) plays the
+destination-embeds role itself, since Postgres has no native embedder:
+
+```csharp
+cdc.AddPgvectorSink("vectors", v =>
+{
+    v.ConnectionString = vectorDbConn;
+    v.Dimensions = 1536;
+    v.EmbeddingGenerator = generator;  // any Microsoft.Extensions.AI IEmbeddingGenerator
+    v.EmbedText = d => $"{d["name"]}\n{d["description"]}";
+    v.EmbeddingVersion = "text-embedding-3-small/1";
+})
+.WithMappings(sink => sink
+    .Map<Product>()
+    .ToDestination("products")
+    .WithBackfillVersion("v1", purgeOnChange: true)
+    .UsingTransform(/* emit name + description as plain text */));
+```
+
+The sink embeds at delivery time and stores a content hash next to each vector, so it
+[re-embeds only rows whose text changed](/sinks/pgvector#how-embedding-is-gated) - across restarts,
+failovers, and re-backfills, with the destination table itself as the durable cache. Embedding-API
+throttling surfaces as a retryable delivery, riding the dispatcher's normal backoff.
+
+## Hand-rolled: embedding in a transform
+
+For destinations that can't embed and can't be read back - Kafka topics, HTTP receivers, or a
+Meilisearch `UserProvided` embedder - compute the vector in the transform and emit it as a document
+field. A `float[]` or `ReadOnlyMemory<float>` value is written as a plain JSON number array by the
+Elasticsearch, OpenSearch, HTTP, and Kafka sinks.
+
+Two rules make this safe and affordable:
+
+- **Batch and retry inside the transform.** Transforms receive whole batches (deduplicated to one
+  change per row) - make one provider call per batch, never per row. And a transform exception
+  **halts the pipeline** (retry classification exists only at sink delivery), so wrap the embedding
+  call in your own retry (e.g. a Polly `ResiliencePipeline` with exponential backoff and jitter)
+  and throw only when retries are exhausted; at that point halting is correct backpressure.
+- **Skip unchanged text.** On updates, `ChangeEvent.Changes` holds the previous values of changed
+  columns - when none of the embedded columns appear in it, skip the API call. Combine with the
+  mapping's [column selection](/providers/entity-framework-core/#declaring-consumed-columns) so
+  updates to irrelevant columns never reach the transform at all. Note `Changes` is `null` on
+  inserts and backfill rows, so a re-backfill re-embeds everything on this path - one more reason
+  to prefer the destination-side options above.
+
+## Model migrations
+
+Changing the embedding model (or the prompt template baked into the text) makes every stored vector
+stale. Encode the model in the [backfill version](/backfill#automatic-backfill) and bump it:
+
+```csharp
+.WithBackfillVersion("text-embedding-3-large/1", purgeOnChange: true)
+```
+
+The bump triggers a full re-backfill of the entity, and `purgeOnChange: true`
+[purges the destination first](/backfill#purging-before-a-backfill) so no old-model vectors survive
+alongside new ones. For the pgvector sink, change `EmbeddingVersion` in the same deploy (it feeds
+the stored hash); for destination-side embedders, update the embedder/endpoint configuration.
+Dimension changes (e.g. 1536 → 3072) also need the index or column recreated - purge handles the
+documents, not the schema.
+
+## Delete propagation
+
+Nothing extra to do: a source row's delete removes its document (and vector) from the destination,
+and a transform returning `null` for a key does the same. Stale-vector cleanup is the pipeline's
+normal delete path, not a special case.

docs/sinks/elasticsearch.md

@@ -59,6 +59,33 @@ default). For explicit settings or mappings (analyzers, `dense_vector` fields, s
 create the index up front — via Kibana Dev Tools, your infrastructure tooling, or a deployment
 script. In-sink index bootstrapping is planned.
 
+## Vector search
+
+A transform can emit an embedding as a `float[]` or `ReadOnlyMemory<float>` field - the sink writes
+either as a plain JSON number array, with no `SerializerOptions` needed. Dynamic mapping would infer
+an ordinary `float` field, so pre-create the index with an explicit `dense_vector` mapping:
+
+```json
+PUT /products
+{
+  "mappings": {
+    "properties": {
+      "embedding": { "type": "dense_vector", "dims": 1536, "index": true, "similarity": "cosine" }
+    }
+  }
+}
+```
+
+Don't pass a quantized vector as `byte[]` - byte arrays serialize as base64 strings, not arrays.
+
+Elasticsearch can also embed for you: map the field as
+[`semantic_text`](https://www.elastic.co/docs/solutions/search/semantic-search/semantic-search-semantic-text)
+(backed by an inference endpoint) and sync plain text - the cluster chunks and embeds at index
+time, with no vectors in your pipeline at all. Weigh the prerequisites and cost profile first: the
+inference API needs an appropriate Elastic subscription (and the default ELSER endpoint needs ML
+nodes), and inference runs on every indexed document - live changes embed incrementally, but a
+[backfill](/backfill) re-runs inference over the whole corpus. See [RAG & Embeddings](/rag).
+
 ## Authentication
 
 `ApiKey` or `Username`/`Password` cover the common schemes. Everything else — Elastic Cloud ids,

docs/sinks/http.md

@@ -239,7 +239,7 @@ against the body your endpoint reads after the middleware has decompressed it.
 ## NativeAOT
 
 The envelope structure and common scalar document values (strings, numbers, booleans, `Guid`, date/time
-types, byte arrays, nested dictionaries, and sequences of these) are written without reflection. Any other
+types, byte arrays, nested dictionaries, `ReadOnlyMemory<float>` vectors, and sequences of these) are written without reflection. Any other
 value type is serialized through `SerializerOptions`; on trimmed/NativeAOT hosts, point it at a
 source-generated context covering the types your transforms emit:
 

docs/sinks/kafka.md

@@ -152,7 +152,7 @@ Leave `Topics` empty when your platform pre-provisions topics or the broker has
 ## NativeAOT
 
 The envelope structure and common scalar document values (strings, numbers, booleans, `Guid`, date/time
-types, byte arrays, nested dictionaries, and sequences of these) are written without reflection. Any other
+types, byte arrays, nested dictionaries, `ReadOnlyMemory<float>` vectors, and sequences of these) are written without reflection. Any other
 value type is serialized through `SerializerOptions`; on trimmed/NativeAOT hosts, point it at a
 source-generated context covering the types your transforms emit:
 

docs/sinks/meilisearch.md

@@ -85,6 +85,44 @@ cdc.AddMeilisearchSink("meili", m =>
 `Settings` is Meilisearch's own settings type, so you have full control (ranking rules, stop words,
 synonyms, faceting, …). Setup is idempotent and re-applied on each leadership acquisition.
 
+### Embedders (vector search)
+
+`Settings.Embedders` rides the same startup settings application, so
+[AI-powered search](https://www.meilisearch.com/docs/learn/ai_powered_search/getting_started_with_ai_search)
+is declared the same way:
+
+```csharp
+m.ConfigureIndex("products", s =>
+{
+    s.SearchableAttributes = ["name", "description"];
+    s.Embedders = new Dictionary<string, Embedder>
+    {
+        ["default"] = new Embedder
+        {
+            Source = EmbedderSource.OpenAi,
+            Model = "text-embedding-3-small",
+            ApiKey = openAiKey,
+            DocumentTemplate = "{{doc.name}}: {{doc.description}}",
+        },
+    };
+});
+```
+
+With a server-side source (`OpenAi`, `HuggingFace`, `Ollama`, `Rest`), Meilisearch computes vectors
+itself from the synced documents: Wallaby delivers plain text and the index becomes semantically
+searchable with zero embedding code in your pipeline. With `EmbedderSource.UserProvided`, the
+transform carries the vector in the document's `_vectors` field instead:
+
+```csharp
+new WallabyDocument
+{
+    ["name"] = p.Name,
+    ["_vectors"] = new Dictionary<string, object?> { ["default"] = embedding }, // float[]
+};
+```
+
+See [RAG & Embeddings](/rag) for the full pattern, including re-embedding on model changes.
+
 ### Attribute validation
 
 By default (`ValidateConfiguredAttributes = true`), every upsert routed to a `ConfigureIndex`-declared index

docs/sinks/opensearch.md

@@ -59,6 +59,33 @@ default). For explicit settings or mappings (analyzers, `knn_vector` fields, sha
 create the index up front — via Dev Tools, your infrastructure tooling, or a deployment script.
 In-sink index bootstrapping is planned.
 
+## Vector search
+
+A transform can emit an embedding as a `float[]` or `ReadOnlyMemory<float>` field - the sink writes
+either as a plain JSON number array, with no `SerializerOptions` needed. Dynamic mapping would infer
+an ordinary `float` field, so pre-create the index with an explicit `knn_vector` mapping:
+
+```json
+PUT /products
+{
+  "settings": { "index.knn": true },
+  "mappings": {
+    "properties": {
+      "embedding": { "type": "knn_vector", "dimension": 1536 }
+    }
+  }
+}
+```
+
+Don't pass a quantized vector as `byte[]` - byte arrays serialize as base64 strings, not arrays.
+
+OpenSearch can also embed for you: attach a
+[neural-search ingest pipeline](https://docs.opensearch.org/latest/vector-search/ai-search/semantic-search/)
+(a `text_embedding` processor over a deployed model) to the index and sync plain text - the cluster
+computes vectors at index time, with none in your pipeline at all. Note the pipeline runs on every
+indexed document, so live changes embed incrementally but a [backfill](/backfill) re-runs the model
+over the whole corpus. See [RAG & Embeddings](/rag).
+
 ## Authentication
 
 `Username`/`Password` cover basic auth. Everything else — AWS SigV4, client certificates,