DEV Community: Hiroki Kameyama

Building a RAG System from Scratch — Wrap-up and What Comes Next

Hiroki Kameyama — Sat, 27 Jun 2026 22:21:00 +0000

In this final article, we'll recap what we built across the series, consolidate the design decisions, and point to where to go next.

What We Built

Starting from a blank Python project, we built a complete AI system step by step:

01_setup_db.py       pgvector table + extension
02_create_index.py   HNSW index (m=16, ef_construction=64)
03_ingest.py         Embed documents → store in pgvector
04_search.py         Cosine similarity search
05_rag.py            Full RAG pipeline

06_tool_basic.py     LLM decides whether to search
07_tool_multi.py     LLM routes between multiple tools
08_tool_agent.py     Multi-step agentic loop

09_agent_basic.py    ReAct pattern
10_agent_memory.py   Persistent memory across sessions
11_agent_planner.py  Plan → Execute → Evaluate

mcp_server/
  server.py          MCP server (stdio, Claude Desktop)
  server_http.py     MCP server (HTTP)
  server_render.py   MCP server (Render deployment)

12_mcp_agent.py      Agent via MCP (local)
13_mcp_http_agent.py Agent via MCP (cloud)

Design Decisions at a Glance

pgvector over a dedicated vector DB

pgvector integrates with existing PostgreSQL, supports SQL + vector in one query, and handles millions of documents comfortably. Start here and migrate only when you have evidence you need to.

768 dimensions

gemini-embedding-001 outputs 3072 dims by default, but pgvector's HNSW index has a 2000-dim hard limit. 768 dims stays well within bounds with negligible quality loss.

Asymmetric `task_type`

Use RETRIEVAL_DOCUMENT when storing, RETRIEVAL_QUERY when searching. The Gemini embedding model is trained to map queries toward documents, not to the same point. Using the same task type for both degrades retrieval accuracy.

HNSW over IVFFlat

HNSW requires no training data, delivers consistent recall at scale, and is faster at query time. IVFFlat is only worth considering under tight memory constraints.

Tool description is routing logic

The LLM selects tools based on their description field. Precise, distinguishing descriptions produce correct tool selection. Vague descriptions produce random behavior.

The conversation history is the agent's memory

Each tool call and result gets appended to contents. The LLM reads the full history on every step — this is how multi-step reasoning works.

MCP makes tools infrastructure

MCP turns hardcoded functions into a standalone server. Claude Desktop, Gemini agents, and any future client can connect to the same server without duplicating tool definitions.

Render + Supabase for zero-cost cloud deployment

Render's free web service hosts the MCP server. Supabase's free tier hosts pgvector. The Connection Pooler (port 6543) is mandatory — Render doesn't support the IPv6 used by Supabase's standard port 5432.

The Architecture We Ended Up With

Local:
  Claude Desktop
      ↓ stdio
  mcp_server/server.py
      ↓ psycopg2
  pgvector (Docker)

Cloud:
  Python agent (13_mcp_http_agent.py)
      ↓ HTTPS
  Render (server_render.py)
      ↓ PostgreSQL + SSL (port 6543)
  Supabase (pgvector)
      ↓
  Gemini Embedding + LLM

What This Series Did Not Cover

This series focused on getting a production-ready RAG system off the ground. Several important topics are out of scope here:

Evaluation (Evals) — How do you know if your RAG is actually working? You need automated quality measurement: Context Recall, Answer Relevancy, and Faithfulness scoring.

Observability — When something goes wrong in production, how do you debug it? Tracing each step with a tool like Langfuse tells you exactly where latency or quality issues originate.

Security — How do you handle adversarial inputs? Prompt injection, jailbreaks, and PII leakage are real threats in any public-facing RAG system.

MLOps / LLMOps — How do you ship changes safely? Prompt versioning, CI/CD quality gates, and API cost tracking become essential when the system is in production.

Fine-tuning — When the base model doesn't behave the way you need, LoRA fine-tuning lets you adapt it to your domain with surprisingly little data and compute.

Multi-Agent Systems — When a single agent isn't enough, orchestrator-worker patterns distribute work across specialized agents.

Governance — The EU AI Act is now fully in force. Compliance for a chatbot system means AI disclosure notices, audit logging, and a documented risk assessment.

All of these are covered in Vol.2 of this series.

Vol.2: Production Operations Guide

The second series picks up where this one leaves off — taking a working RAG system and making it production-grade.

AI Production Operations Guide (Japanese — English Dev.to series coming soon)

Chapter	Topic
1	What "production" actually means
2	Evals — automated quality measurement
3	Observability with Langfuse v4
4	Security — guardrails and prompt injection defense
5	MLOps / LLMOps — CI/CD pipeline
6	Fine-tuning with LoRA
7	Multi-Agent: orchestrator-worker pattern
8	Governance — EU AI Act compliance
9	Wrap-up

Source Code

Everything built in this series is in one repository:

github.com/qameqame/pgvector-tutorial

The README covers setup, directory structure, and the reasoning behind each design decision.

Series Index

Introduction
RAG · Embedding · Vector DB Implementation
Design Decisions Explained
Tool Use — Autonomous Search
AI Agents — Memory and Planning
MCP — Reusable Tool Server
Cloud Deployment — Render × Supabase
Wrap-up and Next Steps (this article)

Thanks for following along. If you found this useful, the GitHub repo and Vol.2 are the best places to continue.

Building a RAG System from Scratch — Cloud Deployment with Render and Supabase

Hiroki Kameyama — Sat, 27 Jun 2026 22:20:09 +0000

In the previous article, we built an MCP server that any LLM client can connect to locally. In this article, we'll deploy it to the cloud — making it accessible from anywhere.

Before: localhost:8000 → pgvector (Docker)
After:  https://your-app.onrender.com/mcp → Supabase (pgvector)

Both services are free to start with no credit card required.

Architecture

Service	Role	Free tier
Render	Host the MCP HTTP server	Persistent web service (sleeps after 15 min)
Supabase	Managed PostgreSQL + pgvector	500MB, persistent

Step 1: Supabase Setup

Create a project

Go to supabase.com and sign up with GitHub
Create a new project — set a database password and choose the Tokyo region
Wait 2–3 minutes for provisioning

Enable pgvector and create the table

Open SQL Editor in the Supabase dashboard and run:

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE IF NOT EXISTS documents (
    id          SERIAL PRIMARY KEY,
    title       TEXT NOT NULL,
    body        TEXT NOT NULL,
    category    TEXT,
    created_at  TIMESTAMP DEFAULT NOW(),
    embedding   vector(768)
);

CREATE INDEX IF NOT EXISTS docs_embedding_idx
ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

CREATE INDEX ON documents (category);

Get the connection string

Click the Connect button at the top of the dashboard (not Settings → Database — the UI has changed).

Select the Connection pooling tab and copy the Transaction mode URI. It looks like:

postgresql://postgres.xxxx:password@aws-0-ap-northeast-1.pooler.supabase.com:6543/postgres

Why port 6543? The standard port 5432 uses IPv6, which Render doesn't support. The Connection Pooler (port 6543) uses IPv4 and is the correct choice for cloud-to-cloud connections.

Step 2: Migrate Local Data to Supabase

Add the Supabase URL to your .env:

DATABASE_URL=postgresql://postgres.xxxx:password@aws-0-ap-northeast-1.pooler.supabase.com:6543/postgres

Then run the migration:

# migrate_to_supabase.py
import psycopg2
from dotenv import load_dotenv
import os

load_dotenv()

local_conn = psycopg2.connect(
    host=os.getenv("DB_HOST"), port=os.getenv("DB_PORT"),
    dbname=os.getenv("DB_NAME"), user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD"),
)
local_cur = local_conn.cursor()

supa_conn = psycopg2.connect(os.getenv("DATABASE_URL"), sslmode="require")
supa_cur = supa_conn.cursor()

local_cur.execute("SELECT title, body, category, embedding FROM documents;")
rows = local_cur.fetchall()
print(f"Migrating {len(rows)} documents...")

for row in rows:
    title, body, category, embedding = row
    supa_cur.execute("""
        INSERT INTO documents (title, body, category, embedding)
        VALUES (%s, %s, %s, %s) ON CONFLICT DO NOTHING;
    """, (title, body, category, embedding))

supa_conn.commit()

supa_cur.execute("SELECT COUNT(*) FROM documents;")
count = supa_cur.fetchone()[0]
print(f"Done. Documents in Supabase: {count}")

local_conn.close()
supa_conn.close()

python migrate_to_supabase.py
# Migrating 5 documents...
# Done. Documents in Supabase: 5

Step 3: Render-Ready MCP Server

Create mcp_server/server_render.py. The only differences from server.py:

DB connection reads from DATABASE_URL env var
Port reads from PORT env var (Render sets this automatically)
Transport is streamable-http instead of stdio

# mcp_server/server_render.py
import psycopg2
from google import genai
from google.genai import types as genai_types
from fastmcp import FastMCP
from dotenv import load_dotenv
import os

load_dotenv()

