An agentic AI assistant for querying databases in plain English, searching internal documents with citations, and executing human-approved actions — built on open-source models and infrastructure, no external API key required.
InsightFlow-AI is a full-stack, multi-agent AI platform designed for organisations that need a conversational interface over their own data — structured databases, internal documents, and operational knowledge — without routing that data through a third-party cloud service.
The system is powered by Ollama running Qwen3 8B, an open-weight model that delivers strong reasoning and tool-calling capability without requiring a commercial API key. The backend is orchestrated via LangGraph, which models the agent workflow as a stateful directed graph with conditional routing, quality-gated retry loops, and human-in-the-loop approval pauses. Every conversation turn is checkpointed to PostgreSQL, so sessions survive restarts and partially-completed HITL workflows resume exactly where they paused.
The frontend is a real-time Next.js application that consumes LangGraph's token stream over Server-Sent Events, rendering agent transitions, tool calls, and text tokens as they are produced — giving users a transparent view of what the system is doing at every step.
Modern organisations accumulate two types of knowledge:
- Structured data in relational databases that requires SQL expertise to query.
- Unstructured knowledge in PDFs, policy documents, and internal wikis that is difficult to search at depth.
Existing solutions force a choice: either use a cloud AI API (which sends your data to an external server and incurs per-token costs at scale) or build a custom pipeline that is expensive to maintain. Neither option provides a conversational interface that spans both data types while retaining full control of the data and infrastructure.
InsightFlow-AI solves this by providing:
- Natural language to SQL — ask a question, get results from your own database with a plain-English summary.
- Document intelligence — upload PDFs and policy documents; get answers with inline citations back to the source.
- Agentic action execution — the system can take consequential actions (send emails, call webhooks) after a human explicitly approves them.
- Open-source LLM runtime — Ollama serves Qwen3 8B locally, eliminating API costs and keeping inference on your own hardware during development and self-hosted deployments.
| Component | Technology |
|---|---|
| Orchestration | LangGraph — StateGraph with conditional routing, HITL interrupt/resume, and PostgresSaver checkpointing |
| Inference Engine | Ollama — open-source runtime serving models without a commercial API key |
| Primary Model | Qwen3 8B — strong tool-calling, structured output, and optional chain-of-thought thinking mode |
| Embeddings | FastEmbed — nomic-embed-text-v1.5 via CPU ONNX runtime, 768-dimensional asymmetric retrieval |
| Vector Store | Qdrant — local-mode Rust store, HNSW dense search, payload filtering, sparse vector hybrid search |
| Workflow DAG | LangGraph StateGraph — deterministic agentic pipeline with typed shared state and per-node streaming |
| Component | Technology |
|---|---|
| Framework | FastAPI — high-performance, fully asynchronous API layer |
| Runtime | Uvicorn — ASGI server with lifespan context for startup/shutdown resource management |
| Checkpoint Store | LangGraph PostgresSaver — serialises full AgentState to PostgreSQL after every node |
| Database Driver | asyncpg — native async PostgreSQL driver, fastest Python PG client |
| ORM / Schema | SQLAlchemy 2.0 Core + Alembic — schema definition and versioned migrations |
| Data Integrity | Pydantic v2 — strict type enforcement on all API request and response models |
| Async HTTP | HTTPX — async client for webhook action execution and Ollama connectivity checks |
| Streaming Protocol | Server-Sent Events (SSE) via FastAPI StreamingResponse — token-by-token delivery with X-Accel-Buffering: no |
| Component | Technology |
|---|---|
| Framework | Next.js 14 — App Router, React Server Components, API route proxying |
| Language | TypeScript — strictly typed across all services, hooks, stores, and API contracts |
| Styling | Tailwind CSS + shadcn/ui — utility-first design system with accessible primitives |
| State Management | Zustand — lightweight stores for chat, documents, HITL, and streaming state |
| Data Fetching | TanStack Query — server-state synchronisation, background refetch, and optimistic updates |
| Streaming | Native fetch + ReadableStream — SSE consumption without the GET-only limitation of EventSource |
| Component | Technology |
|---|---|
| Relational DB | PostgreSQL 16 — application data + LangGraph checkpoint storage in a unified instance |
| Vector DB | Qdrant — local-mode Python library, zero separate process, Rust-backed on-disk HNSW |
| Migrations | Alembic — autogenerate-based versioned schema evolution |
| Configuration | Pydantic Settings — .env loading with field-level validation and typed access |
flowchart TB
USR(["👤 User"])
subgraph FE["🖥️ Frontend — Next.js · TypeScript"]
UI["Chat Interface\nToken-by-token rendering"]
DOCS["Document Manager\nUpload · Status · Delete"]
HITL_UI["HITL Approval Dialog\nApprove · Reject actions"]
TRACE["Agent Trace Panel\nLive execution visibility"]
end
subgraph BACKEND["⚙️ Backend — FastAPI · /api/v1"]
GUARD["Input Guardrails\nInjection · PII · Length"]
INGEST["Document Ingestion\nLoad · Clean · Chunk · Embed"]
STREAM["SSE Token Stream\nper-token · tool events · HITL"]
end
subgraph ORCH["🧠 LangGraph Multi-Agent Orchestration · StateGraph"]
SUP["🎯 Supervisor Agent\nIntent classification · Route decision"]
subgraph AGENTS["Specialist Agents"]
ANA["📊 Analyst Agent\nNatural Language → SQL"]
WRI["✍️ Writer Agent\nRAG retrieval · Citations"]
EXE["⚡ Executor Agent\nHITL pause · Post-approval execution"]
end
CRI["🔍 Critic Agent\nScore ≥ 0.7 → commit · Score < 0.7 → retry (max 2×)"]
SUP -->|sql_query| ANA
SUP -->|rag_search| WRI
SUP -->|action_request| EXE
ANA --> CRI
WRI --> CRI
EXE --> CRI
CRI -->|"score < 0.7 · retry"| SUP
end
subgraph STORE["💾 Storage Layer"]
PG[("PostgreSQL 16\nSessions · Messages · Documents\nAgent Logs · HITL Approvals\nLangGraph Checkpoints")]
QD[("Qdrant · Vector DB\nDocument chunks · Semantic memory")]
end
subgraph LLM["🦙 Inference · Open-Source Runtime"]
OL["Ollama · Qwen3 8B\nTool calling · Thinking mode"]
EMB["FastEmbed · nomic-embed-text-v1.5\nCPU ONNX · 768-dim"]
end
USR <-->|"chat · upload · approve"| FE
FE <-->|"HTTP · SSE"| BACKEND
BACKEND -->|"validated message"| ORCH
INGEST -->|"embed + upsert"| QD
ANA -->|"parameterised queries"| PG
WRI -->|"semantic search"| QD
EXE <-->|"pending approval"| HITL_UI
CRI -->|"score ≥ 0.7 · pass"| STREAM
STREAM -->|"token events"| UI
ORCH <-->|"checkpoint read/write"| PG
ORCH <-->|"LLM inference"| OL
INGEST <-->|"embedding"| EMB
WRI <-->|"query embedding"| EMB
InsightFlow-AI is powered by a LangGraph StateGraph, where every user request is dynamically routed through specialized AI agents instead of a fixed execution pipeline. The Supervisor determines the intent, delegates the task to the appropriate agent, the Critic validates the generated response, and the Executor safely handles real-world actions requiring user approval.
Role: Workflow Orchestrator
Classifies incoming requests into sql_query, rag_search, or action_request, produces a structured RoutingDecision using LLM structured output, and routes execution to the appropriate specialist agent through LangGraph. Separating orchestration from execution enables dynamic routing, modular workflows, and easy extensibility.
Role: Business Intelligence Specialist
Converts natural language into validated SQL queries, executes them asynchronously against PostgreSQL using asyncpg, and formats results into Markdown tables with business-friendly summaries. Allows users to query structured business data using natural language while ensuring accurate analytical responses.
Role: Retrieval-Augmented Generation (RAG) Specialist
Performs semantic retrieval from Qdrant, builds context using the most relevant document chunks, and generates grounded responses with source citations. Produces reliable, context-aware answers while reducing hallucinations through retrieval-augmented generation.
Role: Response Quality Evaluator
Evaluates responses for accuracy, completeness, groundedness, and clarity. Assigns a quality score between 0.0 and 1.0 — responses scoring below 0.70 are regenerated with improvement feedback for up to two iterations. Ensures consistent response quality through an automated evaluation and refinement loop.
Role: Action Execution Specialist
Extracts structured action requests from user conversations, stores them as pending approvals, pauses execution using LangGraph interrupt(), and executes the requested action only after explicit user approval. Provides a secure Human-in-the-Loop (HITL) workflow for external actions such as emails, webhooks, and file operations.
All five agents read from and write to a single AgentState TypedDict — the conversation's complete context at any point in time. Key fields:
| Field | Description |
|---|---|
messages |
Full conversation history — add_messages reducer (append-only) |
intent / route |
Supervisor classification and routing decision |
draft_response |
Current agent output under Critic evaluation |
retrieved_docs / sql_results |
Specialist agent outputs |
eval_score / eval_feedback |
Critic quality gate results |
pending_hitl_action |
Action awaiting human approval |
agent_trace |
Append-only execution log for live debugging |
LangGraph models the agent workflow as a StateGraph — a compiled directed graph where nodes are async Python functions that receive AgentState and return a partial state update, and edges are either deterministic or conditional. The messages field uses add_messages (append-only), ensuring no message is ever silently overwritten.
The compiled CompiledStateGraph is created once during FastAPI startup and reused for all conversations. Thread safety is achieved through the thread_id field in the invocation config — each session_id is an independent conversation lane within the same compiled graph.
When the Executor determines a consequential action is needed, it calls interrupt(). LangGraph serialises the complete current AgentState to the checkpoint store, marks the checkpoint as interrupted, and terminates the current astream_events() generator — the SSE stream closes after emitting a hitl_required event.
When the user approves via POST /hitl/{action_id}/approve, the service calls graph.update_state() with {"hitl_approved": True} and invokes astream_events() again on the same thread_id. LangGraph loads the interrupted checkpoint and continues from the hitl_router conditional edge — no state is lost.
Three independent mechanisms prevent runaway execution:
eval_iterationcounter — Critic increments this on every pass;MAX_EVAL_ITERATIONS(default: 2) forces acceptance.error_countcounter — Three consecutive node failures route to theerror_handlernode, which emits a user-facing message and terminates.- LangGraph recursion limit —
compile(recursion_limit=25)raisesGraphRecursionErrorif more than 25 node executions occur in a single invocation.
PostgreSQL serves two distinct roles in InsightFlow-AI — application data storage and LangGraph checkpoint persistence — in a single database instance with logical schema separation.
| Table | Purpose |
|---|---|
sessions |
One row per conversation. Tracks status, title (auto-generated), and timestamps. |
messages |
Human-readable conversation log with role, content, agent_name, token counts, and evaluation score. |
documents |
Tracks every ingested file: status (pending → processing → complete → failed), chunk count, file type. |
agent_logs |
Per-agent execution record: input/output summaries, latency, token usage, and error details. Primary observability surface. |
eval_logs |
Results from golden-dataset evaluation runs grouped by run_id, with per-query scores and agent traces. |
hitl_approvals |
Pending and resolved HITL decisions. Includes expires_at (24h TTL); expired approvals are never executed. |
Managed exclusively by PostgresSaver.setup() — never touched by Alembic. Three tables store the serialised AgentState after every node execution:
checkpoints— state metadata as JSONB.checkpoint_writes— incremental per-channel writes.checkpoint_blobs— large binary state like message lists.
The parent_checkpoint_id chain enables replaying any historical turn for debugging.
- Every foreign key has a supporting index. Columns used in frequent
WHEREclauses (status,session_id,created_at) have dedicated B-tree indexes. JSONBmetadatacolumns carry GIN indexes for@>operator queries. - All queries are bounded with
LIMITclauses. The most frequent path — loading conversation history — consistently executes in under 5 ms via the(session_id, sequence_number)composite index. - The asyncpg pool is sized
min=2, max=10. Two idle connections handle normal single-user load; the pool headroom accommodates concurrent background tasks (document ingestion, HITL expiry cleanup, eval runs).
InsightFlow-AI maintains four distinct memory tiers, each with a different scope and backing store:
Tier 1 — In-Context Memory
Current AgentState.messages list · limited by LLM context window (8,192 tokens)
Managed by: LangGraph add_messages reducer
Overflow strategy: sliding window + LLM summarisation every ~20 messages
↓ checkpointed after every node
Tier 2 — Checkpoint Memory
Full serialised AgentState stored in PostgreSQL checkpoints table
Scope: per session_id · survives restarts · enables HITL resume
Capacity: unlimited (disk-bounded)
↓ important facts extracted and vectorised
Tier 3 — Semantic Memory
Short text snippets (user preferences, key decisions, stated constraints)
stored in Qdrant session_memory collection · retrieved by cosine similarity
Scope: cross-session per user · retrieved at the start of every Writer turn
↓ documents ingested and chunked
Tier 4 — Document Memory (RAG)
Ingested document chunks in Qdrant documents collection
Scope: global (all sessions) · retrieved on rag_search intent
Point IDs: deterministic UUID5 from doc_id + chunk_index (upsert-safe)
When len(messages) > 20, the oldest messages are summarised by the LLM into a single SystemMessage. The originals are removed via RemoveMessage objects, keeping the active message list within the token budget while preserving conversational continuity.
The Next.js frontend is structured around three concerns: real-time streaming, agent observability, and HITL interaction.
The chat UI consumes the SSE stream via fetch + ReadableStream (not EventSource, which is GET-only and cannot carry a JSON body). The stream parser in services/sse-parser.ts splits the incoming byte stream on \n\n, extracts data: lines, and dispatches typed events to the Zustand streaming-store:
| Event | Effect |
|---|---|
token |
Appends to the in-progress message string |
agent_start |
Updates the agent trace panel |
tool_start / tool_end |
Shows a contextual "thinking" indicator |
hitl_required |
Opens the HITL approval dialog with full action details |
complete / [DONE] |
Finalises the message and re-enables input |
features/dashboard/AgentTrace.tsx renders the live agent_trace list from streaming state — users can see in real time which agents executed, in what order, and what tools were called. This transparency is a deliberate design choice: it lets users understand why the system produced a given answer.
features/hitl/ApprovalDialog.tsx is triggered by the hitl_required SSE event. It displays the full structured action payload (email recipient, subject, body; or webhook URL and payload), gives the user an Approve / Reject choice, and POSTs the decision to /api/v1/hitl/{action_id}/approve or /reject. The resumed SSE stream then picks up automatically.
Five Zustand stores handle client-side state:
| Store | Responsibility |
|---|---|
chat-store |
Conversation history, streaming buffer |
document-store |
Ingested file list and upload progress |
hitl-store |
Pending approval |
settings-store |
User preferences |
streaming-store |
Live token accumulation and agent trace |
All stores are typed and hydrated from TanStack Query's server state on mount.
| Layer | Mechanism |
|---|---|
| Input | Guardrail scan for prompt injection patterns, role-prefix spoofing, and jailbreak prefixes before any agent sees the message |
| SQL | sqlparse AST validation enforces SELECT-only mode when SQL_READONLY=True; forbidden keywords (DROP, DELETE, INSERT, ...) checked programmatically, not by regex |
| Actions | Every consequential action (email, webhook, file write) is blocked behind interrupt() — the Executor cannot execute without an explicit human approved signal |
| Quality | The Critic agent gates every response; nothing reaches the user without passing the quality threshold (or exhausting retry budget) |
| Output | Output guardrails in guardrails/output_guards.py flag hallucinations and unsupported claims before the response is committed to state |
| Trigger | Upgrade |
|---|---|
| More VRAM available | Switch to a larger Qwen3 model — change OLLAMA_MODEL in .env, no code changes |
| Multi-user requirement | Add JWT authentication — the get_current_user() dependency injection point is already wired; implement the function |
| Higher retrieval quality | Enable RAG_HYBRID_SEARCH=true to activate Qdrant's BM25 + dense Reciprocal Rank Fusion |
| Production deployment | Switch Qdrant from local mode to server mode — change one line in vector_store.initialize_qdrant() |
| Observability | agent_logs and eval_logs tables are structured for direct ingestion into any SQL-compatible analytics tool |
MIT