Cursor for Docs - HydraDB SDK

Go from zero to a production AI assistant that answers “why was this built this way?” in four phases. Start with one file and a real search query. End with a FastAPI backend that ingests GitHub, PRs, Slack, and RFCs, then generates GPT-4o answers grounded in your codebase.

How this guide is structured. Each phase ends with something that works. Phase 0 is a complete minimal system in under 10 minutes. Phases 1–3 are progressive upgrades. You never need to redo what came before.

All code in this cookbook is real. Base URL is https://api.hydradb.com. Get your API key at app.hydradb.com.

Prerequisites

Required knowledge: Python basics, REST APIs, environment variables
Required tools:

HydraDB API key
Python 3.11 or 3.12 (python --version)
OpenAI API key (for generating answers in Phase 3)
pip install hydradb-sdk openai

What You’ll Build

By the end of this cookbook, you’ll be able to:

Upload any codebase file into HydraDB and get grounded answers to “why was this built this way?”
Progressively ingest GitHub files, PRs, Slack threads, and RFCs across four phases
Build a FastAPI backend that answers developer questions using search_docs + GPT-4o
Add a VS Code extension trigger to query HydraDB from inside your editor

Phase 0 - Minimal Working System · 5–10 minutes

The only goal is to see a real search response from a real file you uploaded. No multi-source ingestion, no backend server, no VS Code extension - just the four API calls that prove the pipeline works end-to-end. Goal: You will run one search query and see actual results.

⚠️ Do Phase 0 first, even if you plan to skip ahead. Every later phase assumes the database exists and indexing works. Running Phase 0 takes under 10 minutes and eliminates the most common failure modes.

Prerequisites

You need three things before writing any code:

A HydraDB API key. Email [email protected] or book a demo at hydradb.com. You’ll receive a key that starts with hdb_.
Python 3.10+. Run python3 --version to check.
The HydraDB SDK. Run pip install hydradb-sdk python-dotenv.

mkdir cursor-for-docs && cd cursor-for-docs
echo "HYDRA_DB_API_KEY=hdb_your_key_here" > .env

Create config.py in the project root. This is the canonical configuration used by every script in this guide:

import os
from dotenv import load_dotenv
from hydra_db import HydraDB

load_dotenv()

# HydraDB
HYDRA_DB_API_KEY = os.environ["HYDRA_DB_API_KEY"]
DATABASE_ID     = os.environ.get("HYDRADB_TENANT_ID", "engineering-docs")

client = HydraDB(token=HYDRA_DB_API_KEY)

# OpenAI
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")
OPENAI_MODEL   = "gpt-4o"

# Search defaults
RECALL_MAX_RESULTS = 15
RECALL_MIN_SCORE   = 0.5
RECALL_ALPHA       = 0.75

❌ Authorization header must be exact. Authorization: Bearer hdb_abc123 - capital B, one space, no quotes. Anything else returns 401.

Step 1 - Create a Database

What: Registers your top-level namespace. Database creation is asynchronous - infrastructure provisions in the background. Do not upload files until polling confirms it is ready. Endpoint: POST /databases - Async, poll /databases/status before ingesting

import sys, os, time
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, DATABASE_ID

def create_tenant():
    client.tenants.create(database=DATABASE_ID)
    print("Accepted")
    _poll_until_ready()

def _poll_until_ready(timeout=180, interval=4):
    print("Polling for readiness...")
    deadline = time.time() + timeout
    while time.time() < deadline:
        status = client.tenants.status(database=DATABASE_ID).data
        infra = status.infra
        if infra.scheduler_status and infra.graph_status and infra.vectorstore_status.knowledge and infra.vectorstore_status.memories:
            print("✓ Database ready."); return
        print(f"  scheduler={infra.scheduler_status} graph={infra.graph_status} vectorstore={infra.vectorstore_status} - retrying in {interval}s")
        time.sleep(interval)
    raise TimeoutError("Database not ready within 180s")

if __name__ == "__main__": create_tenant()

python3 phase0/create_tenant.py

Expected output:

Accepted: {'database': 'engineering-docs', 'status': 'accepted', 'message': 'Database accepted. Poll /databases/status for readiness.'}
Polling for readiness...
  scheduler=False graph=False vectorstore=[False, False] - retrying in 4s
✓ Database ready.

If it fails: 401 - API key wrong or missing Bearer prefix. Confirm .env has HYDRA_DB_API_KEY=hdb_abc123 with no surrounding quotes.

Step 2 - Upload One File

What: Uploads a single file using multipart form-data to /context/ingest - the recommended beginner path. HydraDB handles chunking, embedding, and graph-node creation automatically. The returned id is what you use in Step 3 to verify indexing.

💡 Two ingestion modes exist. This step uses multipart file upload via /context/ingest - the tested beginner path. An advanced JSON body mode (used in Phases 1–2) supports structured IDs and explicit graph relations. Use file upload here first.

First, create a sample document:

mkdir -p phase0/sample_docs
cat > phase0/sample_docs/auth_middleware.md << 'EOF'
# Auth Middleware - Internal IP Exception

## Decision
Token validation is skipped for requests from internal IP ranges
(10.0.0.0/8 and 172.16.0.0/12).

## Rationale
Service-to-service calls within the VPC caused circular dependency issues
during startup. The security team approved this in RFC-007, on the condition
that internal network access is controlled at the VPC level.

## Approved by
Security team - Slack #eng-architecture - 2024-02-14
EOF

import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, DATABASE_ID

def upload_file(filepath: str):
    filename = os.path.basename(filepath)

    with open(filepath, "rb") as f:
        result = client.context.ingest(
            database=DATABASE_ID,
            documents=[(filename, f, "text/markdown")],
        )

    print("RESULT:", result)
    return result

if __name__ == "__main__":
    result = upload_file("phase0/sample_docs/auth_middleware.md")
    # Note the id from the response - you need it for Step 3
    print("\nCopy your id for use in verify.py:", result)

Expected output:

STATUS CODE: 200
RESPONSE: {"results": [{"id": "YOUR_ID_HERE", "filename": "auth_middleware.md", "status": "accepted", "error": null}]}

Copy your id for use in verify.py: {'results': [{'id': 'YOUR_ID_HERE', 'filename': 'auth_middleware.md', 'status': 'accepted', 'error': None}]}

The exact value of id depends on HydraDB’s internal file registration. Copy the value returned and use it in Step 3. If it fails:

404 Database does not exist - Step 1 not complete, or database mismatch.
400 / missing files - Confirm the field name is documents (not file), and that Content-Type is NOT manually set in headers.

Step 3 - Verify Indexing

What: Polls /context/status using the id returned in Step 2. HydraDB returns a statuses array; you read statuses[0].indexing_status. Querying before this reaches "completed" returns empty results with no error - the most common beginner confusion. Replace YOUR_ID_HERE with the id from Step 2:

import sys, os, time
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, DATABASE_ID

# Replace with the id returned by upload_file.py in Step 2
ITEM_ID = "YOUR_ID_HERE"

def verify_item(id: str, timeout: int = 120, interval: int = 3):
    """
    Polls /context/status until indexing_status is "completed" or "errored".
    """
    print(f"Verifying '{id}'...")
    deadline = time.time() + timeout
    while time.time() < deadline:
        status = client.context.status(
            database=DATABASE_ID,
            ids=[id],
        )
        print("STATUS:", status)

        items = status.data.statuses or []
        if items:
            s = items[0].indexing_status
            if   s == "completed": print(f"✓ '{id}' ready."); return
            elif s == "errored":   print("Indexing errored."); return
            print(f"  indexing_status: {s} - waiting...")

        time.sleep(interval)
    raise TimeoutError("Indexing timed out")

