Claude's Corner: Rhizome AI, The FDA Whisperer for Biotech

# Build a Regulatory Intelligence RAG Platform (Rhizome AI Clone)

A step-by-step guide to building a Rhizome-style regulatory document intelligence platform using Claude Code.

---

## Step 1: Design the Data Ingestion Pipeline

Build a multi-source document crawler and normalizer.

**Schema:**
```sql
CREATE TABLE regulatory_documents (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  source TEXT NOT NULL,           -- 'fda_guidance', 'ema_epar', 'fda_approval', etc.
  jurisdiction TEXT NOT NULL,     -- 'US', 'EU', 'UK', etc.
  document_type TEXT NOT NULL,
  title TEXT,
  document_date DATE,
  url TEXT,
  raw_text TEXT,
  metadata JSONB,
  indexed_at TIMESTAMPTZ DEFAULT now(),
  created_at TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE document_chunks (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  document_id UUID REFERENCES regulatory_documents(id),
  chunk_index INTEGER,
  chunk_text TEXT NOT NULL,
  chunk_metadata JSONB,          -- page number, section heading, table context
  embedding VECTOR(1536),        -- pgvector
  created_at TIMESTAMPTZ DEFAULT now()
);

CREATE INDEX ON document_chunks USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);
```

**Data sources to start with:**
- FDA Drugs@FDA: `https://api.fda.gov/drug/`
- FDA device 510(k): `https://api.fda.gov/device/510k/`
- FDA guidance documents: `https://www.fda.gov/regulatory-information/search-fda-guidance-documents`
- EMA EPARs: `https://www.ema.europa.eu/en/medicines/download-medicine-data`

Write scrapers in Python using `httpx` for async requests and `pdfplumber` + `pymupdf` for PDF extraction. Handle scanned PDFs with `pytesseract` OCR fallback.

---

## Step 2: Build the Document Processing and Chunking Layer

Regulatory documents need domain-specific chunking, not naive 512-token splits.

```python
def chunk_regulatory_document(text: str, metadata: dict) -> list[dict]:
    """
    Split on section headers (numbered sections like '1.', '1.1', 'SECTION IV')
    rather than token limits. Preserve table context by keeping table + 
    surrounding paragraphs together.
    """
    sections = detect_regulatory_sections(text)
    chunks = []
    for section in sections:
        if is_table(section.content):
            # Include preceding paragraph as context
            chunk = {
                "text": section.preceding_para + "\n" + section.content,
                "metadata": {**metadata, "section": section.heading, "is_table": True}
            }
        else:
            # Sliding window with 20% overlap for long sections
            for window in sliding_window(section.content, size=800, overlap=160):
                chunk = {
                    "text": window,
                    "metadata": {**metadata, "section": section.heading}
                }
        chunks.append(chunk)
    return chunks
```

Embed using `text-embedding-3-large` (OpenAI) or `claude-3-5-sonnet` via Anthropic's embedding endpoint. Store in pgvector (Postgres extension).

---

## Step 3: Build the Retrieval and Re-ranking Pipeline

Two-stage retrieval: dense vector search + regulatory-aware re-ranking.

```python
async def retrieve(query: str, top_k: int = 50) -> list[DocumentChunk]:
    query_embedding = await embed(query)
    
    # Stage 1: vector similarity search
    candidates = await db.execute("""
        SELECT *, 1 - (embedding <=> $1) AS score
        FROM document_chunks
        ORDER BY embedding <=> $1
        LIMIT $2
    """, query_embedding, top_k * 3)
    
    # Stage 2: re-rank with regulatory signals
    reranked = rerank_regulatory(candidates, query, {
        "source_authority": {"fda_guidance": 1.0, "fda_approval": 0.9, "fda_label": 0.8},
        "recency_weight": 0.15,      # recent guidance > old guidance
        "specificity_match": 0.25,   # exact product class match boost
    })
    
    return reranked[:top_k]

def rerank_regulatory(chunks, query, weights):
    for chunk in chunks:
        authority = weights["source_authority"].get(chunk.source, 0.5)
        recency = recency_score(chunk.document_date)
        specificity = compute_specificity(query, chunk.metadata)
        chunk.final_score = (
            chunk.score * 0.6 +
            authority * weights["source_authority"] * 0.15 +
            recency * weights["recency_weight"] +
            specificity * weights["specificity_match"]
        )
    return sorted(chunks, key=lambda c: c.final_score, reverse=True)
```

---

## Step 4: Implement Hallucination-Free Answer Generation

The key: every sentence must map to a retrieved passage. Use Claude's citation API.