mcp = FastMCP(
    name="pgvector-search",
    instructions="Document search server using pgvector.",
)

gemini_client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))

# Use DATABASE_URL if available (Render/Supabase), fall back to individual vars
DATABASE_URL = os.getenv("DATABASE_URL")
if DATABASE_URL:
    conn = psycopg2.connect(DATABASE_URL, sslmode="require")
else:
    conn = psycopg2.connect(
        host=os.getenv("DB_HOST", "localhost"),
        port=os.getenv("DB_PORT", "5432"),
        dbname=os.getenv("DB_NAME", "vectordb"),
        user=os.getenv("DB_USER", "postgres"),
        password=os.getenv("DB_PASSWORD", "password"),
    )

cur = conn.cursor()


def get_embedding(text: str) -> list[float]:
    result = gemini_client.models.embed_content(
        model="gemini-embedding-001",
        contents=text,
        config=genai_types.EmbedContentConfig(
            task_type="RETRIEVAL_QUERY",
            output_dimensionality=768,
        ),
    )
    return result.embeddings[0].values


@mcp.tool
def search_documents(query: str, top_k: int = 3) -> list[dict]:
    """Search all document categories for a given query."""
    q = get_embedding(query)
    cur.execute("""
        SELECT title, body, category,
               1 - (embedding <=> %s::vector) AS similarity
        FROM documents ORDER BY embedding <=> %s::vector LIMIT %s;
    """, (q, q, top_k))
    return [
        {"title": r[0], "body": r[1], "category": r[2], "similarity": round(r[3], 4)}
        for r in cur.fetchall()
    ]


@mcp.tool
def search_by_category(query: str, category: str, top_k: int = 3) -> list[dict]:
    """Search within a specific category (ML, Python, or Cloud)."""
    q = get_embedding(query)
    cur.execute("""
        SELECT title, body, category,
               1 - (embedding <=> %s::vector) AS similarity
        FROM documents WHERE category = %s
        ORDER BY embedding <=> %s::vector LIMIT %s;
    """, (q, category, q, top_k))
    return [
        {"title": r[0], "body": r[1], "category": r[2], "similarity": round(r[3], 4)}
        for r in cur.fetchall()
    ]


@mcp.tool
def list_categories() -> list[dict]:
    """Return all available categories and document counts."""
    cur.execute("""
        SELECT category, COUNT(*) as count
        FROM documents GROUP BY category ORDER BY count DESC;
    """)
    return [{"category": r[0], "count": r[1]} for r in cur.fetchall()]


if __name__ == "__main__":
    port = int(os.getenv("PORT", 8000))  # Render sets PORT automatically
    mcp.run(
        transport="streamable-http",
        host="0.0.0.0",
        port=port,
    )

Push to GitHub:

git add .
git commit -m "feat: add Render deployment server"
git push origin main

Step 4: Deploy to Render

Go to render.com and sign up with GitHub (no credit card)
Click New → Web Service
Connect your repository
Set the following:

Field	Value
Name	`pgvector-mcp-server`
Runtime	Python 3
Build Command	`pip install -r requirements.txt`
Start Command	`python mcp_server/server_render.py`
Instance Type	Free

Under Environment, add:

Key	Value
`GEMINI_API_KEY`	`AIza...`
`DATABASE_URL`	The Connection Pooler URI from Supabase (port 6543)

Click Create Web Service

Verify the deployment

Once deployed, visit:

https://pgvector-mcp-server.onrender.com/mcp

You'll see:

{"jsonrpc":"2.0","id":"server-error","error":{"code":-32600,"message":"Not Acceptable: Client must accept text/event-stream"}}

This is correct — it means the server is running. The error appears because browsers aren't MCP clients. Your agent will connect without issues.

First request is slow: Render's free tier sleeps after 15 minutes of inactivity. The first request after sleep takes 30–60 seconds to wake up. Use UptimeRobot (free) to ping every 5 minutes and prevent sleep.

Step 5: Connect Your Agent

Update 13_mcp_http_agent.py with the Render URL:

# 13_mcp_http_agent.py
MCP_SERVER_URL = "https://pgvector-mcp-server.onrender.com/mcp"

async def run_agent(task: str):
    async with Client(MCP_SERVER_URL) as mcp_client:  # URL instead of file path
        mcp_tools = await mcp_client.list_tools()
        print(f"Loaded {len(mcp_tools)} tools from remote MCP server")
        # ... rest of the agentic loop is identical ...

python 13_mcp_http_agent.py
# Loaded 3 tools from remote MCP server
# [Step 1]
#   → list_categories({})
# [Step 2]
#   → search_by_category({'query': 'evaluation metrics', 'category': 'ML'})
# [Done in 3 steps]

The agent is now querying pgvector on Supabase through an MCP server running on Render — entirely in the cloud.

Troubleshooting

Error	Cause	Fix
`Network is unreachable` (IPv6)	Using port 5432	Use Connection Pooler URL (port 6543)
`SSL connection required`	Missing sslmode	Add `sslmode="require"`
`ModuleNotFoundError`	Missing package	Run `pip freeze > requirements.txt` and push
Slow first response	Render sleep	Use UptimeRobot to keep alive
`Not Found` at `/`	Wrong URL	Add `/mcp` to the URL

What We've Built

Local development:
  Python agent → stdio → mcp_server/server.py → Docker pgvector

Cloud deployment:
  Python agent → HTTPS → Render (server_render.py) → Supabase pgvector

The codebase is identical. The infrastructure changed around it.

In the final article, we'll wrap up the series with a summary of all design decisions and point to Vol.2 — where we cover Evals, Observability, Security, MLOps, Fine-tuning, Multi-Agent, and Governance.

Full source code: github.com/qameqame/pgvector-tutorial

Building a RAG System from Scratch — MCP: Exposing pgvector as a Reusable Tool Server

Hiroki Kameyama — Sat, 27 Jun 2026 22:17:47 +0000

In the previous article, we built AI Agents that autonomously search our pgvector database. One limitation remained: the tools were hardcoded inside our Python scripts. Only our code could use them.

MCP (Model Context Protocol) fixes this. It turns our search functions into a standalone server that any LLM client can connect to — Claude Desktop, Gemini agents, or any future client.

Tool Use vs MCP

Tool Use (what we built):
  Python script → hardcoded functions → Gemini API
  Reusable by: this script only

MCP Server (what we're building):
  Any LLM client → MCP protocol → our server → pgvector
  Reusable by: Claude Desktop, any agent, any language

The tools themselves don't change. What changes is where they live and how they're accessed.

MCP's Three Primitives

Primitive	Role	Our implementation
Tools	Functions the LLM can call	`search_documents`, `search_by_category`, `list_categories`
Resources	Data the LLM can read	`db://categories` (category list)
Prompts	Reusable prompt templates	`search_prompt(topic)`

Installing FastMCP

pip install fastmcp
pip freeze > requirements.txt

Step 1: MCP Server — `mcp_server/server.py`

# mcp_server/server.py
import psycopg2
from google import genai
from google.genai import types as genai_types
from fastmcp import FastMCP
from dotenv import load_dotenv
import os

load_dotenv()

mcp = FastMCP(
    name="pgvector-search",
    instructions="Document search server using pgvector. "
                 "Covers machine learning, Python, and cloud topics.",
)

gemini_client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))

conn = psycopg2.connect(
    host=os.getenv("DB_HOST"), port=os.getenv("DB_PORT"),
    dbname=os.getenv("DB_NAME"), user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD"),
)
cur = conn.cursor()


def get_embedding(text: str) -> list[float]:
    result = gemini_client.models.embed_content(
        model="gemini-embedding-001",
        contents=text,
        config=genai_types.EmbedContentConfig(
            task_type="RETRIEVAL_QUERY",
            output_dimensionality=768,
        ),
    )
    return result.embeddings[0].values


# ── Tools ─────────────────────────────────────────────────────
# The @mcp.tool decorator replaces FunctionDeclaration(...) entirely.
# Type hints + docstrings generate the schema automatically.

@mcp.tool
def search_documents(query: str, top_k: int = 3) -> list[dict]:
    """
    Search all document categories for a given query.
    Use when the category is unknown or the question spans multiple categories.

    Args:
        query: Search query
        top_k: Number of documents to retrieve (default: 3)
    """
    q = get_embedding(query)
    cur.execute("""
        SELECT title, body, category,
               1 - (embedding <=> %s::vector) AS similarity
        FROM documents ORDER BY embedding <=> %s::vector LIMIT %s;
    """, (q, q, top_k))
    return [
        {"title": r[0], "body": r[1], "category": r[2], "similarity": round(r[3], 4)}
        for r in cur.fetchall()
    ]


@mcp.tool
def search_by_category(query: str, category: str, top_k: int = 3) -> list[dict]:
    """
    Search within a specific category (ML, Python, or Cloud).
    Use when the category is explicitly mentioned in the question.

    Args:
        query: Search query
        category: Category name — ML, Python, or Cloud
        top_k: Number of documents to retrieve (default: 3)
    """
    q = get_embedding(query)
    cur.execute("""
        SELECT title, body, category,
               1 - (embedding <=> %s::vector) AS similarity
        FROM documents WHERE category = %s
        ORDER BY embedding <=> %s::vector LIMIT %s;
    """, (q, category, q, top_k))
    return [
        {"title": r[0], "body": r[1], "category": r[2], "similarity": round(r[3], 4)}
        for r in cur.fetchall()
    ]


