How to Build an AI Research Assistant with Perplexity API

Table of Contents

• How to Build an AI Research Assistant with Perplexity API
  • Real-World Use Case and Architecture
  • Prerequisites and Environment Setup
• Create virtual environment
• Install dependencies
  • Core Implementation: Building the Research Assistant
    • Database Schema and Session Management
• models.py
• Database initialization
  • Perplexity API Client with Rate Limiting

📺 Watch: Neural Networks Explained

Video by 3Blue1Brown

Building a production-grade AI research assistant requires more than just wrapping an API call. You need to handle context management, citation tracking, rate limiting, and result persistence. In this tutorial, we'll build a complete research assistant using the Perplexity API that can search academic literature, summarize findings, and maintain conversation history with proper attribution.

According to recent research in generative information retrieval, systems that combine real-time web search with large language models achieve significantly better factual accuracy than standalone LLMs [1]. The Perplexity API provides exactly this capability—it searches the web in real-time and returns cited responses, making it ideal for research applications.

Real-World Use Case and Architecture

Before diving into code, let's understand why this matters in production. Research assistants built on pure LLMs suffer from hallucination and stale knowledge. A 2025 study found that AI predictions often lead users to forgo guaranteed rewards when the underlying model lacks access to current information [2]. By integrating Perplexity's real-time search, we ground our assistant in verifiable sources.

Our architecture follows a three-tier pattern:

1. Orchestration Layer: FastAPI endpoints that manage user sessions and request routing
2. Search Layer: Perplexity API client with rate limiting and retry logic
3. Persistence Layer: SQLite database for conversation history and citation storag [1]e

The key design decision is separating search from summarization. Perplexity handles both, but we cache results to avoid redundant API calls and maintain a local citation graph for auditability.

Prerequisites and Environment Setup

You'll need Python 3.10+ and a Perplexity API key. Let's set up the environment:

# Create virtual environment
python -m venv research-assistant
source research-assistant/bin/activate  # On Windows: research-assistant\Scripts\activate

# Install dependencies
pip install fastapi uvicorn httpx pydantic sqlalchemy aiosqlite python-dotenv

Create a .env file in your project root:

PERPLEXITY_API_KEY=your_api_key_here
DATABASE_URL=sqlite+aiosqlite:///research.db
MAX_RETRIES=3
RATE_LIMIT_RPM=10

The rate limit of 10 requests per minute is conservative—Perplexity's actual limits depend on your plan tier. According to their documentation, the Pro plan allows 100 requests per minute, but we'll implement client-side throttling to be safe.

Core Implementation: Building the Research Assistant

Database Schema and Session Management

First, let's define our data models. We need to store conversations, search results, and citations separately for proper attribution:

# models.py
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, JSON
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from datetime import datetime
import uuid

Base = declarative_base()

class Session(Base):
    __tablename__ = "sessions"

    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    metadata = Column(JSON, default=dict)

    messages = relationship("Message", back_populates="session", cascade="all, delete-orphan")

