Vector database for RAG: wording and meaning in one query

docker run -d --name serenedb -p 7890:7890 \
    -e POSTGRES_PASSWORD=<password> \
    -v serenedb-data:/var/lib/serenedb serenedb/serenedb
pip install langchain-serenedb

# for the SQL tab
psql -h localhost -p 7890 -U postgres

from langchain_serenedb import Column, HybridIndexConfig, SereneDBEngine

engine = SereneDBEngine.from_connection_string(
    "host=127.0.0.1 port=7890 user=postgres password=<password> dbname=postgres"
)
engine.init_vectorstore_table(
    "chunks",
    vector_size=768,  # your embedding model's size
    metadata_columns=[Column("access", "TEXT"), Column("source", "TEXT")],
    hybrid_index_config=HybridIndexConfig(),
)

CREATE TEXT SEARCH DICTIONARY rag_en AS
    split_text(case := 'lower')
    | remove_stopwords(stopwords := ['a', 'an', 'and', 'at', 'do', 'does',
        'how', 'i', 'in', 'is', 'it', 'its', 'of', 'on', 'the', 'to',
        'what', 'when', 'who', 'with'])
    | stem_words(locale := 'en')
    WITH (frequency, position, norm);

CREATE TABLE documents (
    id     INTEGER PRIMARY KEY,
    title  VARCHAR,
    source VARCHAR
);
CREATE TABLE chunks (
    id        INTEGER PRIMARY KEY,
    doc_id    INTEGER,
    access    VARCHAR,
    content   VARCHAR,
    embedding FLOAT[4]
);

CREATE INDEX chunks_idx ON chunks
    USING inverted (
        id,
        access,                            -- filter
        content rag_en,                    -- bm25
        embedding ivf (metric = 'cosine')  -- ann
    );

from langchain_serenedb import FusionStrategy, HybridSearchConfig, SereneDBVectorStore

store = SereneDBVectorStore.create_sync(
    engine,
    embedding_service=embeddings,  # any LangChain embeddings model
    table_name="chunks",
    metadata_columns=["access", "source"],
    hybrid_search_config=HybridSearchConfig(
        fusion=FusionStrategy.RRF, primary_top_k=20, secondary_top_k=20
    ),
)
store.add_documents(chunks)

INSERT INTO documents VALUES
 (1, 'Incident AB-1234 postmortem', 'ops/incidents/ab-1234.md'),
 (2, 'Runbook: AB-1234 rollback',   'ops/runbooks/rollback.md'),
 (3, 'Deploy timeout errors',       'wiki/deploy-errors.md'),
 (4, 'Retry policy for deploys',    'wiki/retry.md'),
 (5, 'Release guide',               'wiki/release.md'),
 (6, 'Connection pool sizing',      'wiki/db-pool.md'),
 (7, 'Customer escalation AB-1234', 'crm/escalations/ab-1234.md');

INSERT INTO chunks VALUES
 (1, 1, 'staff', 'Incident AB-1234: migration AB-1234 locked the orders table and the release was rolled back.', [0.10, 0.70, 0.15, 0.62]),
 (2, 2, 'staff', 'When AB-1234 recurs, revert the migration and redeploy the previous release.', [0.10, 0.75, 0.10, 0.55]),
 (3, 3, 'all',   'A deploy times out when the health check does not pass within 60 seconds.', [0.05, 0.97, 0.05, 0.15]),
 (4, 4, 'all',   'A failed rollout retries twice before it pages the on-call engineer.', [0.10, 0.92, 0.05, 0.20]),
 (5, 5, 'all',   'Ship every release to staging first, then promote it to production.', [0.15, 0.90, 0.05, 0.20]),
 (6, 6, 'all',   'Size the connection pool of the orders service to twice the worker count.', [0.05, 0.85, 0.10, 0.30]),
 (7, 7, 'admin', 'Acme reported failed checkouts during release AB-1234 and asked for credits.', [0.05, 0.50, 0.10, 0.85]);

