Skip to content
This repository was archived by the owner on Oct 14, 2026. It is now read-only.

Commit 2d4ef8f

Browse files
author
Tanase Butcaru
committed
feat(data-access): semantic type registries, type-list lookup and shared embedding config (LLMO-7445)
1 parent 97e16ad commit 2d4ef8f

5 files changed

Lines changed: 157 additions & 21 deletions

File tree

‎docs/adr/0002-embedding-client-and-semantic-index-utils.md‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -40,23 +40,29 @@ Following ADR-0001's test (*true for every consumer → shared; specific to one
4040

4141
**Consumers own:**
4242
- Caching. The durable query-embedding cache (`semantic_query_embedding`) and its key policy (`hash(normalizedText)+model+dims`) live in the data-access helpers/consumer, not the client — staleness/keying is a consumer concern.
43-
- Model/dimension policy at the call site (the cache key and the stored `vector(N)` column pin the model+dims contract).
43+
- Model/dimension values come from `SEMANTIC_MATCHING_CONFIG` (`embeddingModel`, `embeddingDims`) in `data-access`, which owns the stored index; consumers import it and pass it as `model` / `dims`. It is a code constant, not env config: the label is persisted on every row and in the cache key, so writer and reader must never disagree, and changing it is a re-embed. `embeddingDims` must match the `vector(N)` columns. The helpers still accept any `model` / `dims`, so a same-dims model change can run the online re-embed below.
4444
- Translating a client failure into an HTTP response (api-service returns 503; the audit-worker write path degrades best-effort).
4545

4646
### 4. Semantic-index helpers live in `data-access`, mirroring the URL index
4747

4848
`semantic-index.utils.js` mirrors `url-index.utils.js`: `syncOpportunitySemantic` (full-replace per `source_type`), `lookupOpportunitiesByVectors` (ANN read RPC wrapper, many query vectors per call), the batched `semantic_query_embedding` cache helpers (`getQueryEmbeddings` / `upsertQueryEmbeddings` / `touchQueryEmbeddings`), and `copyEntityVectors` (a thin wrapper over the `wrpc_copy_opportunity_semantic_vectors` **server-side** write RPC — the copy runs as `INSERT … SELECT`, no rows materialized in the caller). Text normalization + hashing live in this layer so a value written always matches a later read/re-sync. **The shared layer stores and reads pre-embedded vectors; it never embeds** — embedding is the writer's (audit-worker's) job via the client above.
4949

50+
**Type registries.** `OPPORTUNITY_SEMANTIC_SOURCE_TYPES` (kinds of embedded text: `topic`, later `claim`) and `OPPORTUNITY_SEMANTIC_ENTITY_TYPES` (opportunity types, from `OPPORTUNITY_TYPES`) are exported from this layer and are the only values the writer stores and the reader accepts; anything else is a `ValidationError`. The reader takes `sourceTypes` (required, non-empty) and optional `entityTypes` (omitted or empty = all), deduped, so a list can never exceed its registry. The DB does not duplicate the registries: the RPC only rejects an empty source-type list and NULL elements. A team that wants its opportunities searchable adds its source kind and/or opportunity type to the registries, indexes with them, and callers opt in by passing them.
51+
5052
## Consequences
5153

