Skip to main content

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.

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.
FilePurpose
app/launchpad/workflows/examples/pgvector_rag/schema.pyRagExampleEventSchema — accepts a single query: str.
app/launchpad/workflows/examples/pgvector_rag/services.pyPgvectorRAGService — wraps vecs + OpenAI embeddings; exposes get_embedding, get_collection, upsert, parse_results, count_tokens, disconnect.
app/launchpad/workflows/examples/pgvector_rag/nodes/retrieval_node.pyRetrievalNode — embeds the query, performs a cosine-distance lookup, and writes the hits to the context.
app/launchpad/workflows/examples/pgvector_rag/nodes/generation_node.pyGenerationNode — grounds an agent response in the retrieved chunks and returns answer, sources, confidence.
app/launchpad/workflows/examples/pgvector_rag/workflow.pyRagExampleWorkflow 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

1

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.
2

Apply migrations

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

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()
4

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.