VACUUM (REFRESH_TABLE) chunks;

hits = store.similarity_search(
    "deploy fails with AB-1234",
    k=5,
    fts_query="AB-1234",
    filter={"access": {"$in": ["all", "staff"]}},
)
for doc in hits:
    print(doc.metadata["source"], doc.page_content[:40])

WITH fused AS (
  SELECT id, RANK() OVER (ORDER BY s DESC) AS rank FROM (
    SELECT id, BM25(chunks_idx.tableoid) AS s FROM chunks_idx
    WHERE content @@ 'deploy fails with AB-1234'
      AND access IN ('all', 'staff')
    ORDER BY s DESC LIMIT 20) lex
  UNION ALL
  SELECT id, RANK() OVER (ORDER BY d) AS rank FROM (
    SELECT id, embedding <=> [0.05, 0.95, 0.05, 0.10]::FLOAT[4] AS d
    FROM chunks_idx
    WHERE access IN ('all', 'staff')
    ORDER BY d LIMIT 20) vec
)
SELECT c.id, d.title, d.source, round(SUM(1.0 / (60 + f.rank)), 4) AS rrf
FROM fused f
JOIN chunks c ON c.id = f.id
JOIN documents d ON d.id = c.doc_id
GROUP BY c.id, d.title, d.source
ORDER BY rrf DESC, c.id
LIMIT 5;

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

retriever = store.as_retriever(
    search_kwargs={"k": 5, "filter": {"access": {"$in": ["all", "staff"]}}}
)
prompt = ChatPromptTemplate.from_template(
    "Answer from the numbered context and cite it.\n\n{context}\n\nQuestion: {question}"
)

def number(docs):
    return "\n".join(
        f"[{i}] {d.metadata['source']}: {d.page_content}" for i, d in enumerate(docs, 1)
    )

chain = (
    {"context": retriever | number, "question": RunnablePassthrough()}
    | prompt
    | llm  # any LangChain chat model
    | StrOutputParser()
)
answer = chain.invoke("deploy fails with AB-1234")

CREATE OR REPLACE SECRET openai (TYPE openai, api_key '⟨sk-...⟩', model 'gpt-4o-mini');

WITH fused AS (
  SELECT id, RANK() OVER (ORDER BY s DESC) AS rank FROM (
    SELECT id, BM25(chunks_idx.tableoid) AS s FROM chunks_idx
    WHERE content @@ 'deploy fails with AB-1234'
      AND access IN ('all', 'staff')
    ORDER BY s DESC LIMIT 20) lex
  UNION ALL
  SELECT id, RANK() OVER (ORDER BY d) AS rank FROM (
    SELECT id, embedding <=> [0.05, 0.95, 0.05, 0.10]::FLOAT[4] AS d
    FROM chunks_idx
    WHERE access IN ('all', 'staff')
    ORDER BY d LIMIT 20) vec
),
candidates AS (
  SELECT id FROM fused GROUP BY id
  ORDER BY SUM(1.0 / (60 + rank)) DESC, id LIMIT 5
),
best AS (
  SELECT c.id, d.source, c.content,
         ai_rerank('deploy fails with AB-1234', c.content,
                   secret_name := 'openai') AS relevance
  FROM candidates k
  JOIN chunks c ON c.id = k.id
  JOIN documents d ON d.id = c.doc_id
  ORDER BY relevance DESC, c.id
  LIMIT 3
)
SELECT ai_generate(
         'Question: deploy fails with AB-1234' || chr(10) || string_agg(
           '[' || n || '] ' || source || ': ' || content, chr(10) ORDER BY n),
         'Answer from the numbered context only, and cite it as [n].',
         secret_name := 'openai') AS answer
FROM (SELECT *, ROW_NUMBER() OVER (ORDER BY relevance DESC, id) AS n
      FROM best) numbered;