```python
async def generate_answer(query: str, chunks: list[DocumentChunk]) -> AnswerWithCitations:
    context = format_context_with_ids(chunks)
    
    system_prompt = """You are a regulatory intelligence assistant. 
    Answer ONLY based on the provided documents. 
    For every claim, you MUST cite the document ID in [brackets].
    If the documents don't support a claim, say "The available regulatory data does not address this."
    Never speculate about regulatory positions not evidenced in the corpus."""
    
    response = await anthropic.messages.create(
        model="claude-opus-4-7",
        max_tokens=2000,
        system=system_prompt,
        messages=[{"role": "user", "content": f"Documents:\n{context}\n\nQuestion: {query}"}]
    )
    
    # Parse citations from response, validate each maps to a real chunk
    citations = parse_and_validate_citations(response.content, chunks)
    
    return AnswerWithCitations(
        text=response.content,
        citations=citations,
        confidence=compute_confidence(citations, chunks)
    )
```

**Confidence scoring:** Flag answers where cited chunks have low similarity scores or where the number of supporting citations is below threshold. Return a `LOW_CONFIDENCE` warning rather than a confident-looking wrong answer.

---

## Step 5: Build the API Layer

FastAPI backend with streaming support for long answers.

```python
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/api/query")
async def query(request: QueryRequest, user: User = Depends(get_current_user)):
    if not await check_credits(user):
        raise HTTPException(402, "Insufficient credits")
    
    # Retrieve
    chunks = await retrieve(request.question, top_k=50)
    
    if request.stream:
        return StreamingResponse(
            stream_answer(request.question, chunks),
            media_type="text/event-stream"
        )
    
    answer = await generate_answer(request.question, chunks)
    await log_query(user.id, request.question, answer, chunks)
    return answer

# Key endpoints:
# POST /api/query          - submit a question
# GET  /api/query/{id}     - fetch a prior answer
# GET  /api/document/{id}  - fetch source document with page highlights
# GET  /api/datasets       - list available regulatory datasets
```

---

## Step 6: Build the Source Viewer UI

React frontend with a split-pane layout: answer with inline citation numbers on the left, source document viewer on the right.

```tsx
function AnswerPane({ answer }: { answer: AnswerWithCitations }) {
  const [activeCitation, setActiveCitation] = useState<string | null>(null);
  
  return (
    <div className="split-pane">
      <div className="answer-content">
        <ParsedAnswer 
          text={answer.text}
          citations={answer.citations}
          onCitationClick={(id) => setActiveCitation(id)}
        />
        {answer.confidence === 'LOW' && (
          <ConfidenceWarning message="Limited regulatory precedent found for this query" />
        )}
      </div>
      
      <div className="source-viewer">
        {activeCitation && (
          <PDFViewer 
            documentId={activeCitation}
            highlightPage={answer.citations[activeCitation].page}
            highlightText={answer.citations[activeCitation].excerpt}
          />
        )}
      </div>
    </div>
  );
}
```

Use `react-pdf` for in-browser PDF rendering with highlight overlays. Store document byte offsets during indexing so you can jump to the exact page and paragraph.

---

## Step 7: Deploy with Enterprise-Grade Security

Tier 1 (cloud SaaS): Standard Postgres + pgvector on Supabase or RDS, FastAPI on Railway or Fly.io, Next.js frontend on Vercel.

Tier 2 (on-premise for enterprise): Package as a Docker Compose stack with self-contained Postgres + pgvector. The corpus sync becomes a scheduled pull from public APIs rather than a cloud-to-cloud transfer. Add a `DATA_DIR` mount for the document store.

```yaml
# docker-compose.yml for on-prem
services:
  api:
    image: your-registry/regulatory-ai-api:latest
    environment:
      - DATABASE_URL=postgresql://postgres:password@db:5432/regulatory
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - DATA_SYNC_SCHEDULE=0 2 * * *   # nightly corpus refresh
    volumes:
      - ./data:/app/data

  db:
    image: pgvector/pgvector:pg16
    volumes:
      - pgdata:/var/lib/postgresql/data

  sync:
    image: your-registry/regulatory-ai-sync:latest
    command: ["python", "sync_corpus.py", "--sources", "fda,ema"]
    environment:
      - DATABASE_URL=postgresql://postgres:password@db:5432/regulatory
```

**Security notes:**
- All document text and embeddings stay within the customer's VPC on on-prem tier
- Use row-level security in Postgres so multi-tenant cloud deployments can't cross-query
- Rate-limit at the API gateway level to prevent corpus extraction via bulk querying
- Audit log every query for enterprise compliance requirements