@mcp.tool
def list_categories() -> list[dict]:
    """
    Return all available categories and their document counts.
    Use this first to understand what data is available.
    """
    cur.execute("""
        SELECT category, COUNT(*) as count
        FROM documents GROUP BY category ORDER BY count DESC;
    """)
    return [{"category": r[0], "count": r[1]} for r in cur.fetchall()]


# ── Resources ─────────────────────────────────────────────────
# Resources are read-only data the LLM can access directly.

@mcp.resource("db://categories")
def get_categories_resource() -> str:
    cur.execute("""
        SELECT category, COUNT(*) as count
        FROM documents GROUP BY category ORDER BY count DESC;
    """)
    lines = [f"- {r[0]}: {r[1]} documents" for r in cur.fetchall()]
    return "Available categories:\n" + "\n".join(lines)


# ── Prompts ───────────────────────────────────────────────────
# Reusable prompt templates.

@mcp.prompt
def search_prompt(topic: str) -> str:
    """Generate a structured search prompt for a given topic."""
    return f"""Research the following topic using the available tools:

Topic: {topic}

Steps:
1. Call list_categories to see what data is available
2. If a relevant category exists, use search_by_category
3. Otherwise use search_documents for a broad search
4. Synthesize the results into a clear answer"""


# ── Entry point ───────────────────────────────────────────────
if __name__ == "__main__":
    mcp.run()  # stdio mode — standard for Claude Desktop

mkdir mcp_server
touch mcp_server/__init__.py

Step 2: Test the Server — `mcp_server/client_test.py`

# mcp_server/client_test.py
import asyncio
from fastmcp import Client

async def test_server():
    async with Client("mcp_server/server.py") as client:

        # List available tools
        tools = await client.list_tools()
        print("=== Available tools ===")
        for tool in tools:
            print(f"  - {tool.name}: {tool.description[:50]}...")

        # List resources
        resources = await client.list_resources()
        print("\n=== Available resources ===")
        for r in resources:
            print(f"  - {r.uri}")

        # Call a tool
        print("\n=== list_categories ===")
        result = await client.call_tool("list_categories", {})
        print(result)

        print("\n=== search_documents ===")
        result = await client.call_tool(
            "search_documents",
            {"query": "ML evaluation metrics", "top_k": 2}
        )
        print(result)

        # Read a resource
        print("\n=== db://categories resource ===")
        content = await client.read_resource("db://categories")
        print(content)

if __name__ == "__main__":
    asyncio.run(test_server())

python mcp_server/client_test.py
# === Available tools ===
#   - search_documents: Search all document categories for a given...
#   - search_by_category: Search within a specific category...
#   - list_categories: Return all available categories...
#
# === list_categories ===
# [{'category': 'ML', 'count': 2}, {'category': 'Cloud', 'count': 2}, ...]

Step 3: Agent via MCP — `12_mcp_agent.py`

The biggest difference: tool definitions come from the server, not from hardcoded FunctionDeclaration objects.

# 12_mcp_agent.py
import asyncio
from google import genai
from google.genai import types
from fastmcp import Client
from dotenv import load_dotenv
import os
import time

load_dotenv()
gemini_client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))


async def run_agent(task: str):
    print(f"\nTask: {task}")
    print("=" * 60)

    async with Client("mcp_server/server.py") as mcp_client:

        # Fetch tool definitions from the server automatically
        mcp_tools = await mcp_client.list_tools()

        # Convert MCP tool definitions to Gemini format
        gemini_tools = types.Tool(
            function_declarations=[
                types.FunctionDeclaration(
                    name=tool.name,
                    description=tool.description or "",
                    parameters=types.Schema(
                        type=types.Type.OBJECT,
                        properties={
                            name: types.Schema(
                                type=types.Type.STRING
                                if schema.get("type") == "string"
                                else types.Type.INTEGER
                                if schema.get("type") == "integer"
                                else types.Type.STRING,
                                description=schema.get("description", ""),
                            )
                            for name, schema in
                            (tool.inputSchema.get("properties") or {}).items()
                        },
                        required=tool.inputSchema.get("required", []),
                    ),
                )
                for tool in mcp_tools
            ]
        )

        print(f"Loaded {len(mcp_tools)} tools from MCP server")

        contents = [types.Content(role="user", parts=[types.Part(text=task)])]

        for step in range(8):
            print(f"\n[Step {step + 1}]")

            for attempt in range(5):
                try:
                    response = gemini_client.models.generate_content(
                        model="gemini-2.5-flash",
                        contents=contents,
                        config=types.GenerateContentConfig(tools=[gemini_tools]),
                    )
                    break
                except Exception as e:
                    if ("503" in str(e) or "429" in str(e)) and attempt < 4:
                        time.sleep((attempt + 1) * 10)
                    else:
                        raise

            candidates = response.candidates
            if not candidates or not candidates[0].content.parts:
                break

            part = candidates[0].content.parts[0]

            if part.function_call:
                func_name = part.function_call.name
                func_args = dict(part.function_call.args)
                print(f"  → {func_name}({func_args})")

                # Execute via MCP server instead of calling locally
                result = await mcp_client.call_tool(func_name, func_args)
                print(f"  → {len(result) if isinstance(result, list) else result} results")

                contents.append(
                    types.Content(role="model", parts=[types.Part(function_call=part.function_call)])
                )
                contents.append(
                    types.Content(
                        role="user",
                        parts=[types.Part(
                            function_response=types.FunctionResponse(
                                name=func_name,
                                response={"result": result},
                            )
                        )]
                    )
                )
            else:
                text_parts = [
                    p.text for p in candidates[0].content.parts
                    if hasattr(p, 'text') and p.text
                ]
                print(f"\n[Done in {step + 1} steps]")
                return "\n".join(text_parts)

    return "Max steps reached."


async def main():
    result = await run_agent(
        "Check the available categories, then explain ML evaluation metrics in detail."
    )
    print(f"\nFinal answer:\n{result}")

if __name__ == "__main__":
    asyncio.run(main())

python 12_mcp_agent.py
# Loaded 3 tools from MCP server
# [Step 1]
#   → list_categories({})
# [Step 2]
#   → search_by_category({'query': 'evaluation metrics', 'category': 'ML'})
# [Done in 3 steps]

Step 4: Connect to Claude Desktop

If you have Claude Desktop installed, add this to ~/Library/Application Support/Claude/claude_desktop_config.json:

{
  "mcpServers": {
    "pgvector-search": {
      "command": "/path/to/your/project/.venv/bin/python",
      "args": ["/path/to/your/project/mcp_server/server.py"],
      "env": {
        "GEMINI_API_KEY": "AIza...",
        "DB_HOST": "localhost",
        "DB_PORT": "5432",
        "DB_NAME": "vectordb",
        "DB_USER": "postgres",
        "DB_PASSWORD": "password"
      }
    }
  }
}

Restart Claude Desktop. Now you can type "search the pgvector DB for ML evaluation metrics" directly in Claude's chat interface.

Note: Use the full path to your .venv/bin/python, not just python. Claude Desktop doesn't activate virtual environments automatically.

Note: Claude Desktop currently only supports stdio transport, not HTTP. Use server.py (not server_http.py) in the config.

Tool Use vs MCP: The Key Difference

# Tool Use — tools defined in code
tools = types.Tool(function_declarations=[
    types.FunctionDeclaration(name="search_documents", ...)  # handwritten
])
result = search_documents(query)  # called directly

# MCP — tools fetched from server
mcp_tools = await mcp_client.list_tools()    # fetched dynamically
result = await mcp_client.call_tool(name, args)  # executed on server

The tools are identical. The difference is where they live. MCP makes them a shared infrastructure component rather than a per-project implementation.

In the final article of this series, we'll deploy the MCP server to Render and the pgvector database to Supabase — making everything accessible from anywhere.

Full source code: github.com/qameqame/pgvector-tutorial

Building a RAG System from Scratch — AI Agents: Memory, Planning, and Multi-Step Reasoning

Hiroki Kameyama — Sat, 27 Jun 2026 22:15:27 +0000

In the previous article, we gave the LLM the ability to call tools autonomously. Now we'll build a proper AI Agent — one that remembers past interactions, plans ahead, and evaluates its own output.

What Makes Something an "Agent"?

An LLM with Tool Use is powerful but stateless — each run starts fresh. An Agent adds three capabilities:

Tool Use alone:
  question → LLM decides → tools → answer
  (no memory between runs)

Agent:
  question → plan → execute tools → evaluate → answer
  + remembers past interactions
  + adjusts strategy based on results

We'll implement three patterns in order of complexity:

File	Pattern	What it adds
`09_agent_basic.py`	ReAct	Reasoning before each action
`10_agent_memory.py`	Long-term memory	Persistence across sessions
`11_agent_planner.py`	Plan-Execute-Evaluate	Upfront planning + self-evaluation

Pattern 1: ReAct — `09_agent_basic.py`

ReAct (Reasoning + Acting) is the foundational agent pattern. Before taking each action, the agent explicitly reasons about what to do next.

# 09_agent_basic.py
import psycopg2
from google import genai
from google.genai import types
from dotenv import load_dotenv
import os
import time

load_dotenv()

client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
conn = psycopg2.connect(
    host=os.getenv("DB_HOST"), port=os.getenv("DB_PORT"),
    dbname=os.getenv("DB_NAME"), user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD"),
)
cur = conn.cursor()


def get_embedding(text: str) -> list[float]:
    result = client.models.embed_content(
        model="gemini-embedding-001",
        contents=text,
        config=types.EmbedContentConfig(
            task_type="RETRIEVAL_QUERY",
            output_dimensionality=768,
        ),
    )
    return result.embeddings[0].values


def search_documents(query: str, top_k: int = 3) -> list[dict]:
    q = get_embedding(query)
    cur.execute("""
        SELECT title, body, category,
               1 - (embedding <=> %s::vector) AS similarity
        FROM documents ORDER BY embedding <=> %s::vector LIMIT %s;
    """, (q, q, top_k))
    return [{"title": r[0], "body": r[1], "category": r[2], "similarity": round(r[3], 4)}
            for r in cur.fetchall()]


def search_by_category(query: str, category: str, top_k: int = 3) -> list[dict]:
    q = get_embedding(query)
    cur.execute("""
        SELECT title, body, category,
               1 - (embedding <=> %s::vector) AS similarity
        FROM documents WHERE category = %s
        ORDER BY embedding <=> %s::vector LIMIT %s;
    """, (q, category, q, top_k))
    return [{"title": r[0], "body": r[1], "category": r[2], "similarity": round(r[3], 4)}
            for r in cur.fetchall()]


def list_categories() -> list[dict]:
    cur.execute("""
        SELECT category, COUNT(*) as count
        FROM documents GROUP BY category ORDER BY count DESC;
    """)
    return [{"category": r[0], "count": r[1]} for r in cur.fetchall()]


tools = types.Tool(
    function_declarations=[
        types.FunctionDeclaration(
            name="search_documents",
            description="Search all categories. Use when the category is unknown.",
            parameters=types.Schema(
                type=types.Type.OBJECT,
                properties={
                    "query": types.Schema(type=types.Type.STRING),
                    "top_k": types.Schema(type=types.Type.INTEGER),
                },
                required=["query"],
            ),
        ),
        types.FunctionDeclaration(
            name="search_by_category",
            description="Search within a specific category (ML, Python, Cloud). "
                        "Use when the category is explicitly mentioned.",
            parameters=types.Schema(
                type=types.Type.OBJECT,
                properties={
                    "query": types.Schema(type=types.Type.STRING),
                    "category": types.Schema(type=types.Type.STRING),
                    "top_k": types.Schema(type=types.Type.INTEGER),
                },
                required=["query", "category"],
            ),
        ),
        types.FunctionDeclaration(
            name="list_categories",
            description="List all available categories. Use this first to understand what data is available.",
            parameters=types.Schema(type=types.Type.OBJECT, properties={}),
        ),
    ]
)


def dispatch(func_name: str, func_args: dict):
    if func_name == "search_documents":
        return search_documents(**func_args)
    elif func_name == "search_by_category":
        return search_by_category(**func_args)
    elif func_name == "list_categories":
        return list_categories()
    return {"error": f"Unknown: {func_name}"}


# ReAct system prompt — instructs the agent to reason before acting
REACT_SYSTEM = """You are a research agent with access to a document database.

For each step:
1. Think about what information you need
2. Choose the right tool to get it
3. Evaluate whether you have enough to answer

Always start by listing categories to understand what's available.
"""

def run_agent(task: str, max_steps: int = 8) -> str:
    print(f"\nTask: {task}")
    print("=" * 60)

    contents = [types.Content(role="user", parts=[types.Part(text=task)])]

    for step in range(max_steps):
        print(f"\n[Step {step + 1}]")

        for attempt in range(5):
            try:
                response = client.models.generate_content(
                    model="gemini-2.5-flash",
                    contents=contents,
                    config=types.GenerateContentConfig(
                        system_instruction=REACT_SYSTEM,
                        tools=[tools],
                    ),
                )
                break
            except Exception as e:
                if ("503" in str(e) or "429" in str(e)) and attempt < 4:
                    time.sleep((attempt + 1) * 10)
                else:
                    raise

        candidates = response.candidates
        if not candidates or not candidates[0].content.parts:
            break

        part = candidates[0].content.parts[0]

        if part.function_call:
            func_name = part.function_call.name
            func_args = dict(part.function_call.args)
            print(f"  → {func_name}({func_args})")

            result = dispatch(func_name, func_args)
            print(f"  → {len(result) if isinstance(result, list) else result} results")

            contents.append(
                types.Content(role="model", parts=[types.Part(function_call=part.function_call)])
            )
            contents.append(
                types.Content(
                    role="user",
                    parts=[types.Part(
                        function_response=types.FunctionResponse(
                            name=func_name,
                            response={"result": result},
                        )
                    )]
                )
            )
        else:
            text_parts = [p.text for p in candidates[0].content.parts if hasattr(p, 'text') and p.text]
            print(f"\n[Done in {step + 1} steps]")
            return "\n".join(text_parts)

    return "Max steps reached."


result = run_agent(
    "Check what categories are available, then give me a comprehensive "
    "explanation of ML evaluation metrics with Python code examples."
)
print(f"\nFinal answer:\n{result}")

python 09_agent_basic.py
# [Step 1]
#   → list_categories({})
#   → 3 results
# [Step 2]
#   → search_by_category({'query': 'evaluation metrics', 'category': 'ML'})
#   → 2 results
# [Step 3]
#   → search_by_category({'query': 'scikit-learn evaluation', 'category': 'ML'})
#   → 2 results
# [Done in 4 steps]

Pattern 2: Long-term Memory — `10_agent_memory.py`

The agent now persists what it learns about you across sessions.

# 10_agent_memory.py (key additions)
import json
from pathlib import Path

MEMORY_FILE = "agent_memory.json"

def load_memory() -> dict:
    if Path(MEMORY_FILE).exists():
        with open(MEMORY_FILE, "r") as f:
            return json.load(f)
    return {"facts": [], "preferences": [], "history": []}

def save_memory(memory: dict):
    with open(MEMORY_FILE, "w") as f:
        json.dump(memory, f, ensure_ascii=False, indent=2)

def update_memory(memory: dict, task: str, result: str):
    """Ask the LLM to extract facts worth remembering."""
    prompt = f"""Extract any facts about the user worth remembering from this interaction.
Return JSON only: {{"facts": ["..."], "preferences": ["..."]}}

Task: {task}
Result summary: {result[:200]}"""

    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
    )
    try:
        import re
        raw = response.text.strip()
        match = re.search(r'\{.*\}', raw, re.DOTALL)
        if match:
            extracted = json.loads(match.group())
            memory["facts"].extend(extracted.get("facts", []))
            memory["preferences"].extend(extracted.get("preferences", []))
            # Keep only the last 20 unique entries
            memory["facts"] = list(set(memory["facts"]))[-20:]
            memory["preferences"] = list(set(memory["preferences"]))[-20:]
    except Exception:
        pass

def run_agent_with_memory(task: str) -> str:
    memory = load_memory()

    # Inject memory into system prompt
    memory_context = ""
    if memory["facts"]:
        memory_context += f"\nKnown facts about the user: {', '.join(memory['facts'])}"
    if memory["preferences"]:
        memory_context += f"\nUser preferences: {', '.join(memory['preferences'])}"

    system = f"""You are a research agent with access to a document database.
{memory_context}
Use this context to personalize your responses."""

    # ... run agent loop (same as before) ...

    result = run_agent(task)  # simplified for brevity

    # Extract and save new memories
    update_memory(memory, task, result)
    memory["history"].append({"task": task, "summary": result[:100]})
    save_memory(memory)

    return result

After each session, agent_memory.json grows with facts and preferences. The next run picks them up automatically.

Pattern 3: Plan-Execute-Evaluate — `11_agent_planner.py`

The most structured pattern: the agent first makes an explicit plan, executes it step by step, then evaluates whether the goal was achieved.

# 11_agent_planner.py

PLANNER_SYSTEM = """You are a planning agent. Given a task, produce a step-by-step plan.
Return JSON only:
{
  "goal": "...",
  "steps": ["step 1", "step 2", ...],
  "success_criteria": "how to know the task is complete"
}"""

EXECUTOR_SYSTEM = """You are an execution agent. Follow the given plan step by step.
Use the available tools to complete each step.
When all steps are done, produce a final answer."""

EVALUATOR_SYSTEM = """You are an evaluation agent. Assess whether the answer meets the success criteria.
Return JSON only:
{
  "passed": true/false,
  "score": 0.0-1.0,
  "feedback": "what's missing or what to improve"
}"""