5254
- `gpt-client` gains an embeddings surface and its own `AZURE_EMBEDDING_*` config (documented in its README); the intended cost, not a drawback.
5355
- The shared embedding model is a cross-repo contract: both the write and read paths must point at the **same deployment**. ANN search is **scoped to one generation** — `lookupOpportunitiesByVectors` (and the `rpc_opportunity_semantic_search` RPC) filter on `model`+`dims`, and the query cache is keyed on `model+dims`. This makes a **same-dims model change an online, forward-only re-embed** (chosen over an atomic swap — see Alternatives): the read path queries the new generation immediately while the write path back-populates it opportunity-by-opportunity on the normal refresh cadence. **Accepted cost:** during the migration window, opportunities not yet re-embedded don't match new-generation queries (reduced recall), self-healing as the backfill completes. A **dimension change** is out of this online path — it needs a new `vector(N)` column and an atomic swap, not a forward-only re-embed.
54-
- **Deploy ordering:** `lookupOpportunitiesByVectors` calls `rpc_opportunity_semantic_search` with `p_query_embeddings` (a list of vectors) plus `p_model`/`p_dims`, so the data-service release carrying that signature must be deployed **before** any consumer upgrades to this library version — otherwise PostgREST can't resolve the function (`PGRST202`).
56+
- **Deploy ordering:** `lookupOpportunitiesByVectors` calls `rpc_opportunity_semantic_search` with `p_source_types`/`p_entity_types` (lists) and `p_query_embeddings` (a list of vectors) plus `p_model`/`p_dims`, so the data-service release carrying that signature must be deployed **before** any consumer upgrades to this library version — otherwise PostgREST can't resolve the function (`PGRST202`).
5557
- Consumers stay thin: they inject the `EmbeddingProvider`, own their cache/status mapping, and call the `data-access` helpers — no matching logic is reimplemented per consumer.
5658

5759
## Alternatives considered
5860

5961
- **A new `spacecat-shared-embedding-client` package.** Rejected for now: same vendor/auth as `gpt-client`; deferred behind the `EmbeddingProvider` seam so a later extraction is non-breaking.
6062
- **Overloading `AzureOpenAIClient`.** Rejected: embeddings are a different API surface; a distinct class keeps each client cohesive.
6163
- **Embedding inside the shared data-access layer.** Rejected: transport/auth belongs in a client (ADR-0001 boundary), and the write side (audit-worker) owns when/what to embed.
64+
- **Producer-namespaced source types (e.g. `offsite_topic`).** Rejected: `entity_type` already identifies the producer, so namespacing `source_type` would duplicate it. Generic kinds plus an optional `entityTypes` filter keep the two axes separate. Accepted cost: a new producer that indexes `topic` shows up in by-topics results by default unless callers pass `entityTypes`.
65+
- **A `*` (all types) wildcard.** Rejected: results would change silently whenever a new producer starts indexing, and an exact scan's cost grows with every type included. Callers that want every kind pass `Object.values(OPPORTUNITY_SEMANTIC_SOURCE_TYPES)` explicitly.
66+
- **Registries enforced in the DB (CHECK/enum).** Rejected: a migration for every new type; the registries live next to the helpers that enforce them.
67+
- **Embedding model label from env (e.g. derived from `AZURE_EMBEDDING_DEPLOYMENT`).** Rejected: writer and reader have separate env configs, so a mismatch would silently return no matches; a deployment name is an alias, not a model id; and changing the label is a re-embed, which should be a reviewed release.
6268
- **Atomic-swap migration (full index rebuild + coordinated cutover) instead of the online, generation-scoped re-embed.** Rejected: it would require rebuilding the whole opportunity index and flipping the read path over in lockstep across four repos — high operational risk and coordination cost for a rare event. The chosen online path (ANN search filtered by `model`+`dims`) lets the read path move to the new generation immediately while the write path back-populates on its normal cadence; the accepted cost is reduced recall for not-yet-re-embedded opportunities during the migration window (see Consequences). Atomic swap remains the required approach for a **dimension** change, which the online path deliberately does not cover.

‎packages/spacecat-shared-data-access/README.md‎

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -148,14 +148,19 @@ For the topic vector index (`opportunity_semantic_embedding`) and the global que
148148
import {
149149
syncOpportunitySemantic, copyEntityVectors, lookupOpportunitiesByVectors,
150150
getQueryEmbeddings, upsertQueryEmbeddings, touchQueryEmbeddings,
151+
OPPORTUNITY_SEMANTIC_SOURCE_TYPES, OPPORTUNITY_SEMANTIC_ENTITY_TYPES, SEMANTIC_MATCHING_CONFIG,
151152
} from '@adobe/spacecat-shared-data-access';
152153
153154
const { postgrestClient } = dataAccess.services;
154-
const scope = { model: 'azure/text-embedding-3-small', dims: 1536 };
155+
const scope = {
156+
model: SEMANTIC_MATCHING_CONFIG.embeddingModel, dims: SEMANTIC_MATCHING_CONFIG.embeddingDims,
157+
};
158+
const { TOPIC } = OPPORTUNITY_SEMANTIC_SOURCE_TYPES;
159+
const { CITED_ANALYSIS } = OPPORTUNITY_SEMANTIC_ENTITY_TYPES;
155160
156161
// writer (needs the postgrest_writer role): full-replace an opportunity's topic vectors
157162
await syncOpportunitySemantic(postgrestClient, {
158-
siteId, entityId, entityType, sourceType: 'topic', sources: [{ text, vector, ...scope }],
163+
siteId, entityId, entityType: CITED_ANALYSIS, sourceType: TOPIC, sources: [{ text, vector, ...scope }],
159164
});
160165