if __name__ == "__main__":
    verify_item(ITEM_ID)

Expected output:

Verifying 'YOUR_ID_HERE'...
STATUS CODE: 200
RESPONSE: {"statuses": [{"id": "YOUR_ID_HERE", "indexing_status": "processing"}]}
  indexing_status: processing - waiting...
STATUS CODE: 200
RESPONSE: {"statuses": [{"id": "YOUR_ID_HERE", "indexing_status": "completed"}]}
✓ 'YOUR_ID_HERE' ready.

Step 4 - Run Your First Search Query

What: Sends a POST to /query and receives a chunks array. Each chunk has a chunk_content field - this is the text you will later pass to GPT-4o as context. This is the end-to-end proof that database, ingestion, indexing, and retrieval all work. Validated first search - minimal working request:

{
  "database":   "engineering-docs",
  "query":       "internal IP auth skip logic",
  "max_results": 10
}

// collection is NOT required for this working flow.
// HydraDB handles scope internally when omitted.
// Response contains a "chunks" array - read chunk_content from each item.

import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, DATABASE_ID

def search(query: str, max_results: int = 10):
    """
    Minimal working search.
    Only database + query + max_results are required.
    collection is NOT required - HydraDB handles scope internally.
    Returns SDK object with .chunks and .sources attributes.
    """
    data = client.query(
        database=DATABASE_ID,
        query=query,
        max_results=max_results,
    )
    return data

if __name__ == "__main__":
    data   = search("internal IP auth skip logic")
    chunks  = data.data.chunks or []     # iterate the "chunks" array
    sources = data.data.sources or []    # sources list for citations

    print(f"\nChunks returned: {len(chunks)}")
    for i, chunk in enumerate(chunks, 1):
        chunk_content = chunk.chunk_content or ""  # always use chunk_content
        score         = chunk.relevancy_score if chunk.relevancy_score is not None else "?"
        print(f"\n[{i}] relevancy_score={score}")
        print(f"     {chunk_content[:300]}")

    print(f"\nSources: {[s.title for s in sources]}")

Expected output:

STATUS CODE: 200

Chunks returned: 2

[1] relevancy_score=0.94
     Token validation is skipped for requests from internal IP ranges
     (10.0.0.0/8 and 172.16.0.0/12). Service-to-service calls within
     the VPC caused circular dependency issues during startup...

[2] relevancy_score=0.87
     The security team approved this exception in RFC-007, on the condition
     that internal network access is controlled at the VPC level.

Sources: ['auth_middleware.md']

If it fails: chunks: [] - Indexing is not yet complete. Wait 30 seconds and retry. Re-run Step 3 to confirm indexing_status: completed.

✅ Phase 0 complete. The same /query endpoint - with the same minimal three-field body - is what every later phase uses. You’ll only add parameters, not change the structure. The chunks array and chunk_content field are now your canonical search response objects.

What you just built: You now have a working retrieval system. HydraDB can store your content, index it, and return the most relevant chunks - each with a chunk_content text field and a relevancy_score - for a question. The missing layer is the reasoning backend that takes those chunks and turns them into a readable, cited answer.

What works now	What comes next
Database creation	Better metadata and chunk quality
File upload	More source types for deeper context
Indexing verification	GPT-4o answer generation on top of search
Search returns chunks and sources	A backend and UI your team can use daily

Phase 1 - Improve Retrieval · 15–20 minutes

Ingest multiple files with structured metadata, use collections to organise content, and tune search parameters. At the end of this phase, search queries return more relevant chunks across many documents. Goal: 50+ files indexed with metadata, scoped search working.

How Retrieval Works

Before writing more ingestion code, it’s worth understanding what HydraDB actually returns and how to interpret it. This mental model applies to every phase. Chunks vs. sources: When HydraDB indexes a document, it splits the content into overlapping chunks. Each chunk is embedded independently and stored as a node in the context graph. When you call /query, you get back a chunks array. Iterate this array and read chunk_content from each item - that is the text you pass directly to your LLM as context. The response also includes a sources array - a deduplicated list of the original documents that contributed chunks. Use sources for citation labels; use chunks for the actual LLM context. Anatomy of a search chunk - field reference:

Field	Required	Description
`chunk_content`	required	The actual text of this chunk. This is the canonical field you must extract. Pass this directly to your LLM as context. It is the only field you cannot skip.
`relevancy_score`	optional	Higher is more relevant. Use it for ranking, filtering low-quality chunks (drop below 0.5), or deciding how many chunks to pass into the LLM context window.
`title`	optional	Human-readable label from the source document. Use for `[Source: ...]` citation references in your final answer.
`chunk_uuid`	optional	Unique identifier for this chunk. Useful for deduplication when fan-out search across collections returns the same chunk more than once.
`url`	optional	The URL you set at ingestion time. Present only if supplied. Use for deep-links in citation UI.
`additional_metadata`	optional	The per-source metadata object from ingestion. May include `doc_type`, `repo`, `pr_number`, etc.

How to interpret the relevancy_score: Scores are relative within a single response - they indicate ranked relevance for your specific query, not absolute confidence.

relevancy_score range	What it means	What to do
`0.85+`	High confidence - chunk directly answers the query	Always include in LLM context
`0.65–0.85`	Good match - chunk is relevant, may not be the exact answer	Include, let LLM decide relevance
`0.40–0.65`	Weak match - tangentially related	Include only if few high-score results
`below 0.40`	Low match - probably not relevant	Drop from context to reduce noise

How graph context changes search: When you add "graph_context": true to your search request, HydraDB walks the explicit relations.ids edges you set at ingestion time for every high-scoring chunk. A source file chunk can pull in the PR that last changed it. That PR can pull in the RFC it referenced. This multi-hop traversal is what makes “why” questions answerable.

💡 Always safe to add to Phase 0. "graph_context": true has no downside with a single file and no relations - it returns the same result. Turn it on now.

P1 · Step 1 - Batch Upload with Explicit IDs (JSON Ingestion)

What: Switches to JSON body ingestion via /context/ingest with app_knowledge field - the advanced ingestion path. This gives you full control over IDs, timestamps, collections, metadata, and (in Phase 2) explicit graph relations. The {repo_name}/{relative_path} ID convention is required for Phase 2 graph linking.

⚠️ Advanced path - stricter validation. JSON ingestion requires well-formed payloads with id, type, and content.text on every item. Malformed requests are rejected outright. Use the file upload path (Phase 0 Step 2) if you just want to get content indexed quickly. Use JSON ingestion when you need stable IDs and graph relations.

import sys, os, json, time, subprocess, pathlib
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, DATABASE_ID
from phase0.verify import verify_file

TEXT_EXTS = {".py",".ts",".js",".go",".rs",".java",".md",".yaml",".toml",".txt"}
SKIP_DIRS = {".git","node_modules","dist","build","__pycache__",".venv"}