def plan(task: str) -> dict:
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=task,
        config=types.GenerateContentConfig(system_instruction=PLANNER_SYSTEM),
    )
    import json, re
    match = re.search(r'\{.*\}', response.text, re.DOTALL)
    return json.loads(match.group()) if match else {"goal": task, "steps": [task], "success_criteria": "answered"}


def execute(plan_data: dict) -> str:
    task_with_plan = f"""Task: {plan_data['goal']}

Plan:
{chr(10).join(f'{i+1}. {step}' for i, step in enumerate(plan_data['steps']))}

Execute each step using the available tools, then provide a final answer."""

    return run_agent(task_with_plan)  # the agentic loop from Pattern 1


def evaluate(task: str, answer: str, success_criteria: str) -> dict:
    prompt = f"""Task: {task}
Success criteria: {success_criteria}
Answer to evaluate: {answer}"""

    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
        config=types.GenerateContentConfig(system_instruction=EVALUATOR_SYSTEM),
    )
    import json, re
    match = re.search(r'\{.*\}', response.text, re.DOTALL)
    return json.loads(match.group()) if match else {"passed": True, "score": 0.5, "feedback": ""}


def run_planner_agent(task: str) -> str:
    print(f"\nTask: {task}")
    print("=" * 60)

    # Phase 1: Plan
    print("\n[Phase 1: Planning]")
    plan_data = plan(task)
    print(f"Goal: {plan_data['goal']}")
    for i, step in enumerate(plan_data["steps"]):
        print(f"  {i+1}. {step}")

    # Phase 2: Execute
    print("\n[Phase 2: Executing]")
    answer = execute(plan_data)

    # Phase 3: Evaluate
    print("\n[Phase 3: Evaluating]")
    eval_result = evaluate(task, answer, plan_data["success_criteria"])
    print(f"Passed: {eval_result['passed']} | Score: {eval_result['score']}")
    if eval_result.get("feedback"):
        print(f"Feedback: {eval_result['feedback']}")

    return answer


result = run_planner_agent(
    "Research ML evaluation metrics and Python tooling, "
    "then write a practical guide with code examples."
)
print(f"\nFinal answer:\n{result}")

When to Use Each Pattern

Pattern	Best for	Overhead
ReAct	General-purpose agents, most use cases	Low
Long-term memory	Personalized assistants, repeated users	Medium
Plan-Execute-Evaluate	Complex multi-step tasks, quality-critical outputs	High

Start with ReAct. Add memory when you need personalization. Add planning when task complexity demands it.

What We've Built

06: LLM decides whether to use a tool
07: LLM routes between multiple tools by reading descriptions
08: LLM uses tools across multiple steps (agentic loop)
09: LLM reasons before each action (ReAct)
10: LLM remembers across sessions (memory)
11: LLM plans, executes, and evaluates (structured pipeline)

In the next article, we'll take a step further and expose our search functions as an MCP (Model Context Protocol) server — making them reusable by any LLM, not just our Python scripts.

Full source code: github.com/qameqame/pgvector-tutorial

Building a RAG System from Scratch — Tool Use: Let the LLM Search Autonomously

Hiroki Kameyama — Sat, 27 Jun 2026 22:14:30 +0000

In the previous article, we examined the design decisions behind our RAG pipeline. Now we'll give the LLM the ability to call our search functions autonomously — this is Tool Use.

What Changes with Tool Use

In our RAG pipeline so far, we always called search() before generating an answer. The flow was hardcoded:

question → search() → generate_answer()

With Tool Use, the LLM decides whether to search, what to search for, and when it has enough information to answer:

question → LLM decides → search() if needed → LLM decides → answer

This matters when:

The question might already be answerable without retrieval
The right search query isn't the same as the user's question
Multiple searches with different queries improve the answer

How Tool Use Works

The LLM is given a list of available functions with their signatures and descriptions. It responds with either:

A function_call — "call this function with these arguments"
A text answer — "I have enough information to answer directly"

Your code executes the function call and sends the result back. The LLM then decides whether to call another function or produce a final answer.

You → LLM: "here are available tools + user question"
LLM → You: function_call { name: "search_documents", args: { query: "F1 score" } }
You → execute search_documents("F1 score") → results
You → LLM: function_result { ... }
LLM → You: "The F1 score is calculated as..."

Step 1: Basic Tool Use — `06_tool_basic.py`

# 06_tool_basic.py
import psycopg2
from google import genai
from google.genai import types
from dotenv import load_dotenv
import os

load_dotenv()

client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
conn = psycopg2.connect(
    host=os.getenv("DB_HOST"), port=os.getenv("DB_PORT"),
    dbname=os.getenv("DB_NAME"), user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD"),
)
cur = conn.cursor()


def get_embedding(text: str) -> list[float]:
    result = client.models.embed_content(
        model="gemini-embedding-001",
        contents=text,
        config=types.EmbedContentConfig(
            task_type="RETRIEVAL_QUERY",
            output_dimensionality=768,
        ),
    )
    return result.embeddings[0].values


def search_documents(query: str, top_k: int = 3) -> list[dict]:
    query_embedding = get_embedding(query)
    cur.execute("""
        SELECT title, body, category,
               1 - (embedding <=> %s::vector) AS similarity
        FROM documents
        ORDER BY embedding <=> %s::vector
        LIMIT %s;
    """, (query_embedding, query_embedding, top_k))
    rows = cur.fetchall()
    return [
        {"title": r[0], "body": r[1], "category": r[2], "similarity": round(r[3], 4)}
        for r in rows
    ]


# ── Tool definition ──────────────────────────────────────────
# Instead of calling search_documents() directly, we describe it to the LLM.
# The description is what the LLM uses to decide when to call it.
tools = types.Tool(
    function_declarations=[
        types.FunctionDeclaration(
            name="search_documents",
            description="Search documents in the vector DB for a given query. "
                        "Use this when you need information to answer the question.",
            parameters=types.Schema(
                type=types.Type.OBJECT,
                properties={
                    "query": types.Schema(
                        type=types.Type.STRING,
                        description="The search query",
                    ),
                    "top_k": types.Schema(
                        type=types.Type.INTEGER,
                        description="Number of documents to retrieve (default: 3)",
                    ),
                },
                required=["query"],
            ),
        ),
    ]
)


def run(question: str):
    print(f"Question: {question}\n")

    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=question,
        config=types.GenerateContentConfig(tools=[tools]),
    )

    part = response.candidates[0].content.parts[0]

    if part.function_call:
        # LLM decided to call a tool
        func_name = part.function_call.name
        func_args = dict(part.function_call.args)
        print(f"→ LLM called: {func_name}({func_args})")

        result = search_documents(**func_args)
        print(f"→ Retrieved {len(result)} documents")
        print(f"→ Top result: {result[0]['title']}")
    else:
        # LLM answered directly without searching
        print(f"→ LLM answered directly (no search needed)")
        print(part.text)


run("How do you calculate the F1 score?")
run("What is 2 + 2?")  # LLM should answer this without searching

python 06_tool_basic.py
# Question: How do you calculate the F1 score?
# → LLM called: search_documents({'query': 'F1 score calculation'})
# → Retrieved 3 documents
# → Top result: ML Model Evaluation Metrics
#
# Question: What is 2 + 2?
# → LLM answered directly (no search needed)
# → 4

The LLM correctly decides when to search and when not to.

Step 2: Multiple Tools — `07_tool_multi.py`

Now we give the LLM two tools: one for general search and one for category-filtered search. The LLM picks the right one based on the question.

# 07_tool_multi.py (key additions)

def search_by_category(query: str, category: str, top_k: int = 3) -> list[dict]:
    query_embedding = get_embedding(query)
    cur.execute("""
        SELECT title, body, category,
               1 - (embedding <=> %s::vector) AS similarity
        FROM documents
        WHERE category = %s
        ORDER BY embedding <=> %s::vector
        LIMIT %s;
    """, (query_embedding, category, query_embedding, top_k))
    rows = cur.fetchall()
    return [
        {"title": r[0], "body": r[1], "category": r[2], "similarity": round(r[3], 4)}
        for r in rows
    ]


tools = types.Tool(
    function_declarations=[
        types.FunctionDeclaration(
            name="search_documents",
            description="Search all categories when the category is unknown "
                        "or the question spans multiple categories.",
            parameters=types.Schema(
                type=types.Type.OBJECT,
                properties={
                    "query": types.Schema(type=types.Type.STRING),
                    "top_k": types.Schema(type=types.Type.INTEGER),
                },
                required=["query"],
            ),
        ),
        types.FunctionDeclaration(
            name="search_by_category",
            description="Search within a specific category (ML, Python, or Cloud). "
                        "Use this when the question clearly targets one category.",
            parameters=types.Schema(
                type=types.Type.OBJECT,
                properties={
                    "query": types.Schema(type=types.Type.STRING),
                    "category": types.Schema(
                        type=types.Type.STRING,
                        description="Category name: ML, Python, or Cloud",
                    ),
                    "top_k": types.Schema(type=types.Type.INTEGER),
                },
                required=["query", "category"],
            ),
        ),
    ]
)

