Agentic Research Agent Fastapi

Agentic Research Agent (FastAPI)

Skill by ara.so — AI Agent Skills collection.

What This Project Does

The Agentic AI Research Agent is a FastAPI service that orchestrates multi-step research workflows using LLM-powered agents. It implements an agentic workflow pattern with:

• Planning Agent: Breaks down research tasks into subtasks
• Executor Agent: Runs research, writing, and editing agents with tool use
• Research Tools: Tavily search, arXiv papers, Wikipedia lookups
• Reflection/Iteration: Multi-step refinement of research outputs
• Task Tracking: PostgreSQL-backed state management for async workflows
• Web UI: Simple interface to kick off and monitor research tasks

The system runs in a single Docker container with both PostgreSQL and the FastAPI service.

Installation

Prerequisites

1. Docker installed on your system
2. API Keys in a .env file at project root:

OPENAI_API_KEY=sk-your-openai-key
TAVILY_API_KEY=tvly-your-tavily-key

Build and Run

# Clone the repository
git clone https://github.com/https-deeplearning-ai/agentic-ai-public.git
cd agentic-ai-public

# Create .env file with your API keys
cat > .env << EOF
OPENAI_API_KEY=your-openai-key
TAVILY_API_KEY=your-tavily-key
EOF

# Build the Docker image
docker build -t fastapi-postgres-service .

# Run the container (exposes API on 8000, Postgres on 5432)
docker run --rm -it \
  -p 8000:8000 \
  -p 5432:5432 \
  --name fpsvc \
  --env-file .env \
  fastapi-postgres-service

The service will start Postgres, create the database, and launch the FastAPI app on http://localhost:8000.

Project Structure

.
├── main.py                     # FastAPI app with routes and DB models
├── src/
│   ├── planning_agent.py       # Planner and executor agent logic
│   ├── agents.py               # Research, writer, editor agents
│   └── research_tools.py       # Tool implementations (Tavily, arXiv, Wikipedia)
├── templates/
│   └── index.html              # Web UI
├── static/                     # CSS/JS assets
├── docker/
│   └── entrypoint.sh           # Container startup script
├── requirements.txt            # Python dependencies
└── Dockerfile

Key API Endpoints

1. Web UI

# Open in browser
http://localhost:8000/

2. Generate Research Report (POST)

curl -X POST http://localhost:8000/generate_report \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "Large Language Models for scientific discovery",
    "model": "openai:gpt-4o"
  }'

# Response:
# {"task_id": "550e8400-e29b-41d4-a716-446655440000"}

3. Poll Task Progress (GET)

curl http://localhost:8000/task_progress/550e8400-e29b-41d4-a716-446655440000

# Response:
# {
#   "status": "running",
#   "current_step": "research",
#   "steps_completed": ["planning"],
#   "message": "Running research agent..."
# }

4. Get Final Report (GET)

curl http://localhost:8000/task_status/550e8400-e29b-41d4-a716-446655440000

# Response:
# {
#   "status": "completed",
#   "report": "# Research Report\n\n...",
#   "created_at": "2025-01-15T10:30:00",
#   "completed_at": "2025-01-15T10:32:45"
# }

Core Components

1. Planning Agent (src/planning_agent.py)

The planner breaks down user prompts into structured subtasks:

from src.planning_agent import planner_agent, executor_agent_step

# Generate a plan from a user prompt
plan = planner_agent(
    user_prompt="Research quantum computing applications in drug discovery",
    model="openai:gpt-4o"
)

# Plan structure:
# {
#   "steps": [
#     {"step": "research", "description": "Search for papers on quantum computing + drug discovery"},
#     {"step": "write", "description": "Draft report synthesizing findings"},
#     {"step": "edit", "description": "Refine report for clarity"}
#   ]
# }

# Execute each step
for step in plan["steps"]:
    result = executor_agent_step(
        step=step,
        context={},
        model="openai:gpt-4o"
    )
    print(f"Step {step['step']}: {result}")

2. Research Tools (src/research_tools.py)

Three main tools for information gathering:

from src.research_tools import (
    tavily_search_tool,
    arxiv_search_tool,
    wikipedia_search_tool
)

# Tavily web search (requires TAVILY_API_KEY)
results = tavily_search_tool(query="quantum computing drug discovery")
# Returns: list of {"title": "...", "url": "...", "content": "..."}

# arXiv academic paper search
papers = arxiv_search_tool(query="quantum machine learning", max_results=5)
# Returns: list of {"title": "...", "authors": [...], "summary": "...", "pdf_url": "..."}

# Wikipedia lookup
article = wikipedia_search_tool(query="Quantum computing")
# Returns: {"title": "...", "summary": "...", "url": "..."}

3. Agent Implementations (src/agents.py)