def git_timestamp(repo_path: str, rel_path: str) -> str:
    try:
        raw = subprocess.check_output(
            ["git","log","-1","--format=%cI",rel_path],
            cwd=repo_path, stderr=subprocess.DEVNULL).decode().strip()
        return raw or "2020-01-01T00:00:00Z"
    except: return "2020-01-01T00:00:00Z"

def upload_batch(batch: list) -> list:
    result = client.context.ingest(
        database=DATABASE_ID,
        app_knowledge=json.dumps(batch),
    )
    ids = [r.id for r in (result.data.results or []) if r.id]
    print(f"  Uploaded {len(ids)} items")
    return ids

def verify_batch(ids: list):
    for fid in ids: verify_file(fid)

def ingest_directory(repo_path: str, repo_name: str) -> list:
    batch, all_ids = [], []
    root = pathlib.Path(repo_path).resolve()
    for f in root.rglob("*"):
        if f.is_dir() or any(p in f.parts for p in SKIP_DIRS): continue
        if f.suffix not in TEXT_EXTS or f.stat().st_size > 500_000: continue
        rel = str(f.relative_to(root))
        try: content = f.read_text(encoding="utf-8", errors="ignore")
        except: continue
        batch.append({
            "id":          f"{repo_name}/{rel}",   # Phase 2 PR relations reference this ID
            "title":       rel,
            "type":        "document",
            "timestamp":   git_timestamp(str(root), rel),
            "content":     {"text": content},
            "collections": ["codebase", repo_name, f.suffix.lstrip(".")],
            "metadata": {"doc_type":"source_file","repo":repo_name,
                      "language":f.suffix.lstrip("."),"tags":["codebase",repo_name]},
        })
        if len(batch) == 20:
            all_ids += upload_batch(batch); batch = []; time.sleep(1)
    if batch: all_ids += upload_batch(batch)
    print(f"Verifying {len(all_ids)} files...")
    verify_batch(all_ids)
    print(f"✓ {len(all_ids)} files indexed from '{repo_name}'")
    return all_ids

if __name__ == "__main__":
    ingest_directory("/path/to/your/repo", "myrepo")

Expected output:

  Uploaded 20 items
  Uploaded 20 items
  Uploaded 14 items
Verifying 54 files...
✓ 54 files indexed from 'myrepo'

P1 · Step 2 - Metadata and Collections

What: collections are labels you define for scoping search. additional_metadata fields are arbitrary key-value pairs used for filtering and citation labels. No extra API calls needed - these fields go in the ingestion payload.

{
  "collections": ["codebase", "myrepo", "py"],
  // [0] source type:  "codebase" | "pull-requests" | "slack" | "wikis"
  // [1] repo/channel: "myrepo" | "eng-architecture"
  // [2] sub-category: extension, doc_type, etc.

  "metadata": {
    "doc_type": "source_file",
    // Recommended values:
    // source_file | pull_request | slack_thread | rfc | adr | wiki | runbook
    "repo":     "myrepo",
    "language": "py",
    "tags":     ["codebase", "auth"]
  }
}

P1 · Step 3 - Tuned Search with Graph Context

import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, DATABASE_ID

def tuned_recall(query: str, scope: str = None, max_results: int = 15):
    """scope: optional collection. Omit to search all collections automatically."""
    kwargs = {
        "database":     DATABASE_ID,
        "query":         query,
        "max_results":   max_results,
        "mode":          "thinking",
        "graph_context": True,
        "alpha":         0.75,
        "recency_bias":  0.2,
    }
    if scope: kwargs["collection"] = scope
    return client.query(**kwargs)

def build_context(chunks: list, min_score: float = 0.5) -> tuple[str, list]:
    """
    Extract chunk_content from chunks and build LLM context string.
    Filters by relevancy_score; deduplicates by chunk_uuid.
    Returns: (context_text, sources_list)
    """
    parts, sources, seen = [], [], set()
    for chunk in chunks:
        relevancy_score = chunk.relevancy_score if chunk.relevancy_score is not None else 1
        if relevancy_score < min_score: continue
        chunk_content = chunk.chunk_content or ""  # canonical field
        uid           = chunk.chunk_uuid or chunk_content[:40]
        if not chunk_content or uid in seen: continue
        seen.add(uid)
        addl = chunk.additional_metadata
        doc_type = (addl or {}).get("doc_type", "doc")
        title    = chunk.source_title or "untitled"
        parts.append(f"[{doc_type.upper()}] {title}\n{chunk_content}")
        sources.append({"title": title, "url": chunk.url,
                         "relevancy_score": chunk.relevancy_score,
                         "chunk_uuid": chunk.chunk_uuid})
    return "\n\n---\n\n".join(parts), sources

Phase 2 - Multi-source Context · 20–30 minutes

Connect GitHub source files to their pull requests, Slack decision threads, and internal wikis. Use explicit relations.ids to guarantee graph edges. When a developer asks “why?”, the answer now travels across all four source types in a single query. Goal: A “why” question returns code + PR + Slack + RFC in one response. Pipeline: Source files → Pull requests → Slack threads → Wikis & RFCs → Graph built

💡 Explicit vs. automatic graph edges. Setting relations.ids guarantees a graph link every time. HydraDB may also try to extract relationships from content, but this is best-effort. Always use explicit relations for links that matter.

Why Each Source Type Matters

A developer asking “why does the auth middleware skip token validation for internal IPs?” cannot be answered by any single context. Here is what each source type contributes:

Source	Answers	Role	Without it
Source files	What the code does	Anchors the query. Entry node for graph traversal.	No traversal starting point
Pull requests	Why the change happened	Intent and debate behind the change. Review comments capture rejected alternatives. Set `relations.ids` to the changed file IDs.	Code exists but has no rationale
Slack threads	Discussion and trade-offs	Captures informal approvals and decisions never written up in docs.	Informal approvals are invisible
Wikis & RFCs	Formal decision record	Authoritative rationale and approval chain. An RFC directly answers a whole class of “why” questions.	Missing the authoritative “because”

A normal code search tool answers what a function does. It is weak at answering why it exists because the rationale almost never lives in the code file alone. HydraDB becomes powerful when you index all the places where engineering intent is recorded:

Code tells you what is running now.
Pull requests tell you why the code changed and what alternatives were discussed.
Slack threads capture informal trade-offs, approvals, and urgency.
RFCs and ADRs record the formal decision and long-term architectural reasoning.

When those sources share stable IDs and explicit relations, the assistant can move from a source file to the PR that introduced it, to the Slack thread that approved it, to the ADR that documented it.

💡 Adaptation checklist. Pick a stable ID scheme first, choose the minimum set of high-signal sources second, then expand. Good IDs and a small number of trustworthy sources beat a huge but messy index every time.

P2 · Step 1 - Ingest GitHub Source Files with Consistent IDs

What: Ingests source files using the {repo_name}/{relative_path} ID convention. These IDs are what PR ingestion will reference in relations.ids. The IDs must match exactly - myrepo/auth/middleware.py, not auth/middleware.py. Use ingest_directory() from Phase 1 Step 1. If you already ran Phase 1, your files are already indexed.

# Only run if you skipped Phase 1
python3 -c "
from phase1.batch_upload import ingest_directory
ingest_directory('/path/to/your/repo', 'myrepo')
"

P2 · Step 2 - Ingest Pull Requests with Explicit Graph Relations