The description is the routing logic. The LLM reads the description field to decide which tool to call. Write descriptions that clearly distinguish when to use each tool — this is prompt engineering for tool selection.

Step 3: Agentic Loop — `08_tool_agent.py`

The real power of Tool Use is the agentic loop: the LLM can call multiple tools in sequence, building up context before producing a final answer.

# 08_tool_agent.py

def dispatch(func_name: str, func_args: dict):
    """Route function calls to the right Python function."""
    if func_name == "search_documents":
        return search_documents(**func_args)
    elif func_name == "search_by_category":
        return search_by_category(**func_args)
    return {"error": f"Unknown function: {func_name}"}


def run_agent(task: str, max_steps: int = 8):
    print(f"\nTask: {task}")
    print("=" * 60)

    # Conversation history — this is what enables multi-step reasoning
    contents = [types.Content(role="user", parts=[types.Part(text=task)])]

    for step in range(max_steps):
        response = client.models.generate_content(
            model="gemini-2.5-flash",
            contents=contents,
            config=types.GenerateContentConfig(tools=[tools]),
        )

        part = response.candidates[0].content.parts[0]

        if part.function_call:
            func_name = part.function_call.name
            func_args = dict(part.function_call.args)
            print(f"[Step {step+1}] → {func_name}({func_args})")

            result = dispatch(func_name, func_args)

            # Append the tool call and result to conversation history
            contents.append(
                types.Content(role="model", parts=[types.Part(function_call=part.function_call)])
            )
            contents.append(
                types.Content(
                    role="user",
                    parts=[types.Part(
                        function_response=types.FunctionResponse(
                            name=func_name,
                            response={"result": result},
                        )
                    )]
                )
            )
        else:
            # LLM produced a final answer
            text_parts = [p.text for p in response.candidates[0].content.parts if p.text]
            print(f"\n[Done in {step+1} steps]")
            return "\n".join(text_parts)

    return "Max steps reached."


result = run_agent(
    "What evaluation metrics are available for ML models? "
    "Show me both the metric names and how to implement them in Python."
)
print(f"\nFinal answer:\n{result}")

python 08_tool_agent.py
# Task: What evaluation metrics are available for ML models?...
# [Step 1] → search_by_category({'query': 'ML evaluation metrics', 'category': 'ML'})
# [Step 2] → search_by_category({'query': 'scikit-learn model evaluation', 'category': 'ML'})
# [Done in 3 steps]
# Final answer: ML models are evaluated using...

The agent searched twice with different queries, gathered complementary information, then synthesized a comprehensive answer.

Key Patterns

The conversation history is the agent's memory. Each tool call and its result gets appended to contents. The LLM sees the full history on every step, which is how it knows what it has already retrieved and what it still needs.

dispatch() is the bridge. It maps function names (strings from the LLM) to actual Python functions. Keep it simple and exhaustive — every tool the LLM can call must have an entry here.

The description field does the routing. Spend time on tool descriptions. A vague description leads to random tool selection. A precise description ("use this when the category is explicitly mentioned") leads to correct routing almost every time.

What We've Added

Before Tool Use:
  hardcoded: question → search → answer

After Tool Use:
  autonomous: question → LLM decides → search (maybe) → LLM decides → answer

In the next article, we'll build a full AI Agent with memory, planning, and multiple tools working together.

Full source code: github.com/qameqame/pgvector-tutorial

Building a RAG System from Scratch — Design Decisions Explained

Hiroki Kameyama — Sat, 27 Jun 2026 22:08:41 +0000

In the previous article, we built a working RAG pipeline. Now let's step back and ask why we made each design decision — and what alternatives exist when your requirements change.

The Full Picture

Here's what we built:

Ingest phase
  Text → gemini-embedding-001 (RETRIEVAL_DOCUMENT, 768 dims)
       → pgvector (HNSW index, cosine similarity)

Query phase
  Question → gemini-embedding-001 (RETRIEVAL_QUERY, 768 dims)
           → pgvector search (top-k)
           → Gemini 2.5 Flash (answer generation)

Every element in this diagram was a choice. Let's examine each one.

Decision 1: pgvector over a Dedicated Vector DB

We used pgvector, a PostgreSQL extension, rather than a purpose-built vector database like Pinecone, Weaviate, or Qdrant.

Why pgvector works here:

Integrates with existing PostgreSQL infrastructure — no new service to operate
SQL and vector search in the same query: filter by category, join with other tables, all in one round-trip
Handles millions of documents comfortably with HNSW indexing

When to consider a dedicated vector DB:

Signal	Consider moving to
> 10M documents	Pinecone, Weaviate
Multi-modal search (text + image)	Weaviate, Qdrant
Managed cloud with SLA	Pinecone
On-premise, full control	Qdrant

For most enterprise RAG applications at typical document volumes, pgvector is the right starting point. Migrate when you hit actual limits, not anticipated ones.

Decision 2: 768 Dimensions instead of 3072

gemini-embedding-001 outputs 3072 dimensions by default. We set output_dimensionality=768.

The constraint: pgvector's HNSW index has a hard limit of 2000 dimensions.

Why not 2000? We chose 768 because:

It's a well-established embedding size used by BERT and many production systems
Cosine similarity quality degrades only slightly versus the full 3072 dims for typical retrieval tasks
Smaller vectors mean faster index builds and lower storage cost

Dimension vs. quality trade-off:

Dimensions	Index build	Storage	Retrieval quality
256	Fastest	Smallest	Noticeably lower
768	Fast	Small	Near full quality
1536	Moderate	Moderate	Full quality
3072	Slow	Largest	Full quality (no HNSW)

Decision 3: Asymmetric `task_type`

We used different task_type values for ingestion and querying:

# Ingestion
config=types.EmbedContentConfig(task_type="RETRIEVAL_DOCUMENT", ...)

# Query
config=types.EmbedContentConfig(task_type="RETRIEVAL_QUERY", ...)

Why this matters: Gemini's embedding model is trained with asymmetric objectives. A document and a query about the same topic are represented differently in embedding space — the model learns to map queries toward relevant documents, not to the same point. Using the same task type for both degrades retrieval accuracy.

This is analogous to how you'd phrase a document differently from a search query in natural language: "F1 Score is the harmonic mean of Precision and Recall" (document) vs. "how to calculate F1" (query).

Decision 4: HNSW over IVFFlat

pgvector supports two index types. We chose HNSW.

	HNSW	IVFFlat
Query speed	Fast	Moderate
Build time	Moderate	Fast
Memory	Higher	Lower
Accuracy at scale	Higher	Lower
Requires training data	No	Yes (needs `VACUUM` after inserts)

HNSW is the better default for production. IVFFlat is worth considering only when you have very tight memory constraints and can afford slower queries.

HNSW parameter guide:

WITH (
    m = 16,              -- max connections per node
    ef_construction = 64 -- search width during build
)

m: higher = better recall, more memory. Range: 4–64. Default 16 works for most cases.
ef_construction: higher = better index quality, slower build. Range: 16–512. Default 64 is a good production starting point.

Decision 5: Gemini 2.5 Flash for Generation

We used gemini-2.5-flash rather than the more capable gemini-opus models.

Reasoning:

Flash has sufficient quality for document-grounded Q&A — the retrieval step does the heavy lifting
Flash is faster and cheaper (or free-tier eligible during development)
The generation prompt is constrained: "answer based on these documents" limits hallucination regardless of model capability

When to upgrade the generation model:

Complex multi-step reasoning across many documents
Synthesis tasks requiring cross-document inference
When evaluation scores (Faithfulness, Relevancy) are consistently below threshold

When to upgrade the embedding model:

Low Context Recall — the right documents aren't being retrieved
Evaluation reveals semantic mismatch between queries and stored documents

The embedding model matters more for retrieval quality. The generation model matters more for answer quality. Optimize them independently.

The Scaling Path

This architecture scales predictably:

Phase 1 (now): pgvector local → works to ~1M docs
Phase 2:       pgvector + Supabase → managed PostgreSQL, easy scaling
Phase 3:       pgvector + read replicas → horizontal query scaling
Phase 4:       Dedicated vector DB → if you genuinely outgrow pgvector

Most teams never reach Phase 4. Start at Phase 1, move when you have evidence you need to.

Common Pitfalls

Chunking strategy matters more than model choice. If your documents are long (PDFs, reports), how you split them into chunks dramatically affects retrieval quality. A naive split at 512 tokens often breaks context mid-sentence. Consider semantic chunking or overlap.

Don't embed the question alone. For complex questions, consider HyDE (Hypothetical Document Embedding): generate a hypothetical answer to the question, embed that, then search. This often retrieves better documents than embedding the raw question.

Reranking improves precision. After vector search returns top-k candidates, a cross-encoder reranker (like Cohere Rerank) re-scores them for precision. Add this when recall is good but final answer quality is inconsistent.

In the next article, we'll give the LLM the ability to call these search functions autonomously using Tool Use.

