Last reviewed: 2026-09-01

Direct answer

A Haystack CometAPI RAG pipeline can keep document retrieval local and send the final grounded prompt to CometAPI for generation. Haystack’s current retrieval-augmentation tutorial demonstrates the local path: embed documents and questions with the same SentenceTransformers model, retrieve from an InMemoryDocumentStore, render the retrieved documents with ChatPromptBuilder, and pass the resulting messages to a chat generator. Haystack’s OpenAIChatGenerator accepts a custom api_base_url, while the CometAPI quickstart documents an OpenAI-compatible base URL, environment-backed credentials, and the need to select a current model ID.

The source-supported flow is:

  1. Convert a small set of local Document objects into vectors with SentenceTransformersDocumentEmbedder.
  2. Store those documents and vectors in InMemoryDocumentStore.
  3. Embed each question with the same model and retrieve the closest documents.
  4. Render the retrieved content into a constrained question-answering prompt.
  5. Send that prompt through OpenAIChatGenerator, configured with the CometAPI base URL and a current Chat Completions model ID.

This article deliberately keeps embeddings local. The CometAPI embeddings reference confirms that CometAPI offers a remote embeddings endpoint, including batch input and selectable embedding models. The refetched Haystack evidence, however, does not establish the current configuration contract for pointing Haystack’s embedding components at that endpoint. Keeping the embedding step local avoids guessing while still producing a complete CometAPI-backed RAG pipeline.

Who this is for

This tutorial is for Python developers who want a small, inspectable RAG example before adopting a persistent vector database. It is especially useful when you need to see where retrieval ends and generation begins, test grounding with controlled documents, and collect safe operational evidence around the live model call.

The example is not a production storage design. Haystack describes InMemoryDocumentStore as a simple choice for smaller projects and debugging and warns that it does not scale well to larger document collections. Use it to prove the contract, then select a persistent document-store integration for a long-lived or larger corpus.

Key takeaways

  • The document embedder and query embedder must use the same embedding model. Changing only one side invalidates similarity comparisons and requires re-indexing.
  • CometAPI configuration belongs at the generator boundary: a process-local credential, the documented base URL, and a current model ID compatible with Chat Completions.
  • Retrieval and generation need separate acceptance checks. A successful model response does not prove that the right document was retrieved.
  • The prompt should explicitly require an insufficient-context answer rather than inviting the model to fill gaps.
  • Operational logs should contain identifiers, timings, counts, model names, finish state, and usage metadata, but not credentials, prompts, questions, retrieved text, headers, or raw response bodies.

Sources checked

  • The CometAPI quickstart supplies the current OpenAI-compatible setup pattern, credential boundary, base URL, and model-selection requirement.
  • The CometAPI Create embeddings reference defines the remote embedding route, input forms, example embedding model, response vectors, and usage metadata.
  • The Haystack OpenAIChatGenerator documentation documents api_base_url, its default environment-variable lookup, the replies output, reply metadata, and its usual position after ChatPromptBuilder.
  • The Haystack retrieval-augmentation tutorial demonstrates InMemoryDocumentStore, paired SentenceTransformers embedders, InMemoryEmbeddingRetriever, ChatPromptBuilder, and chat-generator connections.

Contract details to verify

Install the source-backed components

Create an isolated environment and install the two packages used by this local retrieval design:

python -m pip install haystack-ai sentence-transformers-haystack

Do not place the CometAPI credential in the script, shell history, screenshots, or repository. The credential storage guide covers the repository boundary in more detail.

The CometAPI quickstart names a CometAPI-specific environment variable in its generic SDK example. OpenAIChatGenerator reads OPENAI_API_KEY by default. For this Haystack process, enter the CometAPI credential interactively under the variable Haystack reads, without writing a value into the script:

read -rsp 'CometAPI credential for this Haystack process: ' OPENAI_API_KEY
printf '\n'
export OPENAI_API_KEY
read -rp 'CometAPI v1 base URL from the official quickstart: ' COMETAPI_BASE_URL
export COMETAPI_BASE_URL
read -rp 'Current Chat Completions model ID: ' COMETAPI_CHAT_MODEL
export COMETAPI_CHAT_MODEL

Enter the base URL exactly as shown in the official quickstart. Select the model from the current CometAPI catalog and confirm that its documented API family is Chat Completions. Do not assume that every model ID works on every text endpoint.

Build the indexing and query pipeline

Save the following as rag_demo.py. The three documents are deliberately small, synthetic policy notes so the expected retrieval result is easy to inspect.

import json
import os
import sys
import time
import uuid

from haystack import Document, Pipeline
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.dataclasses import ChatMessage
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack_integrations.components.embedders.sentence_transformers import (
    SentenceTransformersDocumentEmbedder,
    SentenceTransformersTextEmbedder,
)