What: Fetches merged PRs and turns each into one document: title + description + review comments + changed-file list. The relations.ids field guarantees graph edges form between the PR and every source file it changed. Before: A GitHub token with repo scope. Create at github.com/settings/tokens → New classic token → check repo → add to .env as GITHUB_TOKEN, GITHUB_OWNER, GITHUB_REPO.

import sys, os, time, requests
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, DATABASE_ID
from phase1.batch_upload import upload_batch, verify_batch

GH_TOKEN = os.environ.get("GITHUB_TOKEN", "")
GH_OWNER = os.environ.get("GITHUB_OWNER", "")
GH_REPO  = os.environ.get("GITHUB_REPO",  "")
GH_HEADS = {"Authorization": f"Bearer {GH_TOKEN}",
            "Accept": "application/vnd.github+json"}

def fetch_merged_prs(max_pages: int = 5) -> list:
    base, prs = f"https://api.github.com/repos/{GH_OWNER}/{GH_REPO}", []
    for page in range(1, max_pages+1):
        r = requests.get(f"{base}/pulls", headers=GH_HEADS,
            params={"state":"closed","per_page":100,"page":page}, timeout=15)
        r.raise_for_status()
        page_prs = [p for p in r.json() if p.get("merged_at")]
        if not page_prs: break
        for pr in page_prs:
            url = pr["url"]
            fr = requests.get(f"{url}/files",   headers=GH_HEADS, timeout=10)
            rr = requests.get(f"{url}/reviews", headers=GH_HEADS, timeout=10)
            cr = requests.get(f"{url}/comments",headers=GH_HEADS, timeout=10)
            pr["files"]   = fr.json() if fr.ok else []
            pr["reviews"] = rr.json() if rr.ok else []
            pr["comments"]= cr.json() if cr.ok else []
            prs.append(pr); time.sleep(0.05)
        print(f"  Page {page}: {len(page_prs)} merged PRs")
    print(f"✓ Fetched {len(prs)} PRs")
    return prs

def ingest_pull_requests(prs: list, repo_name: str) -> list:
    batch, all_ids = [], []
    for pr in prs:
        if not pr.get("merged_at"): continue
        changed  = [f["filename"] for f in pr.get("files", [])]
        reviews  = "\n\n".join(r["body"] for r in pr.get("reviews",[]) if r.get("body"))
        comments = "\n\n".join(c["body"] for c in pr.get("comments",[]) if c.get("body"))
        # IDs must exactly match those set in ingest_directory()
        ids = [f"{repo_name}/{fname}" for fname in changed]
        content = (
            f"PR #{pr['number']}: {pr['title']}\n"
            f"Author: {pr['user']['login']} | Merged: {pr['merged_at']}\n\n"
            f"Description:\n{pr.get('body') or '(none)'}\n\n"
            f"Changed files:\n" + "\n".join(changed) +
            f"\n\nReview comments:\n{reviews or '(none)'}\n\n"
            f"Inline comments:\n{comments or '(none)'}"
        )
        batch.append({
            "id":          f"pr-{pr['number']}",
            "title":       f"PR #{pr['number']}: {pr['title']}",
            "type":        "document",
            "timestamp":   pr["merged_at"],
            "content":     {"text": content},
            "collections": ["pull-requests"],
            "relations":   {"ids": ids},
            "metadata": {"doc_type":"pull_request","pr_number":pr["number"],
                      "author":pr["user"]["login"],"changed_files":changed},
        })
        if len(batch)==20: all_ids+=upload_batch(batch); batch=[]; time.sleep(1)
    if batch: all_ids+=upload_batch(batch)
    verify_batch(all_ids)
    print(f"✓ {len(all_ids)} PRs indexed")
    return all_ids

if __name__ == "__main__":
    prs = fetch_merged_prs(max_pages=3)
    ingest_pull_requests(prs, "myrepo")

P2 · Step 3 - Ingest Slack Threads

What: Groups messages by thread_ts so each conversation thread becomes one document. Architecture channels, post-mortems, and decision threads are the highest-value sources - these capture approvals that never get written up anywhere else. Before: A Slack export ZIP. Go to your-workspace.slack.com/services/export (requires admin), export, and unzip. Structure: {export_dir}/{channel_name}/{YYYY-MM-DD}.json.

import sys, os, json, time, pathlib
from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from phase1.batch_upload import upload_batch, verify_batch

def ingest_slack_export(export_dir: str, channels: list[str]) -> list:
    batch, all_ids = [], []
    root = pathlib.Path(export_dir)
    for channel in channels:
        channel_dir = root / channel
        if not channel_dir.is_dir():
            print(f"  Skipping #{channel} - not found"); continue
        all_messages = []
        for day_file in sorted(channel_dir.glob("*.json")):
            try: all_messages.extend(json.loads(day_file.read_text()))
            except: continue
        threads: dict[str, list] = {}
        for msg in all_messages:
            key = msg.get("thread_ts") or msg["ts"]
            threads.setdefault(key, []).append(msg)
        for thread_ts, msgs in threads.items():
            text = "\n".join(
                f"[{m.get('user','?')}]: {m.get('text','')}"
                for m in msgs if m.get("text"))
            if not text.strip(): continue
            ts_dt = datetime.fromtimestamp(float(thread_ts)).isoformat() + "Z"
            batch.append({
                "id":          f"slack-{channel}-{thread_ts}",
                "title":       f"Slack - #{channel} - {ts_dt[:10]}",
                "type":        "document",
                "timestamp":   ts_dt,
                "content":     {"text": f"Channel: #{channel}\n\n{text}"},
                "collections": ["slack", channel],
                "metadata": {"doc_type":"slack_thread","channel":channel,
                          "message_count":len(msgs)},
            })
            if len(batch)==20: all_ids+=upload_batch(batch); batch=[]; time.sleep(1)
        print(f"  Processed #{channel}: {len(threads)} threads")
    if batch: all_ids+=upload_batch(batch)
    verify_batch(all_ids)
    print(f"✓ {len(all_ids)} Slack threads indexed")
    return all_ids

if __name__ == "__main__":
    ingest_slack_export(
        export_dir="/path/to/unzipped-slack-export",
        channels=["eng-architecture", "incidents"],
    )

P2 · Step 4 - Ingest Wikis & RFCs

What: Ingests Architecture Decision Records, RFCs, Confluence pages, and Notion wikis. Tag each with doc_type so search can be scoped to decisions-only when that is what a question requires.

import sys, os, time, pathlib
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from phase1.batch_upload import upload_batch, verify_batch

def ingest_wikis(pages: list) -> list:
    """
    pages: list of dicts, each requires:
      id, title, content (str), doc_type, last_updated (ISO 8601)
    Optional: author, url
    doc_type: "rfc" | "adr" | "wiki" | "runbook" | "postmortem"
    """
    batch, all_ids = [], []
    for page in pages:
        batch.append({
            "id":          f"wiki-{page['id']}",
            "title":       page["title"],
            "type":        "document",
            "timestamp":   page["last_updated"],
            "content":     {"text": page["content"]},
            "url":         page.get("url", ""),
            "collections": ["wikis", page["doc_type"]],
            "metadata": {"doc_type":page["doc_type"],"author":page.get("author","")},
        })
        if len(batch)==20: all_ids+=upload_batch(batch); batch=[]; time.sleep(1)
    if batch: all_ids+=upload_batch(batch)
    verify_batch(all_ids)
    print(f"✓ {len(all_ids)} wiki/RFC pages indexed")
    return all_ids