Full source code: github.com/qameqame/pgvector-tutorial

Building a RAG System from Scratch with pgvector and Gemini — Implementation

Hiroki Kameyama — Sat, 27 Jun 2026 22:07:49 +0000

In the previous article, we covered the three core concepts behind RAG. Now let's build it.

By the end of this article you'll have a working RAG pipeline: documents stored as vectors in pgvector, semantic search retrieving the right context, and Gemini generating grounded answers.

Environment Setup

Prerequisites

Python 3.12 (pyenv recommended)
Docker
Google Gemini API key — get one free at aistudio.google.com

Project setup

mkdir pgvector-tutorial && cd pgvector-tutorial
pyenv local 3.12.0
python -m venv .venv
source .venv/bin/activate

pip install psycopg2-binary google-genai python-dotenv
pip freeze > requirements.txt

Use google-genai (new package), not google-generativeai (deprecated).

Start pgvector with Docker

docker run -d \
  --name pgvector-demo \
  -e POSTGRES_PASSWORD=password \
  -e POSTGRES_DB=vectordb \
  -p 5432:5432 \
  pgvector/pgvector:pg16

`.env` file

GEMINI_API_KEY=AIza...
DB_HOST=localhost
DB_PORT=5432
DB_NAME=vectordb
DB_USER=postgres
DB_PASSWORD=password

Directory Structure

We'll build these five files in order:

pgvector-tutorial/
├── 01_setup_db.py       # Create table + enable pgvector
├── 02_create_index.py   # HNSW index
├── 03_ingest.py         # Embed documents and store
├── 04_search.py         # Vector search
└── 05_rag.py            # Full RAG pipeline

Step 1: Database Setup — `01_setup_db.py`

import psycopg2
from dotenv import load_dotenv
import os

load_dotenv()

conn = psycopg2.connect(
    host=os.getenv("DB_HOST"), port=os.getenv("DB_PORT"),
    dbname=os.getenv("DB_NAME"), user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD"),
)
cur = conn.cursor()

cur.execute("CREATE EXTENSION IF NOT EXISTS vector;")

cur.execute("""
    CREATE TABLE IF NOT EXISTS documents (
        id         SERIAL PRIMARY KEY,
        title      TEXT NOT NULL,
        body       TEXT NOT NULL,
        category   TEXT,
        created_at TIMESTAMP DEFAULT NOW(),
        embedding  vector(768)
    );
""")

conn.commit()
print("Table created.")

python 01_setup_db.py

Why 768 dimensions? gemini-embedding-001 outputs 3072 dimensions by default, but pgvector's HNSW index has a 2000-dimension limit. Setting output_dimensionality=768 keeps us well within that limit with negligible quality loss.

Step 2: HNSW Index — `02_create_index.py`

import psycopg2
from dotenv import load_dotenv
import os

load_dotenv()
conn = psycopg2.connect(
    host=os.getenv("DB_HOST"), port=os.getenv("DB_PORT"),
    dbname=os.getenv("DB_NAME"), user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD"),
)
cur = conn.cursor()

cur.execute("""
    CREATE INDEX IF NOT EXISTS docs_embedding_idx
    ON documents
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);
""")

conn.commit()
print("Index created.")

python 02_create_index.py

HNSW parameter reference:

Use case	m	ef_construction
Dev / testing	8	32
Production (standard)	16	64
High accuracy	32	128

Step 3: Ingest Documents — `03_ingest.py`

import psycopg2
from google import genai
from google.genai import types
from dotenv import load_dotenv
import os

load_dotenv()

client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
conn = psycopg2.connect(
    host=os.getenv("DB_HOST"), port=os.getenv("DB_PORT"),
    dbname=os.getenv("DB_NAME"), user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD"),
)
cur = conn.cursor()

def get_embedding(text: str) -> list[float]:
    result = client.models.embed_content(
        model="gemini-embedding-001",
        contents=text,
        config=types.EmbedContentConfig(
            task_type="RETRIEVAL_DOCUMENT",  # use RETRIEVAL_DOCUMENT for storage
            output_dimensionality=768,
        ),
    )
    return result.embeddings[0].values

def insert_document(title: str, body: str, category: str) -> int:
    embedding = get_embedding(f"{title}\n\n{body}")
    cur.execute("""
        INSERT INTO documents (title, body, category, embedding)
        VALUES (%s, %s, %s, %s) RETURNING id;
    """, (title, body, category, embedding))
    doc_id = cur.fetchone()[0]
    conn.commit()
    return doc_id

sample_docs = [
    {
        "title": "ML Model Evaluation Metrics",
        "body": "Precision, Recall, and F1 Score are the key metrics for classification. "
                "The confusion matrix is used to calculate each metric.",
        "category": "ML",
    },
    {
        "title": "Model Evaluation with scikit-learn",
        "body": "Use cross_val_score and classification_report in Python's scikit-learn "
                "library to evaluate machine learning models.",
        "category": "ML",
    },
    {
        "title": "Data Preprocessing with Pandas",
        "body": "Handle missing values, type conversion, and outliers. "
                "Covers basic DataFrame operations and data cleaning workflows.",
        "category": "Python",
    },
    {
        "title": "AWS Cost Optimization in Practice",
        "body": "Reduce costs through EC2 instance type selection, Spot Instance usage, "
                "and deletion of unused resources.",
        "category": "Cloud",
    },
    {
        "title": "Kubernetes Pod Basics",
        "body": "A Pod is the smallest deployable unit in Kubernetes. "
                "Learn how to define Pod manifests in YAML.",
        "category": "Cloud",
    },
]

for doc in sample_docs:
    doc_id = insert_document(doc["title"], doc["body"], doc["category"])
    print(f"Stored: id={doc_id} / {doc['title']}")

python 03_ingest.py

task_type matters: Use RETRIEVAL_DOCUMENT when storing and RETRIEVAL_QUERY when searching. This asymmetric setup improves retrieval accuracy.

Step 4: Vector Search — `04_search.py`

import psycopg2
from google import genai
from google.genai import types
from dotenv import load_dotenv
import os

load_dotenv()

client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
conn = psycopg2.connect(
    host=os.getenv("DB_HOST"), port=os.getenv("DB_PORT"),
    dbname=os.getenv("DB_NAME"), user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD"),
)
cur = conn.cursor()

def get_query_embedding(text: str) -> list[float]:
    result = client.models.embed_content(
        model="gemini-embedding-001",
        contents=text,
        config=types.EmbedContentConfig(
            task_type="RETRIEVAL_QUERY",  # use RETRIEVAL_QUERY for search
            output_dimensionality=768,
        ),
    )
    return result.embeddings[0].values

def search(query: str, top_k: int = 3) -> list[dict]:
    query_embedding = get_query_embedding(query)
    cur.execute("""
        SELECT id, title, category,
               1 - (embedding <=> %s::vector) AS similarity
        FROM documents
        ORDER BY embedding <=> %s::vector
        LIMIT %s;
    """, (query_embedding, query_embedding, top_k))
    rows = cur.fetchall()
    return [
        {"id": r[0], "title": r[1], "category": r[2], "similarity": round(r[3], 4)}
        for r in rows
    ]

# Basic search
results = search("how to measure model accuracy", top_k=3)
for r in results:
    print(f"[{r['similarity']:.4f}] {r['title']} ({r['category']})")

python 04_search.py
# [0.7806] ML Model Evaluation Metrics (ML)
# [0.7423] Model Evaluation with scikit-learn (ML)
# [0.6015] Data Preprocessing with Pandas (Python)

The query "how to measure model accuracy" finds "ML Model Evaluation Metrics" even without an exact keyword match — that's the power of semantic search.

Step 5: RAG Pipeline — `05_rag.py`

import psycopg2
from google import genai
from google.genai import types
from dotenv import load_dotenv
import os

load_dotenv()

client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
conn = psycopg2.connect(
    host=os.getenv("DB_HOST"), port=os.getenv("DB_PORT"),
    dbname=os.getenv("DB_NAME"), user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD"),
)
cur = conn.cursor()

def get_query_embedding(text: str) -> list[float]:
    result = client.models.embed_content(
        model="gemini-embedding-001",
        contents=text,
        config=types.EmbedContentConfig(
            task_type="RETRIEVAL_QUERY",
            output_dimensionality=768,
        ),
    )
    return result.embeddings[0].values

def search(query: str, top_k: int = 3) -> list[dict]:
    query_embedding = get_query_embedding(query)
    cur.execute("""
        SELECT id, title, body,
               1 - (embedding <=> %s::vector) AS similarity
        FROM documents
        ORDER BY embedding <=> %s::vector
        LIMIT %s;
    """, (query_embedding, query_embedding, top_k))
    rows = cur.fetchall()
    return [
        {"id": r[0], "title": r[1], "body": r[2], "similarity": round(r[3], 4)}
        for r in rows
    ]

