A production-grade multi-agent research pipeline that autonomously plans, researches, critiques, and synthesizes comprehensive cited reports from any research query -- with real-time streaming logs and human-in-the-loop review.
Argus accepts a research query via REST API and runs a supervisor-orchestrated multi-agent pipeline that:
- Plans -- decomposes the query into focused sub-questions based on depth setting
- Researches -- searches the web (Tavily), retrieves papers (ArXiv), and queries Wikipedia
- Critiques -- reviews findings for gaps; loops back if more research is needed
- Pauses for human review -- a Human-in-the-Loop gate fires at the critique boundary, letting the user continue or finalize
- Writes -- synthesizes a structured markdown report with numbered citations
The pipeline runs asynchronously -- the API returns a job_id immediately. Real-time agent logs stream via SSE to the Streamlit dashboard. Every LLM call is traced in LangSmith.
flowchart TB
subgraph Client["Client Layer"]
User["User"]
UI["Streamlit UI port 8501"]
end
subgraph API["API Layer - FastAPI port 8000"]
direction TB
Endpoint["POST /research
GET /jobs/id/status
GET /jobs/id/result
POST /jobs/id/decision"]
SSERoute["GET /jobs/id/stream
Server-Sent Events"]
RateLimit["slowapi rate limiter"]
end
subgraph Execution["Execution Layer - Background Thread"]
Runner["Pipeline Runner
BackgroundTasks thread pool"]
StreamPub["stream_manager
publish log events"]
end
subgraph Graph["LangGraph Supervisor Graph"]
direction TB
Supervisor{{"Supervisor
Command routing"}}
Planner["Planner
Sub-questions"]
Researcher["Researcher
Tavily + ArXiv + Wiki"]
Critic["Critic
Gap analysis"]
HIL["HIL Gate
interrupt check"]
Writer["Writer
Markdown + citations"]
end
subgraph Persistence["Persistence Layer"]
direction LR
JobsDB[("Jobs Table
SQLite or PostgreSQL")]
Checkpoints[("LangGraph
Checkpoints")]
end
User -->|"POST /research"| RateLimit
RateLimit --> Endpoint
Endpoint -->|"202 Accepted job_id"| User
Endpoint --> JobsDB
Endpoint -->|"BackgroundTasks"| Runner
Runner -->|"status: running"| JobsDB
Runner --> Graph
Runner --> StreamPub
Supervisor --> Planner
Planner -->|"sub_questions"| Supervisor
Supervisor --> Researcher
Researcher -->|"research_findings"| Supervisor
Supervisor --> Critic
Critic -->|"gaps_identified"| HIL
HIL -->|"gaps AND iterations less than 3"| Pause["INTERRUPT
await_human 30 min"]
Pause -->|"continue"| Researcher
Pause -->|"finalize"| Writer
HIL -->|"no gaps OR iterations 3 or more"| Writer
Writer -->|"status: complete"| JobsDB
StreamPub -->|"SSE events"| SSERoute
Graph -.->|"save state after each node"| Checkpoints
UI -->|"SSE subscribe"| SSERoute
SSERoute -->|"real-time agent logs"| UI
UI -->|"poll status / fetch result"| Endpoint
stateDiagram-v2
[*] --> supervisor : graph.invoke()
supervisor --> planner : first turn
planner --> supervisor : sub_questions set
supervisor --> researcher : after planner
researcher --> supervisor : findings accumulated
supervisor --> critic : after researcher
critic --> hil_gate : gaps_identified set
hil_gate --> INTERRUPT : gaps found AND iterations less than 3
INTERRUPT --> researcher : decision = continue
INTERRUPT --> writer : decision = finalize
hil_gate --> writer : no gaps OR iterations 3 or more
writer --> supervisor : final_report set
supervisor --> [*] : next_agent = END
Hard cap:
research_iterations >= 3bypasses HIL and routes directly towriter(auto_capped), preventing infinite loops regardless of LLM decisions.
| Component | Choice | Why |
|---|---|---|
| Agent framework | LangGraph supervisor | Multi-agent, cyclic graph, checkpointing, HIL interrupts |
| LLM | Groq Llama 3.3 70B | Free tier, 500+ tok/s, strong instruction-following |
| Web search | Tavily | Semantic search with scored, cited results |
| Paper search | ArXiv | Direct Python library |
| General knowledge | Wikipedia | Fast encyclopedic background |
| REST API | FastAPI + uvicorn | Async-native, OpenAPI docs auto-generated |
| Async tasks | FastAPI BackgroundTasks | Zero extra deps, thread-pool executor |
| Real-time logs | SSE via sse-starlette | Push agent logs to UI without polling |
| Persistence | SQLite / PostgreSQL | Zero infra locally; auto-switches via DATABASE_URL |
| Observability | LangSmith | Per-agent token counts, latency, tool traces |
| Rate limiting | slowapi | 5 req/IP/hour on POST /research |
| UI | Streamlit | SSE log streaming, HIL review panel, report render |
Argus/
+-- src/
+-- api/
| +-- main.py # FastAPI app, CORS, lifespan, startup recovery
| +-- models.py # Pydantic request/response models
| +-- celery_app.py # Celery config (Docker/production)
| +-- stream_manager.py # Thread-safe SSE pub/sub via asyncio.Queue
| +-- routes/
| +-- research.py # All job routes + BackgroundTasks + SSE stream
| +-- health.py # GET /health
+-- agents/
| +-- supervisor.py # LLM routing via Command(goto=...)
| +-- planner.py # Decomposes query into sub-questions
| +-- researcher.py # Tavily + ArXiv + Wikipedia; publishes SSE logs
| +-- critic.py # Identifies research gaps; publishes SSE logs
| +-- writer.py # Synthesizes markdown report; publishes SSE logs
+-- graph/
| +-- state.py # ResearchState TypedDict + add_messages reducer
| +-- pipeline.py # Builds + compiles LangGraph StateGraph
+-- tools/ # tavily_tool, arxiv_tool, wikipedia_tool
+-- persistence/
| +-- db.py # Dual-dialect CRUD -- SQLite or PostgreSQL
| +-- checkpointer.py # Dynamic SqliteSaver / PostgresSaver
+-- tasks/
| +-- research_tasks.py # Celery task wrappers (Docker / production)
+-- ui/
+-- streamlit_app.py # SSE streaming, HIL panel, report render
// Request
{ "query": "Latest breakthroughs in protein folding AI?", "depth": "standard" }
// depth: "quick" (~20s, Tavily only) | "standard" (~45s, all tools) | "deep" (~90s)
// Response 202
{ "job_id": "550e8400-...", "status": "pending", "estimated_seconds": 45 }// While paused for HIL review
{
"status": "awaiting_human",
"hil_payload": {
"gaps": ["Missing AlphaFold 3 comparison", "No benchmarks cited"],
"iteration": 2, "max_iterations": 3, "expires_at": "2026-07-12T18:00:00Z"
}
}
// status: pending | running | awaiting_human | complete | failed{ "decision": "continue" } // loop back to Researcher
{ "decision": "finalize" } // skip to Writerevent: message
data: {"type": "log", "message": "Researcher: Searching Tavily for ..."}
event: ping
data:
{ "status": "complete", "report": "## Report...", "sources": ["..."], "agent_turns": 4 }Interactive docs at /docs (Swagger UI auto-generated by FastAPI).
- Python 3.11+
- API keys: Groq (free), Tavily (free)
- LangSmith (optional, for observability)
- Docker Desktop (optional, for PostgreSQL / full-stack)
git clone https://github.com/noviciusss/Argus.git && cd Argus
cp .env.example .env # add your API keys
pip install -r requirements.txt
# Terminal 1 -- FastAPI
.venv\Scripts\python.exe -m uvicorn src.api.main:app --reload --port 8000
# Terminal 2 -- Streamlit
.venv\Scripts\python.exe -m streamlit run src/ui/streamlit_app.pyNo Redis or Celery needed locally. The pipeline runs in FastAPI's built-in thread pool.
docker-compose up --buildStarts PostgreSQL, Redis, Celery worker, FastAPI, and Streamlit.
- UI:
http://localhost:8501| API docs:http://localhost:8000/docs
# Add to .env
DATABASE_URL=postgresql://argus:secret@localhost:5432/argus_dbBoth the jobs table and LangGraph checkpointer switch automatically. No code changes needed.
| Variable | Required | Description |
|---|---|---|
GROQ_API_KEY |
Yes | console.groq.com -- free tier |
TAVILY_API_KEY |
Yes | tavily.com -- free tier |
LANGSMITH_API_KEY |
Recommended | smith.langchain.com |
LANGSMITH_PROJECT |
Recommended | e.g. deep-research-engine |
LANGSMITH_TRACING_V2 |
Recommended | Set to true |
DATABASE_URL |
Optional | PostgreSQL connection string; falls back to SQLite |
REDIS_URL |
Docker only | redis://redis:6379/0 |
API_BASE |
Docker only | http://api:8000 |
Why multi-agent instead of one ReAct agent?
A single ReAct agent conflates planning, researching, critiquing, and writing in one prompt. Separating into specialists allows independent prompts and error handling. The Critic reviewing findings with fresh context -- rather than the same agent that just produced them -- is the key architectural benefit.
Why put the HIL gate in its own node?
LangGraph re-executes a node from the top when resuming after interrupt(). If the interrupt lived inside supervisor_node, every resume would re-run the LLM routing call -- wasting tokens and risking different routing. A dedicated hil_node with no logic before the interrupt makes re-execution free and deterministic.
Why FastAPI BackgroundTasks instead of Celery/Redis locally?
BackgroundTasks runs sync functions in a thread-pool executor -- zero extra infrastructure. Trade-off: if the server restarts mid-research, the job is lost (status stays "running"). For Docker/production, docker-compose.yml switches to Celery + Redis. The Celery wrappers in src/tasks/research_tasks.py are kept for that path.
Why SQLite locally but PostgreSQL-ready?
src/persistence/db.py detects DATABASE_URL at startup and switches to a psycopg2 connection pool, translating ? placeholders to %s. The checkpointer.py similarly switches between SqliteSaver and PostgresSaver. The schema is identical between both backends.
What would you add with more time?
| Improvement | Status |
|---|---|
| Human-in-the-Loop gate | Done |
| Rate limiting (slowapi) | Done |
| SSE real-time log streaming | Done |
| PostgreSQL support | Done |
| Redis + Celery | In codebase, activated via Docker |
| LLM-as-Judge evaluation | Planned |
| Authentication (API keys) | Planned |
| Report caching (same query, 24h) | Planned |
| PDF export | Planned |
MIT