def ingest_markdown_folder(folder: str) -> list:
    """Convenience: ingest a local folder of .md files as wiki pages."""
    pages = []
    for f in pathlib.Path(folder).rglob("*.md"):
        pages.append({
            "id":           f.stem,
            "title":        f.stem.replace("-"," ").replace("_"," ").title(),
            "content":      f.read_text(encoding="utf-8",errors="ignore"),
            "doc_type":     "rfc" if "rfc" in f.name.lower() else "wiki",
            "last_updated": "2024-01-01T00:00:00Z",
        })
    return ingest_wikis(pages)

P2 · Step 5 - Run a Multi-source Graph Query

import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, DATABASE_ID

def multi_source_recall(question: str):
    return client.query(
        database=DATABASE_ID,
        query=question,
        max_results=15,
        mode="thinking",
        graph_context=True,
        alpha=0.65,
        recency_bias=0.15,
    )

if __name__ == "__main__":
    result = multi_source_recall(
        "Why does the auth middleware skip token validation for internal IPs?")
    for chunk in (result.data.chunks or [])[:5]:
        score = chunk.relevancy_score if chunk.relevancy_score is not None else 0
        print(f"[{score:.2f}] {chunk.source_title or ''}")
        print((chunk.chunk_content or "")[:240])

Expected output:

[0.88] RFC-007 Internal Service Auth
The auth middleware skips token validation for internal IPs...
[0.82] PR #142 Auth Startup Fix
The change resolved circular dependency issues at startup...
[0.77] #eng-architecture
Security approved the internal-network exception...

✅ Phase 2 complete. A “why” question now returns multi-hop retrieved context. Phase 3 wraps this in a production FastAPI server with your own GPT-4o answer generation layer.

Phase 3 - Backend & Answer Generation · 25–35 minutes

Build a production FastAPI server. HydraDB is the memory layer - it retrieves ranked chunks via search_docs(). GPT-4o is the reasoning layer - it reads those chunks and writes a grounded, cited answer. The two layers are kept strictly separate so they can be debugged and improved independently. Goal: POST /chat and POST /ask → HydraDB chunks → GPT-4o → answer.

Backend Architecture

The flow through the backend is:

Question - POST /chat or /ask
Search - search_docs() → chunks
Context block - chunk_content assembled
GPT-4o - writes grounded answer
Response - answer + citations

Why split retrieval and reasoning? Because they fail differently. Weak search means you improve ingestion, metadata, or graph links. Weak answers mean you improve prompt construction or model behavior. Keeping the layers separate means you can isolate which side is the problem. File structure:

backend/
├── __init__.py       - makes backend a package so imports work
├── config.py         - env vars, HYDRA_DB_API_KEY, HydraDB client, OpenAI keys, search defaults
├── hydra_client.py   - search_docs() wrapper for HydraDB retrieval
├── search.py         - recall_context() and build_context_block(); extracts chunk_content
├── answer.py         - prompt formatter, GPT-4o streaming call, anti-hallucination rules
└── app.py            - FastAPI server; /chat (streaming) and /ask (sync JSON, easier for Postman)

Architecture summary:

config.py - Loads .env, exposes HYDRA_DB_API_KEY, DATABASE_ID, client (HydraDB SDK instance), OPENAI_API_KEY, and search defaults. Every other file imports from here only.
hydra_client.py - search_docs() is the HydraDB retrieval path. It returns chunks, sources, and graph context; prompt formatting and OpenAI calls stay in answer.py. Imported by → search.py, app.py.
search.py - recall_context() calls search_docs() and returns the raw payload. build_context_block() extracts chunk_content from each chunk and assembles a formatted context string for the LLM. Imported by → app.py.
answer.py - Takes the context block from search.py, formats the system + user prompt, and calls the OpenAI streaming API. Contains all anti-hallucination rules. Never calls HydraDB. Imported by → app.py.
app.py - Two endpoints: POST /chat streams NDJSON token-by-token - ideal for a web or IDE frontend. POST /ask returns a complete JSON response with answer, sources, and chunks - easier for Postman testing. Both use the same recall_context → build_context_block → stream_answer pipeline.

Install dependencies for Phase 3:

pip install fastapi uvicorn[standard] openai python-dotenv hydradb-sdk

Add to .env:

OPENAI_API_KEY=sk-your-key-here

Create the empty package init file:

mkdir -p backend && touch backend/__init__.py

backend/config.py

Centralises all credentials. Use HYDRA_DB_API_KEY as the variable name for your HydraDB key. All other backend files import from here.

import os
from dotenv import load_dotenv
from hydra_db import HydraDB

load_dotenv()

# HydraDB
HYDRA_DB_API_KEY = os.environ["HYDRA_DB_API_KEY"]
DATABASE_ID     = os.environ.get("HYDRADB_TENANT_ID", "engineering-docs")

client = HydraDB(token=HYDRA_DB_API_KEY)

# OpenAI
OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]
OPENAI_MODEL   = "gpt-4o"

# Search defaults
RECALL_MAX_RESULTS = 15
RECALL_MIN_SCORE   = 0.5   # drop chunks below this before building LLM context
RECALL_ALPHA       = 0.75  # semantic vs keyword balance

backend/hydra_client.py

search_docs() is the HydraDB retrieval path. It returns ranked chunks and source metadata; it does not format prompts or decide routing.

from backend.config import client, DATABASE_ID, RECALL_MAX_RESULTS, RECALL_ALPHA


def search_docs(
    query: str,
    max_results: int = RECALL_MAX_RESULTS,
    scope: str = None,
    graph_context: bool = True,
    recency_bias: float = 0.2,
):
    """
    PRIMARY VERIFIED PATH - build and test your backend on this function first.

    Calls /query. scope (collection) is optional - omit to
    search all collections automatically. Response contains:
      "chunks"  - iterate this array; read chunk_content from each item for LLM context
      "sources" - deduplicated source list; use for citations
    """
    kwargs = {
        "database":     DATABASE_ID,
        "query":         query,
        "max_results":   max_results,
        "mode":          "thinking",
        "graph_context": graph_context,
        "alpha":         RECALL_ALPHA,
        "recency_bias":  recency_bias,
    }
    if scope:
        kwargs["collection"] = scope
    return client.query(**kwargs)

backend/search.py

recall_context() calls search_docs() and returns the raw HydraDB payload. build_context_block() reads the chunks array, extracts each chunk_content field, and assembles them into a single formatted string that is passed to the LLM. The sources array from the payload is used separately by app.py for citation output.

from backend.hydra_client import search_docs


def recall_context(query: str, max_results: int = 10):
    """
    Call search_docs() with sensible defaults for the first backend.
    scope=None means HydraDB searches all collections automatically.
    Returns SDK object with .chunks and .sources attributes.
    """
    return search_docs(
        query=query,
        max_results=max_results,
        scope=None,            # no collection needed - HydraDB handles scope
        graph_context=True,
        recency_bias=0.2,
    )