REQUIRED_ENV = (
    'OPENAI_API_KEY',
    'COMETAPI_BASE_URL',
    'COMETAPI_CHAT_MODEL',
)
missing = [name for name in REQUIRED_ENV if not os.environ.get(name)]
if missing:
    raise RuntimeError(
        'Missing required environment variables: ' + ', '.join(missing)
    )

EMBEDDING_MODEL = 'sentence-transformers/all-MiniLM-L6-v2'
RETRIEVAL_LIMIT = 2

documents = [
    Document(
        content='Diagnostic application logs are retained for 14 days, then deleted.',
        meta={'source_id': 'policy-logging'},
    ),
    Document(
        content='Refund requests are reviewed within five business days.',
        meta={'source_id': 'policy-refunds'},
    ),
    Document(
        content='Standard support coverage runs Monday through Friday.',
        meta={'source_id': 'policy-support'},
    ),
]

document_store = InMemoryDocumentStore()
document_embedder = SentenceTransformersDocumentEmbedder(
    model=EMBEDDING_MODEL
)
documents_with_embeddings = document_embedder.run(documents)['documents']
document_store.write_documents(documents_with_embeddings)

query_embedder = SentenceTransformersTextEmbedder(model=EMBEDDING_MODEL)
retriever = InMemoryEmbeddingRetriever(document_store)

template = [
    ChatMessage.from_user(
        '''Use only the supplied context to answer the question.
If the context does not contain the answer, say that the context is insufficient.
Cite the source_id values used.

Context:
{% for document in documents %}
[{{ document.meta.source_id }}] {{ document.content }}
{% endfor %}

Question: {{ question }}
Answer:'''
    )
]
prompt_builder = ChatPromptBuilder(
    template=template,
    required_variables='*',
)

generator = OpenAIChatGenerator(
    api_base_url=os.environ['COMETAPI_BASE_URL'],
    model=os.environ['COMETAPI_CHAT_MODEL'],
)

rag = Pipeline()
rag.add_component('query_embedder', query_embedder)
rag.add_component('retriever', retriever)
rag.add_component('prompt_builder', prompt_builder)
rag.add_component('generator', generator)
rag.connect('query_embedder.embedding', 'retriever.query_embedding')
rag.connect('retriever.documents', 'prompt_builder.documents')
rag.connect('prompt_builder.prompt', 'generator.messages')


def ask(question):
    request_id = str(uuid.uuid4())
    started = time.perf_counter()

    try:
        result = rag.run(
            {
                'query_embedder': {'text': question},
                'retriever': {'top_k': RETRIEVAL_LIMIT},
                'prompt_builder': {'question': question},
            }
        )
        replies = result['generator']['replies']
        if not replies:
            raise RuntimeError('Generator returned no replies')

        reply = replies[0]
        metadata = reply.meta or {}
        usage = metadata.get('usage') or {}
        event = {
            'event': 'rag_query_complete',
            'request_id': request_id,
            'embedding_model': EMBEDDING_MODEL,
            'chat_model': os.environ['COMETAPI_CHAT_MODEL'],
            'indexed_document_count': len(documents),
            'retrieval_limit': RETRIEVAL_LIMIT,
            'elapsed_ms': round((time.perf_counter() - started) * 1000),
            'finish_reason': metadata.get('finish_reason'),
            'prompt_tokens': usage.get('prompt_tokens'),
            'completion_tokens': usage.get('completion_tokens'),
        }
        print(json.dumps(event, sort_keys=True), file=sys.stderr)
        return reply.text
    except Exception as exc:
        event = {
            'event': 'rag_query_failed',
            'request_id': request_id,
            'chat_model': os.environ['COMETAPI_CHAT_MODEL'],
            'elapsed_ms': round((time.perf_counter() - started) * 1000),
            'error_type': type(exc).__name__,
        }
        status = getattr(exc, 'status_code', None)
        if isinstance(status, int):
            event['http_status'] = status
        print(json.dumps(event, sort_keys=True), file=sys.stderr)
        raise SystemExit(1) from None


if __name__ == '__main__':
    question = (
        sys.argv[1]
        if len(sys.argv) > 1
        else 'How long are diagnostic logs retained?'
    )
    print(ask(question))

The log event intentionally omits the question, rendered prompt, retrieved document text, credential, request headers, and exception message. If an upstream error or local exception exposes sensitive material, replace that material with [REDACTED] before it reaches storage. After writing the allowlisted failure event, the script exits with a numeric status and suppresses exception chaining, so Python does not print the original exception message or traceback. Logging only the exception type and an available numeric status keeps the error path useful without copying an unreviewed response body.

Run the happy path

Use a question answered directly by one demo document:

python rag_demo.py 'How long are diagnostic logs retained?'

Accept the run only if it produces a nonempty answer, identifies policy-logging, and states the 14-day retention period without adding unsupported policy details. The sanitized event should name the embedding and chat models, show three indexed documents and a retrieval limit of two, and contain timing and any available usage metadata. A missing usage field is not automatically a failure because the generator documentation shows that metadata can vary, including during streaming. The answer and cited source remain the primary functional evidence.

