Build a Python semantic FAQ search with CometAPI embeddings and pgvector

Last reviewed: 2026-07-29

Direct answer

A reliable semantic FAQ search uses one embedding contract for both sides of the comparison. Batch the authored FAQ entries through CometAPI, validate every returned vector, and store the vectors in a fixed-width pgvector column. For each user question, request another vector with the same model and dimensions, then order the stored rows by cosine distance.

This walkthrough uses the documented text-embedding-3-small example with an explicit width of 1,024 dimensions. The width is a tutorial choice, not a universal recommendation. Making it explicit prevents a model or configuration change from silently mixing incompatible rows. The search returns existing FAQ answers; it does not ask a generative model to invent an answer when the match is weak.

Prepare the Python process

Install the client and database packages in an isolated environment:

python -m pip install 'openai' 'psycopg[binary]'

Provide COMETAPI_BASE_URL, COMETAPI_KEY, and DATABASE_URL through the process environment. Set the base URL to the CometAPI /v1 value shown in the CometAPI embeddings reference . Do not print the values. If configuration diagnostics must show a sensitive value, emit [REDACTED] or a boolean that only says whether the variable is present. The same boundary is covered in keeping CometAPI keys out of repositories .

Create the pgvector table

The pgvector supporting files must already be installed on the PostgreSQL server before the extension can be loaded. On a managed service, an administrator or provider control may need to enable it. Apply this schema once:

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE IF NOT EXISTS faq_entries (
    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    question text NOT NULL UNIQUE,
    answer text NOT NULL,
    embedding_model text NOT NULL,
    embedding_dimensions integer NOT NULL
        CHECK (embedding_dimensions = 1024),
    embedding vector(1024) NOT NULL
);

The model and dimension columns make the storage contract inspectable. The vector(1024) declaration also causes PostgreSQL to reject a vector with the wrong width instead of accepting a corrupt mixed index.

Embed, validate, and store the FAQ set

The CometAPI contract accepts an array of strings and returns one indexed embedding object for each input. Although the response order is documented to match the input order, validating the response count and index uniqueness before mapping by index makes the ingestion boundary explicit.

import json
import logging
import os
import time

import psycopg
from openai import OpenAI

MODEL_ID = 'text-embedding-3-small'
DIMENSIONS = 1024

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('faq_search')

client = OpenAI(
    base_url=os.environ['COMETAPI_BASE_URL'],
    api_key=os.environ['COMETAPI_KEY'],
)

FAQS = [
    {
        'question': 'Can I embed several FAQ entries in one request?',
        'answer': 'Yes. Send an array of strings and retain the response index for each entry.',
    },
    {
        'question': 'How does this search rank answers?',
        'answer': 'It embeds the query with the same contract and orders stored rows by cosine distance.',
    },
    {
        'question': 'What happens if a vector has the wrong size?',
        'answer': 'Stop the write, verify the model and dimensions, and rebuild the affected batch.',
    },
    {
        'question': 'When should I add an approximate index?',
        'answer': 'Start with exact search and add an approximate index only after measuring recall and latency.',
    },
]


def vector_text(values):
    if len(values) != DIMENSIONS:
        raise ValueError(
            f'Expected {DIMENSIONS} dimensions, received {len(values)}'
        )
    return json.dumps(values, separators=(',', ':'))


