PGVector RAG - GenAI Launchpad

Retrieval Augmented Generation (RAG) grounds LLM responses in documents you control, rather than relying only on what the model learned during training. This example ships as RagExampleWorkflow in app/launchpad/workflows/examples/pgvector_rag/.

Instead of a dedicated vector database, you reuse the PostgreSQL instance the Launchpad already runs with the pgvector extension enabled.

Why pgvector?

Simple setup — no additional vector database; the default PostgreSQL service already has pgvector available.
Single source of truth — vectors live next to your relational data, so retrieval queries can join against business tables.
Cost effective — no separate vector database to operate or pay for.

Components

The workflow is a two-node pipeline: RetrievalNode → GenerationNode.

File	Purpose
`app/launchpad/workflows/examples/pgvector_rag/schema.py`	`RagExampleEventSchema` — accepts a single `query: str`.
`app/launchpad/workflows/examples/pgvector_rag/services.py`	`PgvectorRAGService` — wraps `vecs` + OpenAI embeddings; exposes `get_embedding`, `get_collection`, `upsert`, `parse_results`, `count_tokens`, `disconnect`.
`app/launchpad/workflows/examples/pgvector_rag/nodes/retrieval_node.py`	`RetrievalNode` — embeds the query, performs a cosine-distance lookup, and writes the hits to the context.
`app/launchpad/workflows/examples/pgvector_rag/nodes/generation_node.py`	`GenerationNode` — grounds an agent response in the retrieved chunks and returns `answer`, `sources`, `confidence`.
`app/launchpad/workflows/examples/pgvector_rag/workflow.py`	`RagExampleWorkflow` registered as `WorkflowRegistry.PGVECTOR_RAG`.

Retrieval

class RetrievalNode(Node):
    class OutputType(Node.OutputType):
        results: RetrievalResults

    async def process(self, task_context: TaskContext) -> TaskContext:
        rag_service = PgvectorRAGService()
        collection = rag_service.get_collection()
        event: RagExampleEventSchema = task_context.event
        embedding = rag_service.get_embedding(event.query)

        results = collection.query(
            data=embedding,
            limit=3,
            measure="cosine_distance",
            include_value=False,
            include_metadata=True,
        )
        self.save_output(self.OutputType(results=rag_service.parse_results(results)))
        rag_service.disconnect()
        return task_context

Generation

class GenerationNode(AgentNode):
    class DepsType(AgentNode.DepsType):
        context: RetrievalResults

    class OutputType(AgentNode.OutputType):
        answer: str
        sources: list[str]
        confidence: float = Field(ge=0, le=1)

    def get_agent_config(self) -> AgentConfig:
        return AgentConfig(
            instructions=(
                "You are a helpful assistant that answers questions using the "
                "retrieved documents."
            ),
            output_type=GenerationNode.OutputType,
            deps_type=GenerationNode.DepsType,
            model_provider=ModelProvider.OPENAI,
            model_name="gpt-5.4-mini",
        )

    async def process(self, task_context: TaskContext) -> TaskContext:
        retrieval_results: RetrievalNode.OutputType = self.get_output(RetrievalNode)
        deps = GenerationNode.DepsType(context=retrieval_results.results)

        @self.agent.instructions
        def add_rag_context(ctx: RunContext[GenerationNode.DepsType]) -> str:
            return (
                "Here are the documents I found for your query:\n"
                f"{ctx.deps.context.model_dump_json(indent=2)}"
            )

        result = await self.agent.run(user_prompt=task_context.event.query, deps=deps)
        self.save_output(result.output)
        return task_context

Running the example

Start the stack

From docker/, run ./start.sh to bring up the API, Celery worker, Redis, and PostgreSQL with pgvector. Supabase services and Caddy stay disabled unless you uncomment them in docker-compose.yml.

Apply migrations

From app/launchpad/, run ./migrate.sh so the pgvector extension and tables exist.

Insert embeddings

Use PgvectorRAGService directly to seed your collection:

from launchpad.workflows.examples.pgvector_rag.services import PgvectorRAGService

service = PgvectorRAGService()
text = "Vector search enables semantic retrieval."
embedding = service.get_embedding(text)
service.upsert([("doc-1", embedding, {"source": "handbook", "text": text})])
service.disconnect()

Run the workflow

Use the dedicated playground script to exercise the whole pipeline against a sample query:

uv run playground/pgvector_rag.py

The script loads app/launchpad/workflows/examples/pgvector_rag/request_examples/query.json, instantiates RagExampleWorkflow, and prints the generated answer with sources and confidence.

Customization

Swap embedding model — pass a different embedding_model to PgvectorRAGService (default text-embedding-3-small, 1536 dims).
Change retrieval — adjust limit, measure, or metadata filters on collection.query(...) in RetrievalNode.
Change generation model — update model_provider / model_name in GenerationNode.get_agent_config.

​Why pgvector?

​Components

​Retrieval

​Generation

​Running the example

​Customization

Why pgvector?

Components

Retrieval

Generation

Running the example

Customization