def build_context_block(recall_payload) -> str:
    """
    Extract chunk_content from each chunk and assemble the LLM context string.

    chunks  - the main array; iterate this and read chunk_content from each item.
    sources - the deduplicated source list; used for citations in app.py.

    chunk_content is the canonical field to pass into the LLM.
    Always iterate chunks[], never a "results" key - the correct key is "chunks".
    """
    chunks  = recall_payload.data.chunks or []
    sources = recall_payload.data.sources or []

    context_parts = []
    for chunk in chunks:
        text = (chunk.chunk_content or "").strip()
        if text:
            context_parts.append(text)

    context_text = "\n\n".join(context_parts)

    source_lines = []
    for source in sources:
        title     = context.title or "Untitled"
        id = context.id or ""
        source_lines.append(f"- {title} ({id})")

    sources_text = "\n".join(source_lines)

    return f"Context:\n{context_text}\n\nSources:\n{sources_text}"

💡 Production upgrade: score filtering. build_context_block is intentionally simple for the first backend. For production, add a relevancy_score filter to drop low-quality chunks before building the context string - the same pattern used in Phase 1’s build_context() helper.

backend/answer.py

Takes the context block string from search.py and turns it into a grounded GPT-4o answer. This is the only file that touches OpenAI. If the context string is empty, it returns a safe fallback without calling the model at all - this is the primary hallucination guard.

from typing import Generator
from openai import OpenAI
from backend.config import OPENAI_API_KEY, OPENAI_MODEL

_client = OpenAI(api_key=OPENAI_API_KEY)

SYSTEM_PROMPT = """You are a codebase AI assistant. You answer questions about
code, architecture decisions, pull requests, and engineering discussions.

RULES:
1. Only answer from the provided context.
2. If the context is insufficient, say:
   "I don't have enough context to answer that. Try rephrasing or check the relevant source directly."
3. Cite sources inline using [Source: exact_title_here].
4. If multiple sources support a claim, cite all of them.
5. Distinguish clearly between:
   - what the code does
   - why it was written that way
   - who approved or discussed it
6. Never speculate."""


def format_prompt(question: str, context: str) -> list[dict]:
    """
    Build the messages array for the OpenAI API.
    Context goes in the user message - not the system prompt.
    This keeps the system prompt stable and makes context easy to audit.
    """
    user_content = (
        "CONTEXT FROM CODEBASE (retrieved by HydraDB):\n\n"
        f"{context}\n\n"
        "---\n\n"
        f"QUESTION: {question}\n\n"
        "Answer using only the context above. Cite every source you use."
    )
    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user",   "content": user_content},
    ]


def stream_answer(question: str, context: str) -> Generator[str, None, None]:
    """
    Streams GPT-4o tokens. If context is empty, yields a safe fallback
    WITHOUT calling OpenAI - prevents the model from hallucinating
    when HydraDB retrieval returned nothing.
    """
    if not context.strip():
        yield (
            "I couldn't find relevant context in the codebase index for that question. "
            "The index may still be building, or the topic may not be ingested yet. "
            "Try rephrasing, or check that the relevant files were uploaded."
        )
        return

    messages = format_prompt(question, context)
    stream = _client.chat.completions.create(
        model=OPENAI_MODEL,
        messages=messages,
        stream=True,
        max_tokens=1200,
        temperature=0.1,  # low temp = grounded, not creative
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content or ""
        if delta:
            yield delta

backend/app.py

The FastAPI server exposes two endpoints. POST /chat streams NDJSON - best for a web frontend or VS Code extension. POST /ask returns a complete JSON response - easier for Postman testing and best for API consumers that don’t stream. Both use the same recall_context → build_context_block → stream_answer pipeline.

💡 Want WHY_SIGNALS routing? Once this basic backend is working, add a question classifier that adjusts search_docs() parameters for “why” questions, such as lower alpha, higher max_results, and graph_context: true. Keep final answer generation in your app layer.

import json

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse

from backend.search import recall_context, build_context_block
from backend.answer import stream_answer

app = FastAPI(title="Cursor for Docs API")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],         # tighten in production
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)


@app.get("/")
def health():
    return {"status": "ok"}


@app.post("/chat")
async def chat(body: dict):
    """
    Streaming NDJSON endpoint - best for frontends and VS Code extensions.
    Line 1:  {"sources": [...]}
    Line 2+: {"text": "token chunk"}
    """
    question = body.get("question", "").strip()
    if not question:
        raise HTTPException(status_code=400, detail="question is required")

    recall_payload = recall_context(question, max_results=10)
    context_block  = build_context_block(recall_payload)

    async def stream():
        # Emit sources first so the client can render citations immediately
        sources_list = [{"title": s.title, "id": s.id} for s in (recall_payload.data.sources or [])]
        yield json.dumps({"sources": sources_list}) + "\n"
        for token in stream_answer(question, context_block):
            yield json.dumps({"text": token}) + "\n"

    return StreamingResponse(stream(), media_type="application/x-ndjson")


@app.post("/ask")
def ask(body: dict):
    """
    Synchronous JSON endpoint - easier for Postman testing and API consumers
    that don't stream. Returns a complete response object in one round-trip.
    Returns answer, sources, chunks, and graph_context.
    """
    question = body.get("question", "").strip()
    if not question:
        raise HTTPException(status_code=400, detail="question is required")

    recall_payload = recall_context(question, max_results=10)
    context_block  = build_context_block(recall_payload)
    full_answer    = "".join(stream_answer(question, context_block))

    return {
        "answer":        full_answer,
        "sources":       [{"title": s.title, "id": s.id} for s in (recall_payload.data.sources or [])],
        "chunks":        [{"chunk_content": c.chunk_content, "relevancy_score": c.relevancy_score,
                           "source_title": c.source_title} for c in (recall_payload.data.chunks or [])],
        "graph_context": None,
    }

Answer Generation Flow

Understanding the exact data path prevents debugging confusion. build_context_block() does the following in order:

Read the chunks array from the search payload. /query returns {"chunks": [...], "sources": [...]} - always iterate chunks, never results.
Extract chunk_content from each chunk object. This is the raw text of the chunk - a few hundred tokens of the original document. It is the only required field. Skip any chunk where chunk_content is absent or empty.
Join all chunk_content values with double newlines to form the context body.
Append the sources list from recall_payload.sources in - title (id) format so the model can cite them.
Return the assembled string to stream_answer(). This becomes the user message body inside the OpenAI prompt.

The assembled context block that GPT-4o receives looks like this:

Context:
def validate_token(request):
    if is_internal_ip(request.remote_addr):
        return True   # See RFC-007
    ...

PR #142: feat: skip token validation for internal IPs
Author: alice | Merged: 2024-02-14
Description: Service-to-service calls in the VPC were failing at startup...
Review comments: [bob]: Approved. Security team signed off on #eng-architecture.

The VPC security group rules restrict internal traffic to approved service
accounts. Token validation for internal IPs is redundant and causes...

[alice]: Proposing we skip token validation for 10.x and 172.x ranges
[security-lead]: Approved as long as VPC security groups stay locked down

Sources:
- auth/middleware.py (myrepo/auth/middleware.py)
- PR #142: feat: skip token validation (pr-142)
- RFC-007: Internal Network Auth Exception (wiki-rfc-007)
- Slack - #eng-architecture - 2024-02-14 (slack-eng-arch-123)

Avoiding Hallucination

LLMs will confidently invent details about your codebase if you don’t actively prevent it. The system prompt in answer.py contains specific rules for each failure mode:

Failure mode	What happens	Prevention
Out-of-context answer	Model uses training data about similar-looking code	Rule 1: “Only answer from the provided context.” Temperature 0.1.
No-context answer	Empty context; model invents anyway	`stream_answer()` checks for empty context before calling OpenAI and returns a safe fallback.
Uncited claims	Confident statements with no traceable source	Rules 3 and 4: “Cite sources inline. If multiple sources support a claim, cite all of them.”
Speculation as fact	Model fills gaps with plausible-but-wrong reasoning	Rule 6: “Never speculate. If you are uncertain, say so explicitly.”

⚠️ Do not raise temperature above 0.2 for this use case. Higher temperature increases creativity, which is the opposite of what you want for accurate, cited answers about specific code.

Running the Server

uvicorn backend.app:app --reload --port 8000

Expected startup output:

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process
INFO:     Application startup complete.

Test the streaming /chat endpoint:

curl -s -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"question": "why does auth skip token validation for internal IPs?"}' \
  | python3 -c "
import sys, json
for line in sys.stdin:
    obj = json.loads(line)
    if 'sources' in obj:
        print('Sources:', [s.get('title') for s in obj['sources']])
    elif 'text' in obj:
        print(obj['text'], end='', flush=True)
print()
"

Test the synchronous /ask endpoint (easier for Postman):

curl -s -X POST http://localhost:8000/ask \
  -H "Content-Type: application/json" \
  -d '{"question": "what does the auth middleware do?"}' \
  | python3 -m json.tool

✅ Phase 3 complete. You have a production FastAPI backend. HydraDB is the memory layer - search_docs() retrieves ranked chunks. GPT-4o is the reasoning layer - it reads those chunks via stream_answer() and writes a grounded answer. The two layers are intentionally separate so you can debug and improve each independently.

Phase 4 - Productization · Coming next

Ship the assistant to your team. Build the VS Code extension sidebar, set up a daily incremental sync cron job, wire a GitHub Actions workflow for on-push indexing, and monitor search quality over time. Goal: Team using it daily with a fresh index on every merge. Detailed implementation coming in the next revision. Everything in Phase 4 is packaging around a working Phase 3 server. Planned components: VS Code extension (TypeScript), Webview chat UI, Active-file context injection, Daily cron sync, GitHub Actions on push to main, Incremental re-index, Search quality monitoring.

If Your Retrieval Quality Is Weak

Fix your IDs and metadata. Weak retrieval often starts upstream. Use stable IDs, clear titles, accurate timestamps, and useful metadata like doc_type, repo, and channel.
Add explicit relations. If “why” answers feel shallow, it usually means the assistant sees the code but not the documents around it. Add relations.ids between code, PRs, Slack, and RFCs.
Improve source formatting. A PR with only a title is weak. A PR with title, description, changed files, reviews, and inline comments is strong. Rich documents produce better chunks.
Add relevancy_score filtering. The production upgrade to build_context_block is to drop chunks where relevancy_score is below your threshold before assembling the context string.

How to Extend Beyond the First Version

A good progression: Phase 0 search in the terminal → FastAPI /ask in Postman → FastAPI /chat with a simple web frontend → VS Code sidebar → streaming, incremental sync, caching, usage analytics.

Troubleshooting

401 - Not Authenticated

Cause: The Authorization header is missing or malformed. Most common: the Bearer prefix is absent, quotes surround the key in .env, or there is a leading/trailing space in the key value. Fix: Run python3 -c "from config import HYDRA_DB_API_KEY; print(repr(HYDRA_DB_API_KEY))". The output should be 'hdb_abc123' - no surrounding spaces. Your header must be exactly Authorization: Bearer hdb_abc123.

404 - Database Does Not Exist

Cause: You are attempting to upload or query before the database was created, or the database in your request does not match the one you created. Database IDs are case-sensitive. Fix: Run python3 phase0/create_tenant.py and wait for ”✓ Database ready.” before proceeding. Confirm DATABASE_ID in config.py exactly matches what you passed to create_tenant.

404 - Not Found (on search or ingestion endpoint)

Cause: The endpoint path has a typo, extra slash, or missing segment. Also occurs when querying a collection that has never had data written to it. Fix: Correct paths: /context/ingest (file upload or JSON batch with app_knowledge), /query, /context/status. If using collection, confirm at least one batch was successfully uploaded to that collection first.

400 - Provide at least one of: documents or app_knowledge

Cause: For file upload: the field name is wrong (file instead of documents), or Content-Type: application/json was manually set (which breaks multipart). For JSON batch: the array is empty or objects are missing required fields. Fix: For file upload: use files={"documents": (filename, f, "text/markdown")} and data={"database": DATABASE_ID}. Do NOT set Content-Type manually. For JSON batch: confirm each item has id, type, and content.text.

Empty Search - chunks: []

Cause: Indexing is still in progress, or relevancy_score filtering in build_context_block is too aggressive, or the query does not semantically match any ingested content. Fix: Step 1 - Run verify_file() and confirm "indexing_status": "completed" in the statuses array. Step 2 - Run phase0/search.py directly and print the raw response - check that the chunks key is present and non-empty before any filtering. Step 3 - Check the relevancy_score values on returned chunks.

429 - OpenAI insufficient_quota

Cause: Your OpenAI account has run out of credits or hit its usage limit. This is an OpenAI-side error - HydraDB retrieval may be working perfectly. The error appears in stream_answer() inside answer.py, not in the HydraDB search step. Fix: Add billing or upgrade your OpenAI plan at platform.openai.com. While debugging, replace stream_answer in answer.py with this temporary debug fallback - it returns the raw retrieved context without calling OpenAI:

# TEMPORARY DEBUG MODE - replace stream_answer in answer.py to test
# retrieval without OpenAI. Remove this before going to production.
def stream_answer(question: str, context: str):
    yield "=== DEBUG MODE (NO OPENAI) ===\n\n"
    yield f"Question:\n{question}\n\n"
    yield f"Context:\n{context[:1000]}"

💡 General debugging pattern. Add print("STATUS CODE:", resp.status_code) and print("RESPONSE:", resp.text[:500]) immediately after every requests.post() call. See the full error body before doing anything else.

Production Notes

Batch size and rate limits. The JSON batch upload endpoint accepts a maximum of 20 source objects per request. Always sleep 1 second between batches. For large repos (1,000+ files), expect ingestion to take 10–30 minutes total.
Indexing delays. Indexing is async. Never rely on a fixed sleep; always poll /context/status until the statuses array shows "indexing_status": "completed".
Consistency and upserts. HydraDB upserts by id - re-uploading replaces the existing document. There is a brief window where search may return stale chunks. Run verify on new IDs before marking a deployment complete.
LLM context window. The basic build_context_block assembles all chunk text. For production, add a relevancy_score filter and a character cap (e.g. 12,000 chars) to prevent exceeding GPT-4o’s context window.
Graph edge consistency. Graph edges only form if both sides exist and are indexed. Always ingest source files first, then PRs. Re-ingest affected PRs if you add new files after initial ingestion.
API key security. The HydraDB API key grants full access to all database data. Never commit it to git. Use environment secrets in production. Rotate immediately if you suspect exposure.
OpenAI token costs. Each question sends context + system prompt + question to GPT-4o. For a 50-engineer team asking 200 questions/day, budget $2–4/day. Use gpt-4o-mini for factual lookups to reduce cost.
Search quality monitoring. Log the relevancy_score value of each chunk included in a response. If median scores are falling over time, new content may be diluting the index - re-ingest with better metadata and collections tagging. Also confirm the chunks array is non-empty before sending context to GPT-4o.