def rag_answer(question: str) -> str:
    # Step 1: retrieve relevant documents
    docs = search(question, top_k=3)
    if not docs:
        return "No relevant documents found."

    # Step 2: build context
    context = "\n\n".join([f"[{d['title']}]\n{d['body']}" for d in docs])

    # Step 3: build prompt and generate answer
    prompt = f"""Answer the question based on the documents below.

# Documents
{context}

# Question
{question}

# Answer (concise, grounded in the documents above)"""

    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
    )
    return response.text

answer = rag_answer("How do you calculate the F1 score?")
print(answer)

python 05_rag.py

Verification Checklist

# pgvector extension is active
docker exec -it pgvector-demo psql -U postgres -d vectordb -c "\dx"

# Check stored documents
docker exec -it pgvector-demo psql -U postgres -d vectordb \
  -c "SELECT id, title, category FROM documents;"

# Confirm embedding dimensions
docker exec -it pgvector-demo psql -U postgres -d vectordb \
  -c "SELECT id, title, vector_dims(embedding) AS dims FROM documents LIMIT 3;"
# dims should be 768

What We Built

User question
    ↓
get_query_embedding()   — convert question to vector (RETRIEVAL_QUERY)
    ↓
pgvector search         — find top-3 semantically similar documents
    ↓
Build prompt            — inject retrieved docs as context
    ↓
Gemini generate_content — produce a grounded answer
    ↓
Answer

In the next article, we'll look at this pipeline through an AI Architect's lens — why we made these design choices and how to think about scaling them.

Full source code: github.com/qameqame/pgvector-tutorial

Building a RAG System from Scratch with pgvector and Gemini — Introduction

Hiroki Kameyama — Sat, 27 Jun 2026 21:59:18 +0000

What This Guide Covers

When you start building LLM-powered applications, one pattern becomes unavoidable: RAG (Retrieval-Augmented Generation).

LLMs only know what they were trained on. Your company's internal documents, the latest spec sheets, project-specific information — none of that exists in the model. To handle data the model doesn't know, you need a system that retrieves relevant knowledge in real time and injects it into the context. That's RAG.

In this guide, we'll implement a RAG system from scratch using pgvector and Gemini, then extend it step by step through Tool Use, AI Agents, MCP, and cloud deployment.

Step 1: Embedding · Vector DB · RAG — core implementation
Step 2: AI Architect perspective — design decisions explained
Step 3: Tool Use — LLM autonomously searches the DB
Step 4: AI Agents — combining multiple tools
Step 5: MCP — exposing tools as a server
Step 6: Cloud deployment — Render × Supabase

Three Concepts to Understand First

Embedding

Computers can't measure "semantic similarity" from raw text. Embedding converts text into a list of numbers (a vector), and semantically similar words produce numerically similar patterns.

"dog"  → [0.82, 0.75, 0.10, ...]  768 numbers
"cat"  → [0.78, 0.72, 0.12, ...]  ← similar pattern to "dog"
"bank" → [0.08, 0.10, 0.85, ...]  ← completely different

Gemini's embedding model handles this conversion.

Vector DB

A regular DB searches by keyword matching. A vector DB searches by numeric distance — meaning it finds semantically related documents even when the exact words don't match.

-- Regular search (misses if keywords don't match)
SELECT * FROM docs WHERE body LIKE '%F1 score%';

-- Vector search (finds semantically related docs)
SELECT * FROM docs ORDER BY embedding <=> query_vector LIMIT 3;

Search for "how to measure model performance" and it finds "F1 score calculation" — even without matching words. We use pgvector, a PostgreSQL extension, for this.

RAG

LLMs are limited to their training data. RAG is a design pattern that retrieves relevant documents and passes them to the LLM as context, enabling the model to answer questions about data it has never seen.

[Plain LLM]  question → answers from training data only
[RAG]        question → search Vector DB → pass results to LLM → grounded answer

Who This Is For

Engineers with Python experience who are new to AI application development
Anyone who wants to understand RAG, Embedding, and vector search through code
Anyone who wants to learn hands-on from local implementation to cloud deployment

Tools Used (All Free)

Tool	Purpose	Free Tier
Google Gemini API	Embedding generation · answer generation	1,500 requests/day
pgvector (PostgreSQL extension)	Vector storage · search	Unlimited (local)
Docker	Run pgvector locally	Unlimited
Python 3.12	Implementation language	—
Render	Deploy MCP server	Free web service (with sleep)
Supabase	Cloud pgvector	500MB persistent free

Where This Fits in the AI Architect Roadmap

This guide focuses on the Applied and Design phases — the first big implementation step after learning the fundamentals (LLM basics, Prompt Engineering, API/SDK usage).

	Topic	What we implement
✓	RAG	Full RAG pipeline with pgvector and Gemini
✓	Embedding	Text-to-vector conversion with Gemini Embedding API
✓	Vector DB	Cosine similarity search with pgvector

Let's get started in the next article with environment setup and the first implementation.

Series Index

Introduction (this article)
RAG · Embedding · Vector DB Implementation
Reading RAG Design from an AI Architect's Perspective
Tool Use — Letting the LLM Search Autonomously
AI Agents — Combining Multiple Tools
MCP — Exposing pgvector Search as an MCP Server
Cloud Deployment — Render × Supabase
Wrap-up and Next Steps

Source code: github.com/qameqame/pgvector-tutorial

DEV Community: Hiroki Kameyama

Building a RAG System from Scratch — Wrap-up and What Comes Next

What We Built

Design Decisions at a Glance

pgvector over a dedicated vector DB

768 dimensions

Asymmetric task_type

HNSW over IVFFlat

Tool description is routing logic

The conversation history is the agent's memory

MCP makes tools infrastructure

Render + Supabase for zero-cost cloud deployment

The Architecture We Ended Up With

What This Series Did Not Cover

Vol.2: Production Operations Guide

Source Code

Series Index

Building a RAG System from Scratch — Cloud Deployment with Render and Supabase

Architecture

Step 1: Supabase Setup

Create a project

Enable pgvector and create the table

Get the connection string

Step 2: Migrate Local Data to Supabase

Step 3: Render-Ready MCP Server

Step 4: Deploy to Render

Verify the deployment

Step 5: Connect Your Agent

Troubleshooting

What We've Built

Building a RAG System from Scratch — MCP: Exposing pgvector as a Reusable Tool Server

Tool Use vs MCP

MCP's Three Primitives

Installing FastMCP

Step 1: MCP Server — mcp_server/server.py

Step 2: Test the Server — mcp_server/client_test.py

Step 3: Agent via MCP — 12_mcp_agent.py

Step 4: Connect to Claude Desktop

Tool Use vs MCP: The Key Difference

Building a RAG System from Scratch — AI Agents: Memory, Planning, and Multi-Step Reasoning

What Makes Something an "Agent"?

Pattern 1: ReAct — 09_agent_basic.py

Pattern 2: Long-term Memory — 10_agent_memory.py

Pattern 3: Plan-Execute-Evaluate — 11_agent_planner.py

When to Use Each Pattern

What We've Built

Building a RAG System from Scratch — Tool Use: Let the LLM Search Autonomously

What Changes with Tool Use

How Tool Use Works

Step 1: Basic Tool Use — 06_tool_basic.py

Step 2: Multiple Tools — 07_tool_multi.py

Step 3: Agentic Loop — 08_tool_agent.py

Key Patterns

What We've Added

Building a RAG System from Scratch — Design Decisions Explained

The Full Picture

Decision 1: pgvector over a Dedicated Vector DB

Decision 2: 768 Dimensions instead of 3072

Decision 3: Asymmetric task_type

Decision 4: HNSW over IVFFlat

Decision 5: Gemini 2.5 Flash for Generation

The Scaling Path

Common Pitfalls

Building a RAG System from Scratch with pgvector and Gemini — Implementation

Environment Setup

Prerequisites

Project setup

Start pgvector with Docker

.env file

Directory Structure

Step 1: Database Setup — 01_setup_db.py

Step 2: HNSW Index — 02_create_index.py

Step 3: Ingest Documents — 03_ingest.py

Step 4: Vector Search — 04_search.py

Step 5: RAG Pipeline — 05_rag.py

Verification Checklist

What We Built

Building a RAG System from Scratch with pgvector and Gemini — Introduction

What This Guide Covers

Three Concepts to Understand First

Asymmetric `task_type`

Step 1: MCP Server — `mcp_server/server.py`

Step 2: Test the Server — `mcp_server/client_test.py`

Step 3: Agent via MCP — `12_mcp_agent.py`

Pattern 1: ReAct — `09_agent_basic.py`

Pattern 2: Long-term Memory — `10_agent_memory.py`

Pattern 3: Plan-Execute-Evaluate — `11_agent_planner.py`

Step 1: Basic Tool Use — `06_tool_basic.py`

Step 2: Multiple Tools — `07_tool_multi.py`

Step 3: Agentic Loop — `08_tool_agent.py`

Decision 3: Asymmetric `task_type`

`.env` file

Step 1: Database Setup — `01_setup_db.py`

Step 2: HNSW Index — `02_create_index.py`

Step 3: Ingest Documents — `03_ingest.py`

Step 4: Vector Search — `04_search.py`

Step 5: RAG Pipeline — `05_rag.py`