def embed_texts(texts):
    if not texts or any(not text.strip() for text in texts):
        raise ValueError('Embedding input must contain nonempty text')

    started = time.monotonic()
    try:
        response = client.embeddings.create(
            model=MODEL_ID,
            input=texts,
            dimensions=DIMENSIONS,
            encoding_format='float',
        )
    except Exception as exc:
        logger.error(
            'embedding_request_failed',
            extra={
                'operation': 'create_embeddings',
                'model_id': MODEL_ID,
                'requested_dimensions': DIMENSIONS,
                'input_count': len(texts),
                'elapsed_ms': round((time.monotonic() - started) * 1000),
                'exception_type': type(exc).__name__,
            },
        )
        raise RuntimeError('Embedding request failed') from None

    response_indexes = [item.index for item in response.data]
    unique_response_indexes = set(response_indexes)
    expected_indexes = set(range(len(texts)))
    if (
        len(response.data) != len(texts)
        or len(unique_response_indexes) != len(response_indexes)
        or unique_response_indexes != expected_indexes
    ):
        logger.error(
            'embedding_response_shape_invalid',
            extra={
                'operation': 'create_embeddings',
                'model_id': MODEL_ID,
                'input_count': len(texts),
                'returned_count': len(response.data),
                'unique_index_count': len(unique_response_indexes),
            },
        )
        raise RuntimeError('Embedding response indexes did not match the input')

    vectors_by_index = {
        item.index: item.embedding
        for item in response.data
    }

    for index, vector in vectors_by_index.items():
        if len(vector) != DIMENSIONS:
            logger.error(
                'embedding_dimension_mismatch',
                extra={
                    'operation': 'create_embeddings',
                    'model_id': MODEL_ID,
                    'item_index': index,
                    'expected_dimensions': DIMENSIONS,
                    'returned_dimensions': len(vector),
                },
            )
            raise RuntimeError('Embedding dimensions did not match the contract')

    usage = getattr(response, 'usage', None)
    logger.info(
        'embedding_batch_succeeded',
        extra={
            'operation': 'create_embeddings',
            'model_id': MODEL_ID,
            'requested_dimensions': DIMENSIONS,
            'input_count': len(texts),
            'returned_count': len(response.data),
            'prompt_tokens': getattr(usage, 'prompt_tokens', None),
            'elapsed_ms': round((time.monotonic() - started) * 1000),
        },
    )
    return [vectors_by_index[index] for index in range(len(texts))]


def seed_faqs(faqs):
    documents = [
        f'Question: {item["question"]} Answer: {item["answer"]}'
        for item in faqs
    ]
    vectors = embed_texts(documents)

    with psycopg.connect(os.environ['DATABASE_URL']) as connection:
        with connection.cursor() as cursor:
            for item, vector in zip(faqs, vectors, strict=True):
                cursor.execute(
                    '''
                    INSERT INTO faq_entries (
                        question,
                        answer,
                        embedding_model,
                        embedding_dimensions,
                        embedding
                    )
                    VALUES (%s, %s, %s, %s, %s::vector)
                    ON CONFLICT (question) DO UPDATE SET
                        answer = EXCLUDED.answer,
                        embedding_model = EXCLUDED.embedding_model,
                        embedding_dimensions = EXCLUDED.embedding_dimensions,
                        embedding = EXCLUDED.embedding
                    ''',
                    (
                        item['question'],
                        item['answer'],
                        MODEL_ID,
                        DIMENSIONS,
                        vector_text(vector),
                    ),
                )

The API batch finishes before database writes begin. The connection context commits only when all upserts succeed; an exception rolls the transaction back. This keeps a failed request or malformed vector from leaving a partly refreshed FAQ set.

Embed the question and run cosine search

Use the same helper for the reader’s question. Bind values as query parameters, including the serialized vector. Do not concatenate the question or vector into SQL.

def search_faq(query, limit=3):
    if not query.strip():
        raise ValueError('Query must not be empty')
    if not 1 <= limit <= 10:
        raise ValueError('Limit must be between 1 and 10')

    query_vector = vector_text(embed_texts([query])[0])

    with psycopg.connect(os.environ['DATABASE_URL']) as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                '''
                SELECT
                    question,
                    answer,
                    1 - (embedding <=> %s::vector) AS cosine_similarity
                FROM faq_entries
                WHERE embedding_model = %s
                  AND embedding_dimensions = %s
                ORDER BY embedding <=> %s::vector
                LIMIT %s
                ''',
                (
                    query_vector,
                    MODEL_ID,
                    DIMENSIONS,
                    query_vector,
                    limit,
                ),
            )
            return cursor.fetchall()


if __name__ == '__main__':
    seed_faqs(FAQS)
    matches = search_faq('May I send all of the FAQ text together?')
    for question, answer, score in matches:
        print(f'{score:.4f} | {question}')
        print(answer)

