> ## Documentation Index
> Fetch the complete documentation index at: https://launchpad.datalumina.com/llms.txt
> Use this file to discover all available pages before exploring further.

# PGVector RAG

> Build Retrieval Augmented Generation workflows using PostgreSQL with the pgvector extension

<Info>
  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/`.
</Info>

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

```python theme={null}
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

```python theme={null}
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

<Steps>
  <Step title="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`.
  </Step>

  <Step title="Apply migrations">
    From `app/launchpad/`, run `./migrate.sh` so the pgvector extension and tables exist.
  </Step>

  <Step title="Insert embeddings">
    Use `PgvectorRAGService` directly to seed your collection:

    ```python theme={null}
    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()
    ```
  </Step>

  <Step title="Run the workflow">
    Use the dedicated playground script to exercise the whole pipeline against a sample query:

    ```bash theme={null}
    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.
  </Step>
</Steps>

## 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`.