Insightd: Multi-step Research Agent

Research agent that takes a question, breaks it into sub-questions, searches the web in parallel, fetches & extracts findings from each source, and synthesizes a structured report with citations. Streams progress over SSE.

Built to demonstrate the full AI engineering surface: agent orchestration, structured outputs, prompt caching, observability, evals, tests, and cost tracking.

Architecture

flowchart TD
    Client[Client] -- "POST /research (SSE)" --> API[FastAPI + SSE]
    API --> Graph[LangGraph Agent]

    subgraph Graph[LangGraph Agent]
        direction LR
        Recall[recall] --> Planner[planner]
        Planner --> Searcher[searcher]
        Searcher --> Extractor[extractor]
        Extractor --> Synthesizer[synthesizer]
        Synthesizer --> Persist[persist]
    end

    Recall -. cosine search .-> Memory
    Persist -. INSERT findings .-> Memory
    Planner -. structured output .-> Gemini
    Searcher --> Exa[Exa Web Search]
    Extractor --> Fetch[httpx + readability]
    Extractor -. structured output .-> Gemini
    Synthesizer -. structured output .-> Gemini

    Gemini[Google Gemini 2.5 Flash<br/>structured outputs, free tier]
    Exa[Exa Web Search<br/>parallel per sub-question]

    Memory[(Postgres + pgvector<br/>sessions, findings, traces)]
    Voyage[Voyage AI<br/>voyage-3-large embeddings]

    API <--> Memory
    Recall <--> Voyage
    Persist <--> Voyage

Loading

Tech stack

Layer	Choice
Language	Python 3.13
Package mgmt	uv
Web framework	FastAPI + SSE
Validation	pydantic v2
Agent orchestration	LangGraph
LLM	Google gemini-2.5-flash (free tier; structured outputs)
Embeddings	Voyage AI voyage-3-large (1024-dim)
Vector DB	pgvector (Postgres 16)
Web search	Exa API
HTTP client	httpx
Observability	structlog + Langfuse (optional)
Testing	pytest + pytest-asyncio
Retries	tenacity
Linting	ruff + mypy
Container	Docker + docker-compose

Quick start

# 1. Install deps
uv sync

# 2. Set env
cp .env.example .env
# fill in GEMINI_API_KEY (free at aistudio.google.com), VOYAGE_API_KEY, EXA_API_KEY

# 3. Start Postgres + pgvector
docker compose up -d postgres

# 4. Run the API
uv run uvicorn insightd.main:app --reload --port 8000

Use it

curl -N -X POST http://localhost:8000/research \
  -H "Content-Type: application/json" \
  -d '{"query":"What were the primary causes of the 2022 Terra Luna collapse?","user_id":"demo"}'

You'll see SSE events: started, plan_created, searching, extracting, complete, done. The complete event contains the full report.

Look up a past report by session ID (returned in the X-Session-Id header):

curl http://localhost:8000/research/<session_id>
curl http://localhost:8000/sessions/<user_id>

What this demonstrates

• Multi-step agent reasoning: LangGraph state machine: recall, planner, searcher, extractor, synthesizer, persist
• Tool use: Exa web search, URL fetch + readability extraction, pgvector memory
• Streaming responses: SSE progress events the UI can render in real time
• Structured outputs: every LLM call uses Gemini's response_schema with a Pydantic model
• Prompt caching: Gemini's implicit context caching kicks in for long stable prefixes
• State management: typed AgentState with reducers for accumulating sources/findings
• Full RAG loop: recall_node embeds the query and pulls similar findings from the user's past sessions; persist_node writes new findings back. Synthesizer integrates prior knowledge into the final report.
• Cost tracking: every LLM call records input/output/cache tokens and dollar cost (zero on free tier)
• Retries with backoff: tenacity on transient API errors only (not on BadRequestError)
• Observability: structured JSON logs via structlog; optional Langfuse tracing
• Evals: golden dataset, faithfulness/completeness/citation evaluators, pass-rate report
• Tests: schema unit tests, LLM client retry tests, end-to-end agent integration test

Run tests

uv run pytest -q
uv run pytest --cov=src/insightd --cov-report=term-missing

The integration test_memory.py is gated behind RUN_INTEGRATION_DB=1 and a live pgvector Postgres.

Run evals

uv run python -m evals.run_evals --limit 3   # smoke test on 3 questions
uv run python -m evals.run_evals             # full suite

Results land in evals/results/run_*.json with per-question scores and a pass-rate summary. Three judges run on each report:

Evaluator	What it checks	Threshold
Faithfulness	Each claim is entailed by its evidence (LLM-as-judge)	≥ 0.80
Completeness	Report covers expected topics from the dataset	≥ 0.70
Citation coverage	Every finding has a source URL	≥ 0.95

Project layout

src/insightd/
├── main.py              # ASGI entry: uvicorn insightd.main:app
├── config.py            # pydantic-settings, all env vars
├── schemas/             # All API + agent data shapes (research.py, memory.py)
├── agent/
│   ├── graph.py         # LangGraph wiring
│   ├── nodes.py         # planner / searcher / extractor / synthesizer
│   ├── prompts.py       # Versioned system prompts
│   └── state.py         # AgentState TypedDict with reducers
├── tools/
│   ├── search.py        # Exa web search (retried, normalized)
│   ├── fetch.py         # URL fetch + readability extraction
│   └── memory.py        # pgvector + Voyage embeddings
├── llm/client.py        # Gemini wrapper: retries, caching, parse, cost
├── db/                  # asyncpg pool + 001_init.sql migration
├── observability/       # structlog config + optional Langfuse
└── api/
    ├── routes.py        # /research, /research/{id}, /sessions/{user}, /health
    └── streaming.py     # SSE generator from graph.astream

tests/
├── unit/                # schemas, LLM client (mocked)
└── integration/         # agent end-to-end (mocked tools), pgvector memory

evals/
├── datasets/            # 10-question golden set (research_questions.jsonl)
├── evaluators.py        # 3 evaluators: faithfulness, completeness, citation
└── run_evals.py         # Runner with summary + per-question JSON output

Notes

• Default model gemini-2.5-flash runs on the free tier (get a key at aistudio.google.com).
• Structured outputs use Gemini's response_schema with Pydantic models; response.parsed is a validated instance.
• Search and fetch run in parallel with concurrency limits (asyncio.Semaphore(5) for fetch).
• Failed source fetches don't fail the whole run; they're logged and skipped.
• Cost-tracking math accounts for Gemini's cached input billing (0.25× rate when context cache hits).