Psycopg uses one %s placeholder for each bound value. The vector appears twice because the query calculates a displayed similarity and independently orders by distance. The database receives those values separately from the SQL text.

Run the happy path

Confirm that the three required environment variables are present without printing their values.
Apply the schema and verify that vector appears in pg_extension.
Run the seed process and confirm that four rows share the expected model ID and dimension count.
Search for May I send all of the FAQ text together? and confirm that the batch-input FAQ is the first result.
Review the structured success event for model, counts, dimensions, token usage, and elapsed time.
Add several labeled paraphrases and record whether the expected FAQ appears first. Use those observations to set a product-specific no-match policy rather than copying an arbitrary similarity threshold.

Exercise the error path

Test the dimension guard locally before testing a live request:

try:
    vector_text([0.0] * 8)
except ValueError as exc:
    print(type(exc).__name__)

The test should print only the exception type and must not reach PostgreSQL. For a real API failure, the ingestion process should emit embedding_request_failed, leave the existing FAQ rows untouched, and return a neutral error to its caller. For a database failure, roll back the whole batch and record the stage as database_write at the application boundary.

Useful sanitized fields are operation, model_id, requested_dimensions, input_count, returned_count, prompt_tokens, elapsed_ms, exception_type, database_stage, and an opaque batch identifier. Do not log the environment values, database connection string, source FAQ text, user query, embedding arrays, or a raw response body. Redact any accidentally captured sensitive value as [REDACTED].

Who this is for

This pattern fits Python teams that already use PostgreSQL and need a small semantic lookup feature without adding a separate vector service. It is especially useful for support FAQs, documentation shortcuts, or internal help text where the returned answer must remain authored and reviewable.

It is not a complete retrieval-augmented generation system. There is no document chunker, answer generator, reranker, or universal confidence threshold here. The narrow scope makes it easier to inspect the request contract, storage boundary, ranking query, and failure behavior before expanding the design.

Key takeaways

Index and query with the same model ID, dimensions, and encoding format.
Batch FAQ strings, then validate response count, index uniqueness, and vector widths before writing anything.
Store the model and dimensions beside every vector so drift is visible.
Use Psycopg parameters for text, vectors, filters, and limits.
Begin with pgvector’s exact search; introduce approximate indexing only after measuring the tradeoff.
Treat a weak match as no match. Return an authored fallback instead of presenting an unrelated FAQ confidently.

Sources checked

The CometAPI Create embeddings reference documents the endpoint path, Python client pattern, batch inputs, dimensions option, float encoding, indexed response objects, and usage fields.
The pgvector project documentation documents extension setup, fixed-width vector columns, cosine distance, cosine similarity, exact search, and HNSW and IVFFlat indexes.
The Psycopg parameter documentation explains %s placeholders, separate parameter values, and why values must not be merged into SQL strings.
The PostgreSQL CREATE EXTENSION reference explains server-side extension installation, supporting-file prerequisites, ownership, and privilege considerations.

Contract details to verify

Before deployment, use the model-catalog validation guide to confirm that the intended embedding model remains available and appropriate.

Contract item	Tutorial value	Verification
Route	`POST /v1/embeddings`	Keep the client on the documented CometAPI `/v1` base.
Model	`text-embedding-3-small`	Verify the current model catalog before indexing or rebuilding.
Input	One string for search; an array for ingestion	Reject empty strings and split an item before it exceeds the model limit.
Encoding	`float`	The pgvector column stores numeric vectors.
Dimensions	`1024`	Check every response and keep the SQL column at the same width.
Response mapping	`data[index].embedding`	Require the response count to match and every expected index to appear exactly once.
Distance	Cosine via `<=>`	Use `1 - distance` only when displaying cosine similarity.

The refetched CometAPI reference states an 8,191-token maximum for each text-embedding-3-* input. Do not use that ceiling as a chunking target; keep individual FAQ entries focused and reject or split oversized content before the request.

