From a532f0bebffc2bd191665ef4abea0a41fbc64309 Mon Sep 17 00:00:00 2001 From: Croissant Le Doux Date: Thu, 16 Jul 2026 16:26:41 -0400 Subject: [PATCH] feat(scoring): nightly grant-embedding workflow (Stage 2, step 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit embedGrants (cron 04:15, after the ingest crons): open grants with a synopsis and no vector → buildGrantEmbeddingText (pure, tested: title+funder+program areas+synopsis, 8K cap) → gemini-embedding-001 @ 1536 dims RETRIEVAL_DOCUMENT (org profiles will embed as RETRIEVAL_QUERY on the other side) → grants.synopsis_embedding. Chunked embed→store (100/chunk) so failures resume from the last stored chunk; 500/run spend cap. Core: serverListGrantsNeedingEmbedding (open+unembedded, closest deadline first), serverSetGrantEmbeddings. Worker gains @novelpad/outreach-ai dep; wired into main.ts and run-once (incl. the missed run-once deps injection). Live-verified: 199/199 open grants embedded in 16s; semantic probe ('after-school STEM education for youth') ranks NCI Youth Enjoy Science R25 first at 0.639 cosine via the hnsw index. Co-Authored-By: Claude Fable 5 --- apps/outreach-worker/package.json | 1 + apps/outreach-worker/src/main.ts | 2 + apps/outreach-worker/src/run-once.ts | 7 + .../src/workflows/embed-grants.ts | 170 ++++++++++++++++++ docs/features/ingestion.md | 1 + .../src/grant-embedding-text.test.ts | 39 ++++ .../outreach-ai/src/grant-embedding-text.ts | 35 ++++ packages/outreach-ai/src/index.ts | 1 + .../src/grants/actions/index.server.ts | 1 + .../actions/set-grant-embeddings.server.ts | 27 +++ .../src/grants/queries/index.server.ts | 1 + .../list-grants-needing-embedding.server.ts | 42 +++++ yarn.lock | 3 +- 13 files changed, 329 insertions(+), 1 deletion(-) create mode 100644 apps/outreach-worker/src/workflows/embed-grants.ts create mode 100644 packages/outreach-ai/src/grant-embedding-text.test.ts create mode 100644 packages/outreach-ai/src/grant-embedding-text.ts create mode 100644 packages/outreach-core/src/grants/actions/set-grant-embeddings.server.ts create mode 100644 packages/outreach-core/src/grants/queries/list-grants-needing-embedding.server.ts diff --git a/apps/outreach-worker/package.json b/apps/outreach-worker/package.json index 0afa021..fff53de 100644 --- a/apps/outreach-worker/package.json +++ b/apps/outreach-worker/package.json @@ -17,6 +17,7 @@ "dependencies": { "@dbos-inc/dbos-sdk": "4.17.6", "@dbos-inc/drizzle-datasource": "4.17.6", + "@novelpad/outreach-ai": "workspace:^", "@novelpad/outreach-core": "workspace:^", "drizzle-orm": "0.44.6", "fast-xml-parser": "^4.5.0", diff --git a/apps/outreach-worker/src/main.ts b/apps/outreach-worker/src/main.ts index 5e68072..f0d7d61 100644 --- a/apps/outreach-worker/src/main.ts +++ b/apps/outreach-worker/src/main.ts @@ -30,6 +30,7 @@ import pg from 'pg'; // scheduled-function args, so the `db` handle can't be passed through the // scheduler; it's threaded in via this module-scope registry instead (same // pattern as novelpad-desktop's `setStartDeps`). +import { setEmbedGrantsDeps } from './workflows/embed-grants.js'; import { setEnrichOrgsDeps } from './workflows/enrich-orgs.js'; import { setExpireGrantsDeps } from './workflows/expire-grants.js'; import { setIngestGrantsDeps } from './workflows/ingest-grants.js'; @@ -65,6 +66,7 @@ async function main() { setIngestPndRssDeps({ db }); setIngestNhdojOrgsDeps({ db }); setEnrichOrgsDeps({ db }); + setEmbedGrantsDeps({ db }); DBOS.setConfig({ name: 'helmdocs-outreach-worker', diff --git a/apps/outreach-worker/src/run-once.ts b/apps/outreach-worker/src/run-once.ts index 918cf3b..a2ab8d9 100644 --- a/apps/outreach-worker/src/run-once.ts +++ b/apps/outreach-worker/src/run-once.ts @@ -19,6 +19,10 @@ import { schema } from '@novelpad/outreach-core'; import { drizzle } from 'drizzle-orm/node-postgres'; import pg from 'pg'; +import { + runEmbedGrantsNow, + setEmbedGrantsDeps, +} from './workflows/embed-grants.js'; import { runEnrichOrgsNow, setEnrichOrgsDeps } from './workflows/enrich-orgs.js'; import { runExpireGrantsNow, @@ -43,6 +47,7 @@ const RUNNERS: Record Promise> = { ingestNhdojOrgs: runIngestNhdojOrgsNow, enrichOrgs: runEnrichOrgsNow, expireGrants: runExpireGrantsNow, + embedGrants: runEmbedGrantsNow, }; const FIRST_RUN_ORDER = [ @@ -51,6 +56,7 @@ const FIRST_RUN_ORDER = [ 'ingestNhdojOrgs', 'expireGrants', 'enrichOrgs', + 'embedGrants', ]; if (process.env.DATABASE_URL == null) { @@ -85,6 +91,7 @@ async function main() { setIngestPndRssDeps({ db }); setIngestNhdojOrgsDeps({ db }); setEnrichOrgsDeps({ db }); + setEmbedGrantsDeps({ db }); DBOS.setConfig({ name: 'helmdocs-outreach-worker', diff --git a/apps/outreach-worker/src/workflows/embed-grants.ts b/apps/outreach-worker/src/workflows/embed-grants.ts new file mode 100644 index 0000000..25961b0 --- /dev/null +++ b/apps/outreach-worker/src/workflows/embed-grants.ts @@ -0,0 +1,170 @@ +/** + * Nightly grant-embedding workflow — Stage 2, step 1. + * + * Embeds open grants' synopses with gemini-embedding-001 (1536 dims, + * RETRIEVAL_DOCUMENT task type — org profiles embed as RETRIEVAL_QUERY on + * the other side of the mission-fit comparison) and stores the vectors in + * `grants.synopsis_embedding`, where the hnsw cosine index serves the + * scoring engine's mission-fit subscore. + * + * Runs after the ingest crons (04:15 vs 03:00/03:30) so fresh grants embed + * the same night they land. This is the pipeline's first paid AI call — + * order of magnitude: ~500 tokens/grant, so a 200-grant batch is a few + * cents of Vertex spend. + * + * Registration follows `ingest-grants.ts` exactly (dual workflow+scheduled + * registration, module-scope deps registry, globalThis guard). + */ +import { DBOS, SchedulerMode } from '@dbos-inc/dbos-sdk'; +import { + buildGrantEmbeddingText, + generateDocumentEmbeddings, +} from '@novelpad/outreach-ai'; +import type { schema } from '@novelpad/outreach-core'; +import { + serverListGrantsNeedingEmbedding, + serverSetGrantEmbeddings, + type GrantNeedingEmbedding, +} from '@novelpad/outreach-core/server'; +import type { NodePgDatabase } from 'drizzle-orm/node-postgres'; + +export type OutreachDb = NodePgDatabase; + +/** Per-run cap: bounds nightly spend; the backlog drains across nights. */ +const EMBED_BATCH_LIMIT = 500; +/** gemini-embedding-001 batch endpoint cap handled in the ai package (100). */ +const STORE_CHUNK_SIZE = 100; + +export interface EmbedGrantsDeps { + readonly db: OutreachDb; +} + +let registeredDeps: EmbedGrantsDeps | null = null; + +export function setEmbedGrantsDeps(deps: EmbedGrantsDeps): void { + registeredDeps = deps; +} + +function getEmbedGrantsDeps(): EmbedGrantsDeps { + if (registeredDeps == null) { + throw new Error( + 'EmbedGrantsDeps not registered. Call setEmbedGrantsDeps() before DBOS.launch().', + ); + } + return registeredDeps; +} + +async function listGrantsNeedingEmbedding( + db: OutreachDb, +): Promise { + return serverListGrantsNeedingEmbedding(db, { limit: EMBED_BATCH_LIMIT }); +} +const listGrantsNeedingEmbeddingStep = DBOS.registerStep( + listGrantsNeedingEmbedding, + { name: 'listGrantsNeedingEmbedding', retriesAllowed: true, maxAttempts: 3 }, +); + +async function embedGrantChunk( + grants: GrantNeedingEmbedding[], +): Promise { + const texts = grants.map((grant) => buildGrantEmbeddingText(grant)); + return generateDocumentEmbeddings(texts); +} +const embedGrantChunkStep = DBOS.registerStep(embedGrantChunk, { + name: 'embedGrantChunk', + retriesAllowed: true, + maxAttempts: 3, +}); + +async function storeGrantEmbeddings( + db: OutreachDb, + grants: GrantNeedingEmbedding[], + vectors: number[][], +): Promise { + await serverSetGrantEmbeddings( + db, + grants.map((grant, i) => ({ grantId: grant.id, embedding: vectors[i]! })), + ); +} +const storeGrantEmbeddingsStep = DBOS.registerStep(storeGrantEmbeddings, { + name: 'storeGrantEmbeddings', + retriesAllowed: true, + maxAttempts: 3, +}); + +async function runEmbedGrants(): Promise { + const { db } = getEmbedGrantsDeps(); + + const pending = await listGrantsNeedingEmbeddingStep(db); + if (pending.length === 0) { + console.log('[embed-grants] nothing to embed'); + return; + } + + let embedded = 0; + // Chunked embed→store so a mid-run failure resumes from the last stored + // chunk instead of re-paying for the whole batch. + for (let i = 0; i < pending.length; i += STORE_CHUNK_SIZE) { + const chunk = pending.slice(i, i + STORE_CHUNK_SIZE); + const vectors = await embedGrantChunkStep(chunk); + if (vectors.length !== chunk.length) { + throw new Error( + `[embed-grants] embedding count mismatch: ${vectors.length} vectors for ${chunk.length} grants`, + ); + } + await storeGrantEmbeddingsStep(db, chunk, vectors); + embedded += chunk.length; + } + + console.log( + `[embed-grants] embedded=${embedded} (of ${pending.length} pending this run)`, + ); +} + +const g = globalThis as unknown as { + __outreachEmbedGrantsRegistered?: boolean; + __outreachEmbedGrantsHandle?: ( + scheduledTime: Date, + startedAt: Date, + ) => Promise; +}; + +if (!g.__outreachEmbedGrantsRegistered) { + g.__outreachEmbedGrantsRegistered = true; + + const embedGrants = async (_scheduledTime: Date, _startedAt: Date) => { + try { + await runEmbedGrants(); + } catch (err) { + console.error('[embed-grants] pass failed:', err); + throw err; + } + }; + + // Must be registered as BOTH a workflow and a scheduled function, + // referencing the same function object — see module doc comment. + g.__outreachEmbedGrantsHandle = DBOS.registerWorkflow(embedGrants, { + name: 'embedGrants', + }); + DBOS.registerScheduled(embedGrants, { + crontab: '15 4 * * *', + name: 'embedGrants', + mode: SchedulerMode.ExactlyOncePerInterval, + }); +} + +/** + * Starts one durable run of this workflow immediately through DBOS — + * the exact production path (workflow + checkpointed steps), used by + * `run-once.ts` for supervised/manual passes. Requires deps injected and + * `DBOS.launch()` completed. + */ +export function runEmbedGrantsNow(): Promise { + const handle = g.__outreachEmbedGrantsHandle; + if (handle == null) { + throw new Error( + 'embedGrants is not registered; was this module imported before DBOS.launch()?', + ); + } + return handle(new Date(), new Date()); +} diff --git a/docs/features/ingestion.md b/docs/features/ingestion.md index 834b5fa..2292e2b 100644 --- a/docs/features/ingestion.md +++ b/docs/features/ingestion.md @@ -12,6 +12,7 @@ Grant upserts key on `grants.source_url`; org registry upserts key on case-insen | `ingestPndRss` | `30 3 * * *` | Philanthropy News Digest RFP feed | | `expireGrants` | `0 * * * *` | (sweep: `status='expired'` past `close_date`) | | `enrichOrgs` | `0 5 * * *` | ProPublica Nonprofit Explorer | +| `embedGrants` | `15 4 * * *` | gemini-embedding-001 over open-grant synopses → `grants.synopsis_embedding` (first paid AI call; ~pennies/batch) | | `ingestNhdojOrgs` | `0 4 1 * *` | NHDOJ Charitable Trusts registry PDF | ## Sources diff --git a/packages/outreach-ai/src/grant-embedding-text.test.ts b/packages/outreach-ai/src/grant-embedding-text.test.ts new file mode 100644 index 0000000..9dc51e8 --- /dev/null +++ b/packages/outreach-ai/src/grant-embedding-text.test.ts @@ -0,0 +1,39 @@ +import { describe, expect, it } from 'vitest'; + +import { buildGrantEmbeddingText } from './grant-embedding-text.js'; + +describe('buildGrantEmbeddingText', () => { + it('composes title, funder, program areas, and synopsis', () => { + const text = buildGrantEmbeddingText({ + funder: 'National Science Foundation', + title: 'STEM Education Grants', + synopsis: 'Funding for after-school STEM programs.', + programAreas: ['B25', 'O50'], + }); + expect(text).toBe( + 'STEM Education Grants — National Science Foundation.\n' + + 'Program areas: B25, O50.\n' + + 'Funding for after-school STEM programs.', + ); + }); + + it('omits empty program areas and blank synopsis', () => { + const text = buildGrantEmbeddingText({ + funder: 'F', + title: 'T', + synopsis: ' ', + programAreas: [], + }); + expect(text).toBe('T — F.'); + }); + + it('truncates very long synopses', () => { + const text = buildGrantEmbeddingText({ + funder: 'F', + title: 'T', + synopsis: 'x'.repeat(20_000), + programAreas: null, + }); + expect(text.length).toBe(8_000); + }); +}); diff --git a/packages/outreach-ai/src/grant-embedding-text.ts b/packages/outreach-ai/src/grant-embedding-text.ts new file mode 100644 index 0000000..37d3c22 --- /dev/null +++ b/packages/outreach-ai/src/grant-embedding-text.ts @@ -0,0 +1,35 @@ +/** + * Composes the text that represents a grant in embedding space. Pure — + * shared by the nightly embed job and any ad-hoc re-embedding so a grant + * is always embedded from identically-composed text. + * + * Kept intentionally simple: funder + title + program areas + synopsis, + * truncated to stay well inside the embedding model's context. Mission-fit + * retrieval compares org-profile text against this, so the composition + * should read like a description, not a key-value dump. + */ + +export interface GrantEmbeddingInput { + readonly funder: string; + readonly title: string; + readonly synopsis: string | null; + readonly programAreas: readonly string[] | null; +} + +/** ~8K chars ≈ well under gemini-embedding-001's 2048-token input limit + * for typical English prose after the API's own truncation; the tail of a + * long federal synopsis is boilerplate anyway. */ +const MAX_TEXT_LENGTH = 8_000; + +export function buildGrantEmbeddingText(grant: GrantEmbeddingInput): string { + const parts: string[] = [`${grant.title} — ${grant.funder}.`]; + + if (grant.programAreas != null && grant.programAreas.length > 0) { + parts.push(`Program areas: ${grant.programAreas.join(', ')}.`); + } + if (grant.synopsis != null && grant.synopsis.trim() !== '') { + parts.push(grant.synopsis.trim()); + } + + return parts.join('\n').slice(0, MAX_TEXT_LENGTH); +} diff --git a/packages/outreach-ai/src/index.ts b/packages/outreach-ai/src/index.ts index a94db87..b2116bf 100644 --- a/packages/outreach-ai/src/index.ts +++ b/packages/outreach-ai/src/index.ts @@ -5,3 +5,4 @@ export * from './agents/org-profiler/schema.js'; export * from './agents/org-profiler/run.js'; export * from './agents/mission-fit-judge/schema.js'; export * from './agents/mission-fit-judge/run.js'; +export * from './grant-embedding-text.js'; diff --git a/packages/outreach-core/src/grants/actions/index.server.ts b/packages/outreach-core/src/grants/actions/index.server.ts index f91e9ee..a57be42 100644 --- a/packages/outreach-core/src/grants/actions/index.server.ts +++ b/packages/outreach-core/src/grants/actions/index.server.ts @@ -1,2 +1,3 @@ export * from './expire-closed-grants.server.js'; export * from './insert-grants.server.js'; +export * from './set-grant-embeddings.server.js'; diff --git a/packages/outreach-core/src/grants/actions/set-grant-embeddings.server.ts b/packages/outreach-core/src/grants/actions/set-grant-embeddings.server.ts new file mode 100644 index 0000000..a672de4 --- /dev/null +++ b/packages/outreach-core/src/grants/actions/set-grant-embeddings.server.ts @@ -0,0 +1,27 @@ +import { eq, sql } from 'drizzle-orm'; + +import type { NpOutreachDatabase, NpOutreachTransaction } from '#~/db/db.js'; +import { schema } from '#~/db/db.js'; + +export interface GrantEmbedding { + readonly grantId: string; + /** 1536-dim vector from gemini-embedding-001 (RETRIEVAL_DOCUMENT). */ + readonly embedding: number[]; +} + +/** + * Stores synopsis embeddings for a batch of grants. Plain sequential + * updates — batches are small (≤100) and this runs inside a nightly job, + * so a multi-row VALUES join isn't worth the SQL gymnastics yet. + */ +export async function serverSetGrantEmbeddings( + db: NpOutreachDatabase | NpOutreachTransaction, + embeddings: ReadonlyArray, +): Promise { + for (const { grantId, embedding } of embeddings) { + await db + .update(schema.grants) + .set({ synopsisEmbedding: embedding, updatedAt: sql`now()` }) + .where(eq(schema.grants.id, grantId)); + } +} diff --git a/packages/outreach-core/src/grants/queries/index.server.ts b/packages/outreach-core/src/grants/queries/index.server.ts index 598abac..5577826 100644 --- a/packages/outreach-core/src/grants/queries/index.server.ts +++ b/packages/outreach-core/src/grants/queries/index.server.ts @@ -1 +1,2 @@ export * from './list-open-grants.server.js'; +export * from './list-grants-needing-embedding.server.js'; diff --git a/packages/outreach-core/src/grants/queries/list-grants-needing-embedding.server.ts b/packages/outreach-core/src/grants/queries/list-grants-needing-embedding.server.ts new file mode 100644 index 0000000..793b171 --- /dev/null +++ b/packages/outreach-core/src/grants/queries/list-grants-needing-embedding.server.ts @@ -0,0 +1,42 @@ +import { and, asc, eq, isNotNull, isNull } from 'drizzle-orm'; + +import type { NpOutreachDatabase, NpOutreachTransaction } from '#~/db/db.js'; +import { schema } from '#~/db/db.js'; + +export interface GrantNeedingEmbedding { + id: string; + funder: string; + title: string; + synopsis: string | null; + programAreas: string[] | null; +} + +/** + * Open grants that have text to embed but no synopsis embedding yet. + * Closed/expired grants are skipped — they can't be matched, so embedding + * them is wasted spend; if one re-opens, its upsert clears nothing, and it + * was embedded while open anyway. + */ +export async function serverListGrantsNeedingEmbedding( + db: NpOutreachDatabase | NpOutreachTransaction, + { limit }: { limit: number }, +): Promise { + return db + .select({ + id: schema.grants.id, + funder: schema.grants.funder, + title: schema.grants.title, + synopsis: schema.grants.synopsis, + programAreas: schema.grants.programAreas, + }) + .from(schema.grants) + .where( + and( + eq(schema.grants.status, 'open'), + isNull(schema.grants.synopsisEmbedding), + isNotNull(schema.grants.synopsis), + ), + ) + .orderBy(asc(schema.grants.closeDate)) + .limit(limit); +} diff --git a/yarn.lock b/yarn.lock index 7ae0dce..3d426bf 100644 --- a/yarn.lock +++ b/yarn.lock @@ -1540,7 +1540,7 @@ __metadata: languageName: unknown linkType: soft -"@novelpad/outreach-ai@workspace:packages/outreach-ai": +"@novelpad/outreach-ai@workspace:^, @novelpad/outreach-ai@workspace:packages/outreach-ai": version: 0.0.0-use.local resolution: "@novelpad/outreach-ai@workspace:packages/outreach-ai" dependencies: @@ -1605,6 +1605,7 @@ __metadata: "@dbos-inc/dbos-sdk": "npm:4.17.6" "@dbos-inc/drizzle-datasource": "npm:4.17.6" "@novelpad/config": "workspace:^" + "@novelpad/outreach-ai": "workspace:^" "@novelpad/outreach-core": "workspace:^" "@types/node": "npm:^22" "@types/pg": "npm:8.20.0"