deeplearning-ai-agentic-research

# DeepLearning.AI Agentic Research Agent > Skill by [ara.so](https://ara.so) — AI Agent Skills collection. A FastAPI-based research agent service that orchestrates multi-step research workflows using planning agents, tool-using agents (Tavily, arXiv, Wikipedia), and Postgres for task state management. The system breaks down research queries into planned steps, executes them with specialized agents (researcher, writer, editor), and tracks progress in real-time. ## What This Project Does - **Multi-Agent Research Pipeline**: Planner agent creates workflow → Research agent gathers data → Writer agent drafts → Editor agent refines - **Tool Integration**: Tavily web search, arXiv academic papers, Wikipedia knowledge base - **Task Management**: Postgres-backed task tracking with live progress updates - **REST API**: FastAPI endpoints for kicking off research, polling progress, retrieving results - **Single-Container Deploy**: Runs Postgres + FastAPI in one Docker container for local development ## Installation ### Prerequisites - Docker (Desktop or Engine) - API keys for OpenAI and Tavily ### Environment Setup Create a `.env` file in the project root: ```bash # .env OPENAI_API_KEY=sk-your-openai-key TAVILY_API_KEY=tvly-your-tavily-key ``` ### Build and Run ```bash # 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 container's entrypoint automatically: - Starts Postgres cluster - Creates application user and database - Runs database migrations - Launches FastAPI with Uvicorn ## Key API Endpoints ### Generate Research Report ```python import requests # Start a research task response = requests.post( "http://localhost:8000/generate_report", json={ "prompt": "Large Language Models for scientific discovery", "model": "openai:gpt-4o" } ) task_id = response.json()["task_id"] print(f"Task started: {task_id}") ``` ### Poll Task Progress ```python # Check live progress (substeps, tool calls, etc.) progress = requests.get(f"http://localhost:8000/task_progress/{task_id}") print(progress.json()) # Returns: {"status": "running", "steps": [...], "current_step": "research"} ``` ### Get Final Report ```python # Retrieve completed research report status = requests.get(f"http://localhost:8000/task_status/{task_id}") result = status.json() if result["status"] == "completed": print(result["report"]) ``` ### UI Access - Web Interface: `http://localhost:8000/` - API Docs (Swagger): `http://localhost:8000/docs` ## Project Structure ``` . ├── main.py # FastAPI app, endpoints, DB models ├── src/ │ ├── planning_agent.py # planner_agent(), executor_agent_step() │ ├── agents.py # research_agent, writer_agent, editor_agent │ └── research_tools.py # tavily_search_tool, arxiv_search_tool, wikipedia_search_tool ├── templates/ │ └── index.html # Web UI ├── static/ # CSS/JS assets ├── docker/ │ └── entrypoint.sh # Container startup script ├── requirements.txt ├── Dockerfile └── .env # API keys (not committed) ``` ## Core Patterns ### Creating a Research Tool ```python # src/research_tools.py import os from typing import Dict, Any def tavily_search_tool(query: str, max_results: int = 5) -> Dict[str, Any]: """Search the web using Tavily API""" import requests api_key = os.getenv("TAVILY_API_KEY") if not api_key: return {"error": "TAVILY_API_KEY not set"} response = requests.post( "https://api.tavily.com/search", json={ "api_key": api_key, "query": query, "max_results": max_results } ) return response.json() def arxiv_search_tool(query: str, max_results: int = 5) -> list: """Search arXiv for academic papers""" import requests from xml.etree import ElementTree as ET url = f"http://export.arxiv.org/api/query?search_query=all:{query}&max_results={max_results}" response = requests.get(url) root = ET.fromstring(response.content) papers = [] for entry in root.findall("{http://www.w3.org/2005/Atom}entry"): papers.append({ "title": entry.find("{http://www.w3.org/2005/Atom}title").text, "summary": entry.find("{http://www.w3.org/2005/Atom}summary").text, "link": entry.find("{http://www.w3.org/2005/Atom}id").text }) return papers ``` ### Building a Planning Agent ```python # src/planning_agent.py import os import aisuite as ai def planner_agent(prompt: str, model: str = "openai:gpt-4o") -> dict: """Create a multi-step research plan""" client = ai.Client() planning_prompt = f""" You are a research planning agent. Break down this research task into steps: "{prompt}" Provide a JSON plan with steps like: {{"steps": [ {{"type": "research", "query": "...", "tools": ["tavily", "arxiv"]}}, {{"type": "write", "instruction": "..."}}, {{"type": "edit", "focus": "..."}} ]}} """ response = client.chat.completions.create( model=model, messages=[{"role": "user", "content": planning_prompt}], response_format={"type": "json_object"} ) import json return json.loads(response.choices[0].message.content) def executor_agent_step(step: dict, context: dict) -> dict: """Execute a single planned step""" from src.research_tools import tavily_search_tool, arxiv_search_tool if step["type"] == "research": results = {} for tool in step["tools"]: if tool == "tavily": results["tavily"] = tavily_search_tool(step["query"]) elif tool == "arxiv": results["arxiv"] = arxiv_search_tool(step["query"]) return {"type": "research", "results": results} elif step["type"] == "write": # Use writer_agent to draft content from src.agents import writer_agent return writer_agent(step["instruction"], context) elif step["type"] == "edit": # Use editor_agent to refine from src.agents import editor_agent return editor_agent(context["draft"], step["focus"]) ``` ### Database Models (SQLAlchemy) ```python # main.py from sqlalchemy import Column, String, Text, DateTime, create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import os from datetime import datetime 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" task_id = Column(String, primary_key=True) prompt = Column(Text) model = Column(String) status = Column(String) # pending, running, completed, failed current_step = Column(String) report = Column(Text, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) # Create tables Base.metadata.create_all(bind=engine) ``` ### FastAPI Endpoint Implementation ```python # main.py from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel import uuid import threading app = FastAPI() class ResearchRequest(BaseModel): prompt: str model: str = "openai:gpt-4o" def run_research_workflow(task_id: str, prompt: str, model: str): """Background task that runs the full research workflow""" from src.planning_agent import planner_agent, executor_agent_step db = SessionLocal() task = db.query(Task).filter(Task.task_id == task_id).first() try: # Update status task.status = "running" task.current_step = "planning" db.commit() # Generate plan plan = planner_agent(prompt, model) # Execute steps context = {} for i, step in enumerate(plan["steps"]): task.current_step = f"step_{i}_{step['type']}" db.commit() result = executor_agent_step(step, context) context[f"step_{i}"] = result # Final report task.report = context.get("final_report", "Research completed") task.status = "completed" task.current_step = "done" except Exception as e: task.status = "failed" task.report = f"Error: {str(e)}" finally: db.commit() db.close() @app.post("/generate_report") def generate_report(request: ResearchRequest, background_tasks: BackgroundTasks): """Start a research task""" task_id = str(uuid.uuid4()) # Create DB record db = SessionLocal() task = Task( task_id=task_id, prompt=request.prompt, model=request.model, status="pending" ) db.add(task) db.commit() db.close() # Run in background thread thread = threading.Thread( target=run_research_workflow, args=(task_id, request.prompt, request.model) ) thread.start() return {"task_id": task_id} @app.get("/task_progress/{task_id}") def task_progress(task_id: str): """Get live progress of a task""" db = SessionLocal() task = db.query(Task).filter(Task.task_id == task_id).first() db.close() if not task: return {"error": "Task not found"} return { "task_id": task.task_id, "status": task.status, "current_step": task.current_step, "created_at": task.created_at.isoformat() } @app.get("/task_status/{task_id}") def task_status(task_id: str): """Get final status and report""" db = SessionLocal() task = db.query(Task).filter(Task.task_id == task_id).first() db.close() if not task: return {"error": "Task not found"} return { "task_id": task.task_id, "status": task.status, "report": task.report, "prompt": task.prompt, "model": task.model } ``` ## Configuration ### Database Connection Override the default DATABASE_URL: ```bash docker run --rm -it \ -p 8000:8000 \ -e DATABASE_URL="postgresql://user:pass@host:5432/db" \ --env-file .env \ fastapi-postgres-service ``` ### Postgres Credentials Set custom database credentials: ```bash # In .env or via -e flags POSTGRES_USER=myuser POSTGRES_PASSWORD=mypassword POSTGRES_DB=research_db ``` ### Disable DB Reset on Startup By default, `main.py` may drop tables on startup (dev mode): ```python # main.py import os # Guard the drop operation if os.getenv("RESET_DB_ON_STARTUP") == "1": Base.metadata.drop_all(bind=engine) Base.metadata.create_all(bind=engine) ``` Set `RESET_DB_ON_STARTUP=0` to preserve data between restarts. ## Common Patterns ### Custom Agent Implementation ```python # src/agents.py import os import aisuite as ai