The pgvector vector type supports up to 2,000 dimensions, so the tutorial’s 1,024-dimension choice fits that type. If you change the model or requested width, create a migration and rebuild the affected vectors. Do not merely change the Python constant while leaving the existing column and rows in place.

For a small FAQ table, exact search is the clearest baseline. When measured latency requires an approximate index, pgvector supports a cosine HNSW index:

CREATE INDEX faq_entries_embedding_cosine_hnsw
ON faq_entries
USING hnsw (embedding vector_cosine_ops);

Approximate search trades some recall for speed. Compare it against the exact baseline with labeled questions before enabling it in production.

Failure modes

The extension cannot be created. The server may not have pgvector’s supporting files, or the database role may lack the necessary privileges. Check pg_available_extensions, involve the database administrator, and do not repeatedly run the migration under a broader role than required.

The embedding request is rejected. First verify the configured base, selected model ID, nonempty input, and requested dimensions against the current CometAPI contract. Keep the prior searchable dataset active while the request problem is resolved.

An FAQ exceeds the input limit. Reject it before ingestion or split it into independently useful entries. Preserve a stable source identifier if one logical article becomes several searchable rows.

The response count or indexes do not match the batch. A missing, duplicate, unexpected, or surplus index causes the complete response to be rejected before the index-to-vector mapping is built. Do not guess which vector belongs to which FAQ. Retain the old rows and log only counts and contract metadata.

The vector width differs from the table. Stop before the database write. Verify the model and dimensions settings, then rebuild into a table or column with one consistent contract. The Python check gives a clearer error than relying solely on the database cast.

A model change mixes old and new rows. The embedding_model filter prevents this example from comparing rows labeled with another model, but it does not rebuild them. Treat a model or dimension change as a versioned reindex operation and switch traffic only after the new set is complete.

Unsafe SQL composition reaches the query. Never interpolate the question, limit, or serialized vector into the SQL string. Psycopg parameters keep values separate from query structure. Dynamic table or column identifiers require Psycopg’s SQL composition tools, not value placeholders.

Approximate indexing lowers result quality. HNSW and IVFFlat can improve speed by accepting a recall tradeoff. Keep an exact-search evaluation set, compare rankings, and tune or remove the approximate index if expected answers disappear.

The highest-ranked FAQ is still irrelevant. Ranking always produces a first row when rows exist. Build a labeled evaluation set, choose a threshold from observed application data, and provide a no-match response when the result is not trustworthy.

FAQ

Why request 1,024 dimensions?

It creates an explicit, pgvector-compatible teaching contract and reduces the size of each stored vector compared with a larger default. CometAPI documents the dimensions option for text-embedding-3-* models. Validate search quality on your own FAQ set before selecting a production width.

Can I change embedding models later?

Yes, but rebuild the corpus as a new version. Keep the old table or version active until the replacement batch passes count, dimension, and ranking checks. A model change should not silently overwrite only some rows.

Do I need HNSW for a few hundred FAQs?

Usually the right first step is to measure exact search. pgvector performs exact nearest-neighbor search by default. Add HNSW or IVFFlat only when measured latency or scale justifies accepting and evaluating approximate results.

Why embed both the question and answer?

Combining both fields lets the stored representation include the wording users may ask and the information the answer contains. Keep the composition stable. If you later embed only questions or add metadata, rebuild every row so the corpus follows one rule.

Should the application always return the first match?

No. A nearest-neighbor query identifies the closest stored row, not whether that row is good enough for the user. Evaluate labeled queries, set a no-match policy, and avoid displaying an unrelated answer with false confidence.

Why serialize the vector and still use a parameter?

The JSON serialization creates pgvector’s bracketed numeric input representation. %s::vector then sends that representation as a bound value and asks PostgreSQL to parse it as a vector. Serialization is not a reason to concatenate the value into SQL.

Reader next step

Create the table in a staging database, index the four sample FAQs, and run at least five paraphrased questions with an expected first answer. Confirm the happy-path logs, run the local dimension error test, and verify that a failed refresh leaves the previous rows available. Then replace the sample entries with a small reviewed FAQ set, document the chosen model and width, and measure exact-search rankings before considering an approximate index.