Specialized agents for different workflow stages:

from src.agents import research_agent, writer_agent, editor_agent

# Research agent: Gathers information using tools
research_output = research_agent(
    topic="quantum computing applications",
    tools=[tavily_search_tool, arxiv_search_tool],
    model="openai:gpt-4o"
)

# Writer agent: Synthesizes research into a draft
draft = writer_agent(
    research_data=research_output,
    outline="Introduction, Applications, Conclusion",
    model="openai:gpt-4o"
)

# Editor agent: Refines and polishes the draft
final_report = editor_agent(
    draft=draft,
    style_guide="academic, concise",
    model="openai:gpt-4o"
)

4. Database Models (main.py)

Task tracking with SQLAlchemy:

from sqlalchemy import create_engine, Column, String, Text, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os

DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://app:local@127.0.0.1:5432/appdb")
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

class Task(Base):
    __tablename__ = "tasks"
    
    id = Column(String, primary_key=True)
    status = Column(String, default="pending")  # pending, running, completed, failed
    prompt = Column(Text)
    model = Column(String)
    report = Column(Text, nullable=True)
    error = Column(Text, nullable=True)
    created_at = Column(DateTime)
    completed_at = Column(DateTime, nullable=True)

# Create tables
Base.metadata.create_all(bind=engine)

Building a Custom Research Workflow

Example: Multi-Step Research Task

import threading
from uuid import uuid4
from datetime import datetime
from fastapi import FastAPI, BackgroundTasks
from src.planning_agent import planner_agent, executor_agent_step
from src.research_tools import tavily_search_tool, arxiv_search_tool

app = FastAPI()

def run_research_workflow(task_id: str, prompt: str, model: str):
    """Background task that runs the full agentic workflow"""
    db = SessionLocal()
    task = db.query(Task).filter(Task.id == task_id).first()
    
    try:
        # Update status
        task.status = "running"
        db.commit()
        
        # Step 1: Planning
        plan = planner_agent(user_prompt=prompt, model=model)
        
        # Step 2: Execute each step in plan
        context = {}
        for step_info in plan["steps"]:
            result = executor_agent_step(
                step=step_info,
                context=context,
                model=model
            )
            context[step_info["step"]] = result
        
        # Step 3: Store final report
        task.report = context.get("final_report", context.get("edit", ""))
        task.status = "completed"
        task.completed_at = datetime.utcnow()
        
    except Exception as e:
        task.status = "failed"
        task.error = str(e)
    
    finally:
        db.commit()
        db.close()

@app.post("/generate_report")
async def generate_report(
    background_tasks: BackgroundTasks,
    request: dict
):
    task_id = str(uuid4())
    prompt = request["prompt"]
    model = request.get("model", "openai:gpt-4o")
    
    # Create task record
    db = SessionLocal()
    task = Task(
        id=task_id,
        prompt=prompt,
        model=model,
        status="pending",
        created_at=datetime.utcnow()
    )
    db.add(task)
    db.commit()
    db.close()
    
    # Run workflow in background
    thread = threading.Thread(
        target=run_research_workflow,
        args=(task_id, prompt, model)
    )
    thread.start()
    
    return {"task_id": task_id}

Configuration

Environment Variables

# Required
OPENAI_API_KEY=sk-...           # OpenAI API key for LLM calls
TAVILY_API_KEY=tvly-...         # Tavily API for web search

# Optional (defaults provided by entrypoint)
DATABASE_URL=postgresql://app:local@127.0.0.1:5432/appdb
POSTGRES_USER=app
POSTGRES_PASSWORD=local
POSTGRES_DB=appdb

# Development
RESET_DB_ON_STARTUP=0           # Set to 1 to drop tables on startup

Model Selection

The system supports multiple LLM providers via aisuite:

# OpenAI models
model = "openai:gpt-4o"
model = "openai:gpt-4o-mini"

# Anthropic models (if configured)
model = "anthropic:claude-3-sonnet"

# Pass to any agent or planning function
plan = planner_agent(prompt, model="openai:gpt-4o")

Common Patterns

1. Custom Research Tool

def custom_search_tool(query: str) -> list[dict]:
    """Add your own search/retrieval logic"""
    import requests
    
    response = requests.get(
        "https://api.example.com/search",
        params={"q": query},
        headers={"Authorization": f"Bearer {os.getenv('CUSTOM_API_KEY')}"}
    )
    
    return response.json()["results"]

# Register in agents.py
research_output = research_agent(
    topic="AI safety",
    tools=[tavily_search_tool, custom_search_tool],
    model="openai:gpt-4o"
)

2. Streaming Task Updates