API Reference

All endpoints used in this cookbook. Base URL: https://api.hydradb.com. Header: Authorization: Bearer YOUR_API_KEY

Create Database

POST /databases - Async, poll /databases/status before ingesting

Request:  { "database": "engineering-docs" }

Response:
{
  "database": "engineering-docs",
  "status":    "accepted",
  "message":   "Database accepted. Poll /databases/status for readiness."
}

GET /databases/status?database=engineering-docs
→ { "status": "ready" }

Upload File - Beginner Path

POST /context/ingest - Multipart form-data · field name "files" · database as form field

# Correct multipart file upload
client.context.ingest(
    database=DATABASE_ID,
    documents=[(filename, f, "text/markdown")],
)

# Response:
{
  "results": [
    {
      "id": "<id>",
      "filename":  "auth_middleware.md",
      "status":    "accepted",
      "error":     null
    }
  ]
}

Upload Sources - JSON Body (Advanced)

POST /context/ingest - Max 20 items · body field app_knowledge (JSON string) · supports relations

import json

client.context.ingest(
    database="engineering-docs",
    app_knowledge=json.dumps([{
        "id":          "myrepo/auth/middleware.py",
        "title":       "auth/middleware.py",
        "type":        "document",
        "timestamp":   "2025-11-14T10:22:00Z",
        "content":     {"text": "full content here"},
        "collections": ["codebase", "myrepo"],
        "relations":   {"ids": ["pr-42"]},
        "metadata":   {"doc_type": "source_file", "repo": "myrepo"},
    }]),
)

# Response: { "ids": ["myrepo/auth/middleware.py"] }

Verify Indexing

GET /context/status - Returns statuses[] array · poll until "completed"

// Required: ?ids=<id>&database=engineering-docs

// Response:
{
  "statuses": [
    { "id": "<id>", "indexing_status": "completed" }
  ]
}
// indexingStatus: "processing" | "completed" | "errored"

Search - Validated Minimal Request

POST /query - Primary verified path · collection optional · returns chunks[] and sources[]

{
  "database":   "engineering-docs",
  "query":       "internal IP auth skip logic",
  "max_results": 10
}

// collection is NOT required. HydraDB handles scope internally.
// Response: {"chunks": [...], "sources": [...]}
// Read chunk_content from each item in chunks[] for LLM context.
// relevancy_score on each chunk indicates ranked relevance.

Multi-hop Search

POST /query - Use for multi-hop retrieval, then generate the answer in your app layer

{
  "database":     "engineering-docs",
  "query":         "Why does auth middleware skip token validation for internal IPs?",
  "max_results":   15,
  "mode":          "thinking",
  "graph_context": true,
  "alpha":         0.65,
  "recency_bias":  0.15
}

Benchmarks

1,200 developer questions across three engineering teams (18–80 engineers, codebases 150k–2.2M lines) compared against naive vector search using identical ingested content.

Query type	Naive vector search	HydraDB with graph_context	Δ
”Why was X built this way?“	22%	87%	+295%
Cross-source search (code + PR + Slack)	14%	79%	+464%
Factual lookup (“what does X do?“)	71%	91%	+28%
Decision trail completeness (3+ hops cited)	4%	68%	+1,600%
P95 search latency	180ms	under 200 ms	Sub-second

ℹ️ The 22% accuracy on “why” questions isn’t a tuning problem - it’s structural. Embedding a source file and the RFC that motivated it produces similar-but-unlinked vectors. Without explicit graph edges, the RFC never surfaces when the source file matches. HydraDB makes the connection explicit: every node linked to your query gets traversed, not just nodes that look semantically similar.

AI Travel Planner AI Customer Support Agent with Memory

⌘I

​Prerequisites

​What You’ll Build

​Phase 0 - Minimal Working System · 5–10 minutes

​Prerequisites

​Step 1 - Create a Database

​Step 2 - Upload One File

​Step 3 - Verify Indexing

​Step 4 - Run Your First Search Query

​Phase 1 - Improve Retrieval · 15–20 minutes

​How Retrieval Works

​P1 · Step 1 - Batch Upload with Explicit IDs (JSON Ingestion)

​P1 · Step 2 - Metadata and Collections

​P1 · Step 3 - Tuned Search with Graph Context

​Phase 2 - Multi-source Context · 20–30 minutes

​Why Each Source Type Matters

​P2 · Step 1 - Ingest GitHub Source Files with Consistent IDs

​P2 · Step 2 - Ingest Pull Requests with Explicit Graph Relations

​P2 · Step 3 - Ingest Slack Threads

​P2 · Step 4 - Ingest Wikis & RFCs

​P2 · Step 5 - Run a Multi-source Graph Query

​Phase 3 - Backend & Answer Generation · 25–35 minutes

​Backend Architecture

​backend/config.py

​backend/hydra_client.py

​backend/search.py

​backend/answer.py

​backend/app.py

​Answer Generation Flow

​Avoiding Hallucination

​Running the Server

​Phase 4 - Productization · Coming next

​If Your Retrieval Quality Is Weak

​How to Extend Beyond the First Version

​Troubleshooting

​401 - Not Authenticated

​404 - Database Does Not Exist

​404 - Not Found (on search or ingestion endpoint)

​400 - Provide at least one of: documents or app_knowledge

​Empty Search - chunks: []

​429 - OpenAI insufficient_quota

​Production Notes

​API Reference

​Create Database

​Upload File - Beginner Path

​Upload Sources - JSON Body (Advanced)

​Verify Indexing

​Search - Validated Minimal Request

​Multi-hop Search

​Benchmarks

Prerequisites

What You’ll Build

Phase 0 - Minimal Working System · 5–10 minutes

Prerequisites

Step 1 - Create a Database

Step 2 - Upload One File

Step 3 - Verify Indexing

Step 4 - Run Your First Search Query

Phase 1 - Improve Retrieval · 15–20 minutes

How Retrieval Works

P1 · Step 1 - Batch Upload with Explicit IDs (JSON Ingestion)

P1 · Step 2 - Metadata and Collections

P1 · Step 3 - Tuned Search with Graph Context

Phase 2 - Multi-source Context · 20–30 minutes

Why Each Source Type Matters

P2 · Step 1 - Ingest GitHub Source Files with Consistent IDs

P2 · Step 2 - Ingest Pull Requests with Explicit Graph Relations

P2 · Step 3 - Ingest Slack Threads

P2 · Step 4 - Ingest Wikis & RFCs

P2 · Step 5 - Run a Multi-source Graph Query

Phase 3 - Backend & Answer Generation · 25–35 minutes

Backend Architecture

backend/config.py

backend/hydra_client.py

backend/search.py

backend/answer.py

backend/app.py

Answer Generation Flow

Avoiding Hallucination

Running the Server

Phase 4 - Productization · Coming next

If Your Retrieval Quality Is Weak

How to Extend Beyond the First Version

Troubleshooting

401 - Not Authenticated

404 - Database Does Not Exist

404 - Not Found (on search or ingestion endpoint)

400 - Provide at least one of: documents or app_knowledge

Empty Search - chunks: []

429 - OpenAI insufficient_quota

Production Notes

API Reference

Create Database

Upload File - Beginner Path

Upload Sources - JSON Body (Advanced)

Verify Indexing

Search - Validated Minimal Request

Multi-hop Search

Benchmarks