class Message(Base):
    __tablename__ = "messages"

    id = Column(Integer, primary_key=True, autoincrement=True)
    session_id = Column(String, ForeignKey("sessions.id"), nullable=False)
    role = Column(String, nullable=False)  # "user" or "assistant"
    content = Column(Text, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

    session = relationship("Session", back_populates="messages")
    citations = relationship("Citation", back_populates="message", cascade="all, delete-orphan")

class Citation(Base):
    __tablename__ = "citations"

    id = Column(Integer, primary_key=True, autoincrement=True)
    message_id = Column(Integer, ForeignKey("messages.id"), nullable=False)
    source_url = Column(String, nullable=False)
    source_title = Column(String)
    snippet = Column(Text)
    relevance_score = Column(Integer)  # 0-100

    message = relationship("Message", back_populates="citations")

# Database initialization
engine = create_async_engine("sqlite+aiosqlite:///research.db", echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

The schema design addresses a critical production concern: citation provenance. Each assistant message has a one-to-many relationship with citations, allowing us to trace every claim back to its source. This is essential for research integrity, as highlighted in recent work on ethical AI use in research practices [3].

Perplexity API Client with Rate Limiting

Now let's build the core API client. We'll implement exponential backoff and token bucket rate limiting:

# perplexity_client.py
import asyncio
import time
from typing import Optional, List, Dict
import httpx
from pydantic import BaseModel, Field
from dotenv import load_dotenv
import os

load_dotenv()

class SearchRequest(BaseModel):
    query: str
    max_tokens: int = Field(default=1024, le=4096)
    temperature: float = Field(default=0.2, ge=0.0, le=1.0)
    top_p: float = Field(default=0.9, ge=0.0, le=1.0)
    search_domain_filter: Optional[List[str]] = None  # e.g., ["arxiv.org", "scholar.google.com"]
    return_citations: bool = True

class SearchResult(BaseModel):
    content: str
    citations: List[Dict[str, str]]
    model: str
    usage: Dict[str, int]

class RateLimiter:
    """Token bucket rate limiter for API requests."""

    def __init__(self, requests_per_minute: int = 10):
        self.tokens = requests_per_minute
        self.max_tokens = requests_per_minute
        self.refill_rate = requests_per_minute / 60.0  # tokens per second
        self.last_refill = time.monotonic()
        self.lock = asyncio.Lock()

    async def acquire(self):
        async with self.lock:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(self.max_tokens, self.tokens + elapsed * self.refill_rate)
            self.last_refill = now

            if self.tokens < 1:
                wait_time = (1 - self.tokens) / self.refill_rate
                await asyncio.sleep(wait_time)
                self.tokens = 0
            else:
                self.tokens -= 1

class PerplexityClient:
    """Production-grade client for Perplexity API with retry and rate limiting."""

    BASE_URL = "https://api.perplexity.ai"

    def __init__(self, api_key: str = None, max_retries: int = 3):
        self.api_key = api_key or os.getenv("PERPLEXITY_API_KEY")
        if not self.api_key:
            raise ValueError("PERPLEXITY_API_KEY must be provided or set in environment")

        self.max_retries = max_retries
        self.rate_limiter = RateLimiter(int(os.getenv("RATE_LIMIT_RPM", "10")))
        self.client = httpx.AsyncClient(
            base_url=self.BASE_URL,
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            timeout=30.0
        )

    async def search(self, request: SearchRequest) -> SearchResult:
        """Execute a search with exponential backoff retry."""

        for attempt in range(self.max_retries):
            try:
                await self.rate_limiter.acquire()

                payload = {
                    "model": "sonar-pro",  # Perplexity's research-optimized model
                    "messages": [
                        {
                            "role": "system",
                            "content": "You are a research assistant. Provide detailed, cited answers. Focus on academic and technical sources."
                        },
                        {
                            "role": "user",
                            "content": request.query
                        }
                    ],
                    "max_tokens": request.max_tokens,
                    "temperature": request.temperature,
                    "top_p": request.top_p,
                    "return_citations": request.return_citations,
                    "search_domain_filter": request.search_domain_filter or ["arxiv.org", "scholar.google.com"]
                }

                response = await self.client.post("/chat/completions", json=payload)
                response.raise_for_status()
                data = response.json()

                # Parse citations from response
                citations = []
                if "citations" in data:
                    for citation in data["citations"]:
                        citations.append({
                            "url": citation.get("url", ""),
                            "title": citation.get("title", ""),
                            "snippet": citation.get("snippet", "")
                        })

                return SearchResult(
                    content=data["choices"][0]["message"]["content"],
                    citations=citations,
                    model=data["model"],
                    usage=data["usage"]
                )

            except httpx.HTTPStatusError as e:
                if e.response.status_code == 429:  # Rate limited
                    wait_time = min(2 ** attempt * 10, 60)  # Exponential backoff
                    await asyncio.sleep(wait_time)
                    continue
                elif e.response.status_code == 401:
                    raise PermissionError("Invalid API key. Check your Perplexity API credentials.")
                else:
                    raise
            except httpx.TimeoutException:
                if attempt == self.max_retries - 1:
                    raise TimeoutError("Perplexity API request timed out after all retries")
                await asyncio.sleep(2 ** attempt)

        raise RuntimeError("Max retries exceeded")

    async def close(self):
        await self.client.aclose()

Key design decisions in this client:

1. Token bucket rate limiting: More sophisticated than simple time.sleep()—it allows burst requests up to the limit while maintaining average throughput.
2. Domain filtering: We default to academic sources (arxiv.org, scholar.google.com) but allow override. This is crucial for research credibility.
3. Exponential backoff: With jitter (implicit through async timing), we handle transient failures gracefully.

FastAPI Application with Session Management

Now let's wire everything together with FastAPI:

# main.py
from fastapi import FastAPI, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from contextlib import asynccontextmanager
import logging

from models import async_session, init_db, Session as DBSession, Message, Citation
from perplexity_client import PerplexityClient, SearchRequest

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Global client instance
perplexity_client = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Handle startup and shutdown events."""
    global perplexity_client
    await init_db()
    perplexity_client = PerplexityClient()
    logger.info("Research assistant initialized")
    yield
    await perplexity_client.close()
    logger.info("Research assistant shutdown")

app = FastAPI(
    title="AI Research Assistant",
    version="1.0.0",
    lifespan=lifespan
)

# CORS for frontend integration
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Restrict in production
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

async def get_db():
    async with async_session() as session:
        yield session

@app.post("/sessions")
async def create_session(db: AsyncSession = Depends(get_db)):
    """Create a new research session."""
    session = DBSession()
    db.add(session)
    await db.commit()
    await db.refresh(session)
    return {"session_id": session.id, "created_at": session.created_at.isoformat()}

@app.post("/sessions/{session_id}/query")
async def research_query(
    session_id: str,
    request: SearchRequest,
    db: AsyncSession = Depends(get_db)
):
    """Execute a research query within a session context."""

    # Verify session exists
    result = await db.execute(select(DBSession).where(DBSession.id == session_id))
    session = result.scalar_one_or_none()
    if not session:
        raise HTTPException(status_code=404, detail="Session not found")

    # Store user message
    user_message = Message(
        session_id=session_id,
        role="user",
        content=request.query
    )
    db.add(user_message)

    try:
        # Execute search
        search_result = await perplexity_client.search(request)

        # Store assistant response
        assistant_message = Message(
            session_id=session_id,
            role="assistant",
            content=search_result.content
        )
        db.add(assistant_message)
        await db.flush()  # Get message ID

        # Store citations
        for citation in search_result.citations:
            db_citation = Citation(
                message_id=assistant_message.id,
                source_url=citation["url"],
                source_title=citation.get("title", ""),
                snippet=citation.get("snippet", ""),
                relevance_score=85  # Default score, could be refined
            )
            db.add(db_citation)

        await db.commit()

        return {
            "content": search_result.content,
            "citations": search_result.citations,
            "model": search_result.model,
            "usage": search_result.usage
        }

    except Exception as e:
        await db.rollback()
        logger.error(f"Query failed: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Research query failed: {str(e)}")

@app.get("/sessions/{session_id}/history")
async def get_history(session_id: str, db: AsyncSession = Depends(get_db)):
    """Retrieve conversation history with citations."""

    result = await db.execute(
        select(Message)
        .where(Message.session_id == session_id)
        .order_by(Message.created_at)
    )
    messages = result.scalars().all()

    history = []
    for msg in messages:
        msg_dict = {
            "role": msg.role,
            "content": msg.content,
            "created_at": msg.created_at.isoformat()
        }

        if msg.role == "assistant":
            # Fetch citations for this message
            citations_result = await db.execute(
                select(Citation).where(Citation.message_id == msg.id)
            )
            citations = citations_result.scalars().all()
            msg_dict["citations"] = [
                {
                    "url": c.source_url,
                    "title": c.source_title,
                    "snippet": c.snippet
                }
                for c in citations
            ]

        history.append(msg_dict)

    return {"session_id": session_id, "messages": history}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Edge Cases and Production Considerations

Handling API Limits and Failures

The Perplexity API has rate limits that vary by plan. Our implementation handles several edge cases:

1. Rate limit exceeded (429): Exponential backoff with jitter prevents thundering herd problems
2. Authentication failure (401): Clear error message helps debugging
3. Timeout: Configurable retry with increasing wait times
4. Empty results: The API may return no citations for very specific queries—we handle this gracefully

Memory Management

For long-running sessions, conversation history can grow large. Consider implementing:

# Optional: Session pruning for memory management
async def prune_old_sessions(max_sessions: int = 100):
    """Remove oldest sessions when limit exceeded."""
    async with async_session() as db:
        result = await db.execute(
            select(DBSession).order_by(DBSession.updated_at.desc())
        )
        sessions = result.scalars().all()

        if len(sessions) > max_sessions:
            to_delete = sessions[max_sessions:]
            for session in to_delete:
                await db.delete(session)
            await db.commit()

Citation Quality Assurance

Not all citations are equally valuable. The relevance_score field in our schema allows for future refinement. You could implement a post-processing step that:

1. Validates URLs are still accessible
2. Checks domain authority (e.g., .edu vs .com)
3. Cross-references citations across multiple queries for consistency

Testing Your Research Assistant

Start the server and test with curl:

# Start the server
python main.py

# In another terminal, create a session
curl -X POST http://localhost:8000/sessions

# Use the returned session_id to query
curl -X POST http://localhost:8000/sessions/{session_id}/query \
  -H "Content-Type: application/json" \
  -d '{"query": "What are the latest advances in transformer architectures for NLP?"}'

What's Next

This research assistant provides a solid foundation, but production deployment requires additional considerations:

1. Authentication: Add JWT-based user authentication for multi-tenant support
2. Caching: Implement Redis-based response caching to reduce API costs
3. Streaming: Use Server-Sent Events (SSE) for real-time response streaming
4. Monitoring: Integrate with OpenTelemetry for observability
5. Feedback Loop: Allow users to rate responses and flag incorrect citations

The integration of real-time search with LLMs represents a paradigm shift in research tools. As the field evolves, we'll see more sophisticated citation graphs and cross-referencing capabilities. The ethical considerations raised in recent research [3] remind us that these tools should augment, not replace, human judgment in research.

For further reading on the theoretical foundations, check out the comprehensive survey on generative information retrieval [1], which provides context for why search-augmented LLMs outperform standalone models.