@app.get("/task_progress/{task_id}")
async def task_progress(task_id: str):
    """Real-time progress endpoint"""
    db = SessionLocal()
    task = db.query(Task).filter(Task.id == task_id).first()
    
    if not task:
        return {"error": "Task not found"}
    
    # You can store step-by-step progress in a separate table
    # or use a JSONB column for detailed state
    return {
        "status": task.status,
        "current_step": get_current_step(task_id),  # Custom logic
        "steps_completed": get_completed_steps(task_id),
        "message": f"Running {task.status}..."
    }

3. Reflection/Critique Loop

def reflective_research_loop(prompt: str, model: str, max_iterations: int = 3):
    """Iteratively improve research output with reflection"""
    draft = None
    
    for i in range(max_iterations):
        # Research or refine
        if draft is None:
            draft = research_agent(topic=prompt, model=model)
        else:
            # Critique current draft
            critique = editor_agent(
                draft=draft,
                instruction="Identify gaps and areas for improvement",
                model=model
            )
            
            # Research to fill gaps
            additional_research = research_agent(
                topic=critique["gaps"],
                model=model
            )
            
            # Re-write with new info
            draft = writer_agent(
                research_data=draft + additional_research,
                model=model
            )
    
    return draft

Troubleshooting

Container won't start / Postgres issues

# Check logs
docker logs fpsvc

# Common issue: port conflicts
# Solution: change port mapping
docker run -p 8001:8000 -p 5433:5432 --env-file .env fastapi-postgres-service

# Verify Postgres is running inside container
docker exec -it fpsvc bash -lc "pg_isready"

Database connection errors

# Verify DATABASE_URL is set correctly
import os
print(os.getenv("DATABASE_URL"))

# Test connection manually
from sqlalchemy import create_engine
engine = create_engine(os.getenv("DATABASE_URL"))
conn = engine.connect()
print("Connection successful!")
conn.close()

API key not found

# Verify .env file is loaded
docker exec -it fpsvc env | grep API_KEY

# If missing, restart with explicit env vars
docker run --rm -it \
  -p 8000:8000 \
  -e OPENAI_API_KEY=sk-... \
  -e TAVILY_API_KEY=tvly-... \
  fastapi-postgres-service

Tavily rate limits

import time

def tavily_search_with_retry(query: str, max_retries: int = 3):
    """Add retry logic for rate limits"""
    for attempt in range(max_retries):
        try:
            return tavily_search_tool(query)
        except Exception as e:
            if "rate limit" in str(e).lower() and attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            raise

Hot reload for development

# Mount code as volume for live updates
docker run --rm -it \
  -p 8000:8000 -p 5432:5432 \
  -v "$PWD":/app \
  --env-file .env \
  --name fpsvc \
  fastapi-postgres-service \
  bash -lc "
    pg_ctlcluster \$(psql -V | awk '{print \$3}' | cut -d. -f1) main start && \
    uvicorn main:app --host 0.0.0.0 --port 8000 --reload
  "

Connect to database from host

# Use psql client
psql "postgresql://app:local@localhost:5432/appdb"

# Or with explicit connection string
psql -h localhost -p 5432 -U app -d appdb

# List all tasks
SELECT id, status, prompt, created_at FROM tasks;

Reset database

# Add to main.py startup
import os
from sqlalchemy import create_engine

if os.getenv("RESET_DB_ON_STARTUP") == "1":
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)
    print("✅ Database reset complete")

Then run:

docker run -e RESET_DB_ON_STARTUP=1 --env-file .env fastapi-postgres-service

Best Practices

1. Always use environment variables for API keys (never hardcode)
2. Implement retry logic for external API calls (Tavily, OpenAI)
3. Store intermediate results in the database for long-running tasks
4. Use background threads for async workflows (don't block API requests)
5. Log extensively to track agent decision-making
6. Set timeouts for tool calls to prevent hanging tasks
7. Validate user input before passing to agents (injection risks)
8. Monitor token usage when using paid LLM APIs

Advanced Usage

Custom Agent Chain

from src.planning_agent import planner_agent, executor_agent_step

def custom_workflow(prompt: str, model: str):
    """Define your own agent sequence"""
    
    # 1. Break down the task
    plan = planner_agent(prompt, model)
    
    # 2. Add custom pre-processing step
    context = {"original_prompt": prompt}
    
    # 3. Execute with custom logic
    for step_info in plan["steps"]:
        # Add your own conditionals
        if step_info["step"] == "research":
            # Use specialized research logic
            context["research"] = deep_research(step_info, model)
        else:
            # Use default executor
            result = executor_agent_step(step_info, context, model)
            context[step_info["step"]] = result
    
    # 4. Post-process
    final_output = synthesize_results(context)
    return final_output

This skill provides comprehensive knowledge for AI agents to work with the Agentic Research Agent system, from basic setup to advanced customization.