Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/missionmatch / internal/modules/ai/retrieval.go |
| Date | Not specified in source |
| Repository | Janady13/missionmatch |
| Revision | 720d47fae476b0d38477ae4401697d372b1ea2bc |
| Branch | feature/opportunity-to-outcome |
| Path or archive member | internal/modules/ai/retrieval.go |
| Original SHA-256 | cafd69a00b0cd8c0125a746c189cabb54763a5b7ff0ca94fd2ae57f9ac115aaa |
| Git blob SHA | 21b53d834d9c422a19d4c7e0b4c72f159476857c |
| Reading-copy SHA-256 | cafd69a00b0cd8c0125a746c189cabb54763a5b7ff0ca94fd2ae57f9ac115aaa |
Recovered source transcript
Download complete source text ↓243 lines · 10.5 KB
package ai
import (
"context"
"errors"
"github.com/jackc/pgx/v5"
pgvector "github.com/pgvector/pgvector-go"
"github.com/missionworkforce/missionmatch/internal/platform/database/dbgen"
"github.com/missionworkforce/missionmatch/internal/platform/inference"
"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)
// Hit is one retrieved candidate with its fused relevance score and the matched text span.
//
// Snippet is the matched text span (a Postgres ts_headline over the SAME text the candidate's search_tsv
// is built from, with the SAME tsquery the full text leg already binds), so the faithfulness gate and the
// narration can cite WHY a candidate matched, the matched TEXT, not only the name. An empty Snippet is a
// valid no highlight result (a pure semantic or empty query, or a hit reached only through the vector arm),
// never an error.
type Hit struct {
CandidateID string `json:"candidate_id"`
FullName string `json:"full_name"`
Score float64 `json:"rrf_score"`
Snippet string `json:"snippet"`
}
// HybridSearch is the RAG retrieval leg: it embeds the query through the MEGAMIND plane and runs the
// existing hybrid query that fuses pgvector cosine kNN with full text rank via Reciprocal Rank Fusion
// (docs/AI_RESEARCH.md, pgvector + RRF). It runs RLS scoped as the tenant, so it returns only the
// caller's candidates, and it tunes the HNSW scan per the pgvector 0.8 guidance before the search.
//
// When mission is a non empty mission keyword (from MissionFromQuestion, drawn only from the fixed
// mission_spine vocabulary), retrieval is additionally constrained to that mission. An empty mission runs the
// unscoped query byte for byte as before, so the default path is unchanged.
//
// WS6 to WS7 upgrade (the mission gate is now structured, additively, with the WS6 text filter kept as a
// fallback): when the tenant has been backfilled by the mission tagging pass (candidate_mission_tags has at
// least one row, probed RLS scoped in the same transaction) AND the keyword maps to a known mission code, the
// scoped query is SearchCandidatesHybridMissionTagged, which gates both arms on a STRUCTURED EXISTS predicate
// over candidate_mission_tags rather than the WS6 search_tsv text match. That is the candidate's real
// classified mission, so it reaches a candidate whose mission lives in the resume body, not only one whose
// full_name, headline, or location text happens to contain the mission keyword (the WS6 limit, lifted once
// tags exist). When the tenant is NOT yet tagged, or the keyword has no code, retrieval falls back to the WS6
// SearchCandidatesHybridMission text query, so an un backfilled tenant keeps exact WS6 behavior. Either scoped
// path keeps the RLS scoping, the ForbidEEO guard, and the HNSW SET LOCAL tuning of the unscoped path.
//
// It never touches the eeo schema (the query reads only candidates and their own mission tags, never the eeo
// schema, and mission tags are never a scoring input), and the query string is guarded so a demographic
// reference cannot enter the retrieval path.
func HybridSearch(ctx context.Context, db *tenant.DB, client *inference.Client, query, mission string, limit int) ([]Hit, error) {
if db == nil || client == nil {
return nil, errors.New("ai: retrieval needs a database and an inference client")
}
if err := ForbidEEO(query); err != nil {
return nil, err
}
if limit <= 0 || limit > 100 {
limit = 20
}
emb, err := client.Embed(ctx, []string{query})
if err != nil {
return nil, err
}
if len(emb.Embeddings) != 1 {
return nil, errors.New("ai: query embedding failed")
}
qv := pgvector.NewVector(emb.Embeddings[0])
var hits []Hit
err = db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
// pgvector 0.8 tuning: a wider ef_search lifts recall on the filtered (RLS) scan, and relaxed
// iterative scan avoids overfiltering under the tenant predicate (docs/AI_RESEARCH.md, pgvector).
if _, err := tx.Exec(ctx, "SET LOCAL hnsw.ef_search = 100"); err != nil {
return err
}
if _, err := tx.Exec(ctx, "SET LOCAL hnsw.iterative_scan = 'relaxed_order'"); err != nil {
return err
}
q := dbgen.New(tx)
if mission != "" {
// WS7 structured mission gate, preferred when the tenant has been tagged and the keyword maps to a
// known mission code. The probe is RLS scoped, so it counts only this tenant's tags. When either
// condition is unmet the WS6 text query runs instead, so an un backfilled tenant is byte identical
// to WS6.
code := MissionCodeForTerm(mission)
if code != "" {
tagCount, err := q.CountCandidateMissionTags(ctx)
if err != nil {
return err
}
if tagCount > 0 {
rows, err := q.SearchCandidatesHybridMissionTagged(ctx, dbgen.SearchCandidatesHybridMissionTaggedParams{
QueryVector: qv, RowLimit: int32(limit), TextQuery: query, MissionCode: code,
})
if err != nil {
return err
}
hits = make([]Hit, len(rows))
for i, r := range rows {
hits[i] = Hit{CandidateID: r.ID.String(), FullName: r.FullName, Score: r.RrfScore, Snippet: r.Snippet}
}
return nil
}
}
// WS6 fallback: the tenant has no mission tags yet, or the keyword has no structured code, so the
// documented WS6 search_tsv text filter scopes the mission (byte for byte the WS6 behavior).
rows, err := q.SearchCandidatesHybridMission(ctx, dbgen.SearchCandidatesHybridMissionParams{
QueryVector: qv, RowLimit: int32(limit), TextQuery: query, MissionQuery: mission,
})
if err != nil {
return err
}
hits = make([]Hit, len(rows))
for i, r := range rows {
hits[i] = Hit{CandidateID: r.ID.String(), FullName: r.FullName, Score: r.RrfScore, Snippet: r.Snippet}
}
return nil
}
rows, err := q.SearchCandidatesHybrid(ctx, dbgen.SearchCandidatesHybridParams{
Column1: qv, Limit: int32(limit), WebsearchToTsquery: query,
})
if err != nil {
return err
}
hits = make([]Hit, len(rows))
for i, r := range rows {
hits[i] = Hit{CandidateID: r.ID.String(), FullName: r.FullName, Score: r.RrfScore, Snippet: r.Snippet}
}
return nil
})
return hits, err
}
// OppHit is one retrieved federal opportunity with its fused relevance score and the matched text span. It
// is the market plane's counterpart to Hit: same fusion, same snippet contract, different corpus.
type OppHit struct {
RecordID string `json:"record_id"`
Title string `json:"title"`
Agency string `json:"agency"`
Kind string `json:"record_kind"`
Deadline string `json:"response_deadline"`
Score float64 `json:"rrf_score"`
Snippet string `json:"snippet"`
SourceName string `json:"source_name"`
}
// HybridSearchOpportunities is the SECOND retrieval leg, over the GLOBAL federal opportunity catalog. It is
// the same retrieval contract as HybridSearch (embed the query on the plane, fuse pgvector cosine kNN with
// full text rank by Reciprocal Rank Fusion, tune the HNSW scan first) pointed at a different corpus, so the
// orchestrator gains market recall without a second retrieval design.
//
// Two deliberate differences from the candidate leg, both consequences of what the corpus IS:
//
// 1. opportunity_records is PUBLIC federal reference data with no organization_id and no row level
// security, the same status as the skills ontology. There is therefore no tenant predicate to write and
// none is written. It still runs inside RunInTenantTx because under per tenant routing that is how the
// caller's leaf pool is resolved, and the global catalog lives in each tenant database as part of the
// shared migration set.
// 2. The SQL is written here rather than generated, because sqlc models a table from the migration set at
// generation time and this query fuses two arms over a column added later; the shape is the same
// MATERIALIZED two arm fusion with k=60 the candidate query uses.
//
// The EEO guard runs exactly as it does on the candidate leg. A public solicitation carries no protected
// attribute, but the guard is on the QUESTION, and a question that reaches for demographics must not be
// able to reach a retrieval path at all just because it names a contract.
func HybridSearchOpportunities(ctx context.Context, db *tenant.DB, client *inference.Client, query string, limit int) ([]OppHit, error) {
if db == nil || client == nil {
return nil, errors.New("ai: opportunity retrieval needs a database and an inference client")
}
if err := ForbidEEO(query); err != nil {
return nil, err
}
if limit <= 0 || limit > 100 {
limit = 20
}
emb, err := client.Embed(ctx, []string{query})
if err != nil {
return nil, err
}
if len(emb.Embeddings) != 1 {
return nil, errors.New("ai: query embedding failed")
}
qv := pgvector.NewVector(emb.Embeddings[0])
const oppRRF = `
WITH vec AS MATERIALIZED (
SELECT id, row_number() OVER (ORDER BY embedding <=> $1) AS rnk
FROM opportunity_records
WHERE embedding IS NOT NULL
ORDER BY embedding <=> $1
LIMIT 60
), fts AS (
SELECT id, row_number() OVER (ORDER BY ts_rank(search_tsv, websearch_to_tsquery('english', $2)) DESC) AS rnk
FROM opportunity_records
WHERE $2 <> '' AND search_tsv @@ websearch_to_tsquery('english', $2)
LIMIT 60
)
SELECT r.id::text,
COALESCE(r.title, ''),
COALESCE(r.agency, ''),
COALESCE(r.record_kind, ''),
COALESCE(to_char(r.response_deadline, 'YYYY-MM-DD'), ''),
(COALESCE(1.0/(60+vec.rnk), 0) + COALESCE(1.0/(60+fts.rnk), 0))::float8 AS rrf_score,
COALESCE(ts_headline('english',
COALESCE(r.title,'') || '. ' || COALESCE(r.agency,'') || ' ' || COALESCE(r.sub_agency,'') ||
'. NAICS ' || COALESCE(r.naics,'') || '. ' || COALESCE(r.set_aside,''),
websearch_to_tsquery('english', NULLIF($2, '')),
'MaxFragments=1,MaxWords=22,MinWords=6'), ''),
COALESCE(r.source_name, '')
FROM opportunity_records r
LEFT JOIN vec ON vec.id = r.id
LEFT JOIN fts ON fts.id = r.id
WHERE vec.id IS NOT NULL OR fts.id IS NOT NULL
ORDER BY rrf_score DESC
LIMIT $3`
var hits []OppHit
err = db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
if _, err := tx.Exec(ctx, "SET LOCAL hnsw.ef_search = 100"); err != nil {
return err
}
if _, err := tx.Exec(ctx, "SET LOCAL hnsw.iterative_scan = 'relaxed_order'"); err != nil {
return err
}
rows, err := tx.Query(ctx, oppRRF, qv, query, limit)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var h OppHit
if err := rows.Scan(&h.RecordID, &h.Title, &h.Agency, &h.Kind, &h.Deadline, &h.Score, &h.Snippet, &h.SourceName); err != nil {
return err
}
hits = append(hits, h)
}
return rows.Err()
})
return hits, err
}