Exercise two error paths

First, verify the local configuration guard without making a live model request:

unset COMETAPI_CHAT_MODEL
python rag_demo.py 'How long are diagnostic logs retained?'

The script should stop with a missing-environment-variable error before constructing the generator. Restore the model setting through the interactive setup before continuing.

Second, test the grounding guard with a question that the sample documents do not answer:

python rag_demo.py 'What is the holiday support schedule?'

The acceptable result says the supplied context is insufficient. If the response invents a holiday schedule, treat that as a prompt or model-behavior failure even though the API call itself succeeded. This distinction is important: transport success, retrieval relevance, and grounded generation are three separate checks.

For a live query failure, the handler writes one allowlisted rag_query_failed event and exits with status 1. It does not re-emit the provider exception after that event, so terminal output remains limited to the safe JSON record.

Failure modes

  • Missing runtime configuration: The preflight guard reports which environment variable is absent. Restore it interactively; do not add a fallback credential or model value to the source file.
  • Incorrect base URL: A malformed or stale base URL prevents the generator from reaching the documented OpenAI-compatible service. Recopy the value from the current CometAPI quickstart rather than guessing suffixes.
  • Model and endpoint mismatch: A model ID may be unavailable, renamed, or intended for another API family. Confirm the current ID and its Chat Completions compatibility before changing retry behavior. The model validation walkthrough provides a repeatable catalog check.
  • Authentication rejection: Confirm that the process-local credential exists and is active. Never print the value, authorization material, request headers, or a raw exception body while diagnosing the rejection.
  • Rate or concurrency rejection: Preserve the request ID, model, elapsed time, error class, and numeric status when available. The handler exits after its allowlisted event instead of exposing the provider exception. Consult the current provider guidance before adding bounded retry behavior; do not retry every failure indiscriminately.
  • Embedding model drift: If indexing uses one embedding model and querying uses another, rebuild the document embeddings with the same model used for questions. The Haystack tutorial explicitly requires this pairing.
  • Weak or empty retrieval: Reduce the corpus to a known fixture, ask a question with a literal answer, and verify the cited source_id. If that passes, revisit document splitting and retrieval settings before changing the generator.
  • In-memory data loss or scale pressure: InMemoryDocumentStore is intentionally temporary and suited to small projects and debugging. Choose a persistent Haystack document-store integration before relying on the index across restarts or expanding it substantially.
  • Unexpected generator output: The documented output is a replies list. Keep the explicit empty-list guard and review component release notes before changing result parsing.
  • Unsafe diagnostic logging: Do not solve an incident by logging prompts, retrieved passages, credentials, headers, complete response bodies, or an unfiltered traceback. Add allowlisted fields instead of trying to remove sensitive fields after collection.

FAQ

Does this tutorial send embeddings to CometAPI?

No. It uses the local SentenceTransformers document and text embedders shown in the refetched Haystack tutorial, then uses CometAPI for the final chat generation. CometAPI does document a remote embeddings endpoint, but the supplied Haystack sources do not establish the matching remote-embedder configuration contract. That integration should be added only after checking the current Haystack component documentation for both document and query embedders.

Why use OpenAIChatGenerator for CometAPI?

The two refetched contracts line up: Haystack documents a configurable api_base_url on OpenAIChatGenerator, and CometAPI documents an OpenAI-compatible base URL and credential pattern. The generator reads its documented default environment variable, while the model still needs to be a current CometAPI model that supports the Chat Completions family.

Can I change the local embedding model?

Yes, but change it in both SentenceTransformersDocumentEmbedder and SentenceTransformersTextEmbedder, then rebuild every stored document embedding. Do not reuse vectors produced by a different model.

Is the in-memory store enough for production?

Not for a large or durable corpus. The Haystack tutorial positions it as the simplest option for small projects and debugging and notes that it does not scale well. The pipeline boundary remains useful when you later replace the store and retriever.

Which fields are safe to log?

Use an allowlist such as event name, generated request ID, embedding model, chat model, indexed-document count, retrieval limit, elapsed milliseconds, finish reason, token counts, exception class, and numeric status. Do not log the user’s question, retrieved content, rendered messages, credential, authorization material, headers, raw error body, or exception traceback.

How should I decide whether a run passed?

Check three layers independently: the request completed, the relevant fixture was retrieved, and the answer stayed within that fixture. Also run an out-of-scope question and require an insufficient-context response. A polished answer is not evidence of correct retrieval by itself.

Reader next step

Run the happy path, the missing-configuration check, and the out-of-scope grounding check before adding your own documents. Then validate the current CometAPI model ID and record the allowlisted results using the local smoke-test evidence guide . When the in-memory fixture is stable, compare the storage boundary with the CometAPI embeddings and pgvector tutorial before choosing a persistent retrieval backend.