161166
// writer: copy an opportunity's vector rows to another opportunity in the same site
@@ -166,12 +171,17 @@ const cached = await getQueryEmbeddings(postgrestClient, { texts, ...scope });
166171
await upsertQueryEmbeddings(postgrestClient, { entries: [{ text, vector }], ...scope });
167172
await touchQueryEmbeddings(postgrestClient, { textHashes, ...scope });
168173

169-
// reader, batched: one best-first list of { entityId, entityType, score } per query vector
174+
// reader, batched: one best-first list of { entityId, entityType, score } per query vector.
175+
// entityTypes is optional (omitted or [] searches all entity types).
170176
const matches = await lookupOpportunitiesByVectors(postgrestClient, {
171-
siteId, sourceType: 'topic', vectors, k: 10, minScore: 0.5, ...scope,
177+
siteId, sourceTypes: [TOPIC], entityTypes: [CITED_ANALYSIS], vectors, k: 10, minScore: 0.5, ...scope,
172178
});
173179
```
174180
181+
`OPPORTUNITY_SEMANTIC_SOURCE_TYPES` (kinds of embedded text) and `OPPORTUNITY_SEMANTIC_ENTITY_TYPES` (opportunity types) are the registries for `source_type` and `entity_type`: the writer and reader reject any value outside them. To index a new kind or opportunity type, add it to the registry first.
182+
183+
`SEMANTIC_MATCHING_CONFIG` (`embeddingModel`, `embeddingDims`) is the embedding generation the writer and reader both use; pass it as `model` / `dims` rather than hardcoding the values. It is a code constant, not env config, so the two sides can never disagree.
184+
175185
The batched helpers group their calls internally (search in groups of up to 20 vectors, cache reads/touches in groups of 50 hashes), so callers pass whole lists rather than looping per item.
176186
177187
## Field Mapping Behavior

‎packages/spacecat-shared-data-access/src/util/index.js‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,9 @@ export {
4444
SEMANTIC_CHUNK_SIZE,
4545
QUERY_HASH_CHUNK_SIZE,
4646
MAX_SOURCE_TEXT_LENGTH,
47+
OPPORTUNITY_SEMANTIC_SOURCE_TYPES,
48+
OPPORTUNITY_SEMANTIC_ENTITY_TYPES,
49+
SEMANTIC_MATCHING_CONFIG,
4750
normalizeText,
4851
hashText,
4952
cleanTopicText,

‎packages/spacecat-shared-data-access/src/util/semantic-index.utils.js‎

Lines changed: 74 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,8 @@
1212

1313
import { createHash } from 'crypto';
1414

15+
import { OPPORTUNITY_TYPES } from '@adobe/spacecat-shared-utils';
16+
1517
import { DataAccessError, ValidationError } from '../errors/index.js';
1618
import { DEFAULT_PAGE_SIZE } from './postgrest.utils.js';
1719

@@ -39,6 +41,34 @@ export const QUERY_HASH_CHUNK_SIZE = 50;
3941
/** Max topic/query text length; matches the `source_text` DB CHECK. */
4042
export const MAX_SOURCE_TEXT_LENGTH = 2048;
4143

44+
/**
45+
* Kinds of source text in `opportunity_semantic_embedding.source_type`. The writer and reader
46+
* reject any other value; add a kind here before indexing or searching it.
47+
*/
48+
export const OPPORTUNITY_SEMANTIC_SOURCE_TYPES = Object.freeze({
49+
TOPIC: 'topic',
50+
});
51+
52+
/**
53+
* Opportunity types allowed in `opportunity_semantic_embedding.entity_type`. The writer and
54+
* reader reject any other value; add a type here before indexing or filtering by it.
55+
*/
56+
export const OPPORTUNITY_SEMANTIC_ENTITY_TYPES = Object.freeze({
57+
CITED_ANALYSIS: OPPORTUNITY_TYPES.CITED_ANALYSIS,
58+
REDDIT_ANALYSIS: OPPORTUNITY_TYPES.REDDIT_ANALYSIS,
59+
YOUTUBE_ANALYSIS: OPPORTUNITY_TYPES.YOUTUBE_ANALYSIS,
60+
});
61+
62+
/**
63+
* The embedding generation every semantic writer and reader uses: stored on each row and part of
64+
* the query-cache key. `embeddingDims` must match the data-service `vector(1536)` columns and
65+
* CHECKs. A code constant, not env config: writer and reader must never disagree.
66+
*/
67+
export const SEMANTIC_MATCHING_CONFIG = Object.freeze({
68+
embeddingModel: 'azure/text-embedding-3-small',
69+
embeddingDims: 1536,
70+
});
71+
4272
function assertClient(postgrestClient) {
4373
if (!postgrestClient || typeof postgrestClient.from !== 'function') {
4474
throw new ValidationError('postgrestClient is required');
@@ -66,6 +96,34 @@ function assertModel(value) {
6696
}
6797
}
6898

99+
function assertAllowed(value, name, registry) {
100+
const allowed = Object.values(registry);
101+
if (!allowed.includes(value)) {
102+
throw new ValidationError(`${name} must be one of: ${allowed.join(', ')}`);
103+
}
104+
}
105+
106+
/**
107+
* Validates a type list against its registry and dedups it, so its length never exceeds the
108+
* registry's. Omitted/empty returns null (no filter) unless `required`.
109+
*/
110+
function toAllowedList(values, name, registry, { required }) {
111+
if (values === undefined || values === null || (Array.isArray(values) && values.length === 0)) {
112+
if (required) {
113+
throw new ValidationError(`${name} must be a non-empty array`);
114+
}
115+
return null;
116+
}
117+
if (!Array.isArray(values)) {
118+
throw new ValidationError(`${name} must be an array`);
119+
}
120+
const allowed = Object.values(registry);
121+
if (!values.every((value) => allowed.includes(value))) {
122+
throw new ValidationError(`${name} must only contain: ${allowed.join(', ')}`);
123+
}
124+
return [...new Set(values)];
125+
}
126+
69127
/** Only an empty/omitted `sources` clears; a non-empty input reduced to nothing throws. */
70128
function assertClearable(sources, entityId, sourceType) {
71129
const explicitClear = sources === undefined || sources === null
@@ -268,8 +326,9 @@ async function clearEntitySource(postgrestClient, table, siteId, entityId, sourc
268326
* @param {object} params
269327
* @param {string} params.siteId - the site the opportunity belongs to
270328
* @param {string} params.entityId - the opportunity id
271-
* @param {string} params.entityType - the opportunity type stored on each row
272-
* @param {string} params.sourceType - the source kind, e.g. `topic`
329+
* @param {string} params.entityType - the opportunity type stored on each row, one of
330+
* `OPPORTUNITY_SEMANTIC_ENTITY_TYPES`
331+
* @param {string} params.sourceType - the source kind, one of `OPPORTUNITY_SEMANTIC_SOURCE_TYPES`
273332
* @param {Array<{text: string, vector: number[], model: string, dims: number, sourceId?: string}>}
274333
* params.sources - one entry per source text (normalized + hashed here)
275334
* @returns {Promise<number>} size of the stored set for this (entity, sourceType) after sync
@@ -280,8 +339,8 @@ export async function syncOpportunitySemantic(postgrestClient, {
280339
assertClient(postgrestClient);
281340
assertId(siteId, 'siteId');
282341
assertId(entityId, 'entityId');
283-
assertId(entityType, 'entityType');
284-
assertId(sourceType, 'sourceType');
342+
assertAllowed(entityType, 'entityType', OPPORTUNITY_SEMANTIC_ENTITY_TYPES);
343+
assertAllowed(sourceType, 'sourceType', OPPORTUNITY_SEMANTIC_SOURCE_TYPES);
285344

286345
const table = 'opportunity_semantic_embedding';
287346
const rows = toRows({
@@ -349,7 +408,10 @@ export async function copyEntityVectors(postgrestClient, {
349408
* @param {object} postgrestClient - `@supabase/postgrest-js` client
350409
* @param {object} params
351410
* @param {string} params.siteId - the site to scope the search to
352-
* @param {string} params.sourceType - the source kind, e.g. `topic`
411+
* @param {string[]} params.sourceTypes - source kinds to search, from
412+
* `OPPORTUNITY_SEMANTIC_SOURCE_TYPES` (required, non-empty)
413+
* @param {string[]} [params.entityTypes] - opportunity types to narrow to, from
414+
* `OPPORTUNITY_SEMANTIC_ENTITY_TYPES`; omitted or empty searches all
353415
* @param {number[][]} params.vectors - the query embeddings
354416
* @param {string} params.model - the model the queries were embedded with
355417
* @param {number} params.dims - the query embedding dimension
@@ -359,11 +421,12 @@ export async function copyEntityVectors(postgrestClient, {
359421
* best-first list per input vector, in input order
360422
*/
361423
export async function lookupOpportunitiesByVectors(postgrestClient, {
362-
siteId, sourceType, vectors, model, dims, k = 10, minScore = 0,
424+
siteId, sourceTypes, entityTypes, vectors, model, dims, k = 10, minScore = 0,
363425
} = {}) {
364426
assertClient(postgrestClient);
365427
assertId(siteId, 'siteId');
366-
assertId(sourceType, 'sourceType');
428+
const sourceTypeList = toAllowedList(sourceTypes, 'sourceTypes', OPPORTUNITY_SEMANTIC_SOURCE_TYPES, { required: true });
429+
const entityTypeList = toAllowedList(entityTypes, 'entityTypes', OPPORTUNITY_SEMANTIC_ENTITY_TYPES, { required: false });
367430
assertModel(model);
368431
assertDims(dims);
369432
if (!Array.isArray(vectors)) {
@@ -384,23 +447,24 @@ export async function lookupOpportunitiesByVectors(postgrestClient, {
384447
// eslint-disable-next-line no-await-in-loop
385448
const { data, error } = await postgrestClient.rpc(SEMANTIC_SEARCH_RPC, {
386449
p_site_id: siteId,
387-
p_source_type: sourceType,
450+
p_source_types: sourceTypeList,
451+
p_entity_types: entityTypeList,
388452
p_query_embeddings: group,
389453
p_model: model,
390454
p_dims: dims,
391455
p_limit: k,
392456
p_min_score: minScore,
393457
});
394458
if (error) {
395-
throw new DataAccessError(`Failed semantic search for site ${siteId}`, { siteId, sourceType }, error);
459+
throw new DataAccessError(`Failed semantic search for site ${siteId}`, { siteId, sourceTypes: sourceTypeList }, error);
396460
}
397461
for (const row of data ?? []) {
398462
// Checked against this group, not all results, so a bad index can't land in another group.
399463
const { query_index: queryIndex } = row;
400464
if (!Number.isInteger(queryIndex) || queryIndex < 0 || queryIndex >= group.length) {
401465
throw new DataAccessError(
402466
`Unexpected query_index ${queryIndex} from ${SEMANTIC_SEARCH_RPC}`,
403-
{ siteId, sourceType },
467+
{ siteId, sourceTypes: sourceTypeList },
404468
);
405469
}
406470
results[offset + queryIndex].push({

0 commit comments

Comments
 (0)