agentic-research-agent-deeplearning

# Agentic Research Agent (DeepLearning.AI) > Skill by [ara.so](https://ara.so) — AI Agent Skills collection. ## Overview The Reflective Research Agent is a FastAPI-based service that orchestrates multi-step research workflows using AI agents. It features: - **Planning Agent**: Breaks down research tasks into executable steps - **Tool-Using Agents**: Research, writer, and editor agents with access to Tavily search, arXiv papers, and Wikipedia - **State Management**: Postgres database tracks task state, progress, and results - **Single Container**: All-in-one Docker image with Postgres + FastAPI - **Live Progress**: Real-time status updates for long-running research tasks The service exposes a web UI and REST API for kicking off research workflows, monitoring progress, and retrieving final reports. ## Installation ### Prerequisites ```bash # Ensure Docker is installed docker --version # Create .env file with API keys cat > .env << EOF OPENAI_API_KEY=your-openai-key TAVILY_API_KEY=your-tavily-key EOF ``` ### Clone and Build ```bash # Clone the repository git clone https://github.com/deeplearning-ai/agentic-ai-public.git cd agentic-ai-public # Build Docker image docker build -t fastapi-postgres-service . ``` ### Run the Service ```bash # Run container with ports exposed docker run --rm -it \ -p 8000:8000 \ -p 5432:5432 \ --name fpsvc \ --env-file .env \ fastapi-postgres-service ``` Access the service: - **Web UI**: http://localhost:8000 - **API Docs**: http://localhost:8000/docs - **Postgres**: postgresql://app:local@localhost:5432/appdb ## Project Structure ``` . ├── main.py # FastAPI application entry point ├── src/ │ ├── planning_agent.py # Planner and executor logic │ ├── agents.py # Research, writer, editor agents │ └── research_tools.py # Tool implementations (Tavily, arXiv, Wikipedia) ├── templates/ │ └── index.html # Web UI template ├── static/ # CSS/JS assets ├── docker/ │ └── entrypoint.sh # Container startup script ├── requirements.txt └── Dockerfile ``` ## Key API Endpoints ### Start Research Task ```bash # Kick off a research workflow 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"} ``` ### Poll Progress ```bash # Get live progress updates curl http://localhost:8000/task_progress/550e8400-e29b-41d4-a716-446655440000 # Response shows current step, substeps, and status { "task_id": "550e8400-e29b-41d4-a716-446655440000", "status": "in_progress", "current_step": "research", "steps_completed": 1, "total_steps": 4, "substeps": [ {"name": "tavily_search", "status": "completed"}, {"name": "arxiv_search", "status": "in_progress"} ] } ``` ### Get Final Report ```bash # Retrieve completed research report curl http://localhost:8000/task_status/550e8400-e29b-41d4-a716-446655440000 # Response includes full report and metadata { "task_id": "550e8400-e29b-41d4-a716-446655440000", "status": "completed", "report": "# Research Report\n\n## Introduction\n...", "created_at": "2024-01-15T10:30:00Z", "completed_at": "2024-01-15T10:35:00Z" } ``` ## Core Code Patterns ### Creating Custom Research Tools ```python # src/research_tools.py import requests from typing import Dict, Any def tavily_search_tool(query: str, max_results: int = 5) -> Dict[str, Any]: """Search using Tavily API""" api_key = os.getenv("TAVILY_API_KEY") response = requests.post( "https://api.tavily.com/search", json={ "query": query, "max_results": max_results, "search_depth": "advanced" }, headers={"Authorization": f"Bearer {api_key}"} ) return response.json() def arxiv_search_tool(query: str, max_results: int = 5) -> list: """Search arXiv papers""" import arxiv search = arxiv.Search( query=query, max_results=max_results, sort_by=arxiv.SortCriterion.Relevance ) return [ { "title": result.title, "summary": result.summary, "authors": [author.name for author in result.authors], "pdf_url": result.pdf_url } for result in search.results() ] ``` ### Building Custom Agents ```python # src/agents.py from aisuite.client import Client def research_agent(prompt: str, tools: list, model: str = "openai:gpt-4o") -> str: """Agent that conducts research using available tools""" client = Client() # Prepare tool descriptions for the agent tool_descriptions = "\n".join([ f"- {tool['name']}: {tool['description']}" for tool in tools ]) system_prompt = f"""You are a research agent. You have access to: {tool_descriptions} Use these tools to gather comprehensive information about the topic. Synthesize findings into a cohesive research summary.""" messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt} ] response = client.chat.completions.create( model=model, messages=messages, temperature=0.7 ) return response.choices[0].message.content def writer_agent(research_content: str, model: str = "openai:gpt-4o") -> str: """Agent that transforms research into a formatted report""" client = Client() messages = [ {"role": "system", "content": "Transform research findings into a well-structured report with sections, headings, and citations."}, {"role": "user", "content": research_content} ] response = client.chat.completions.create( model=model, messages=messages, temperature=0.5 ) return response.choices[0].message.content def editor_agent(draft_report: str, model: str = "openai:gpt-4o") -> str: """Agent that refines and polishes the report""" client = Client() messages = [ {"role": "system", "content": "Review and improve the report for clarity, coherence, and professionalism."}, {"role": "user", "content": draft_report} ] response = client.chat.completions.create( model=model, messages=messages, temperature=0.3 ) return response.choices[0].message.content ``` ### Implementing Planning Agent ```python # src/planning_agent.py import json from aisuite.client import Client def planner_agent(prompt: str, model: str = "openai:gpt-4o") -> dict: """Create a multi-step plan for research workflow""" client = Client() system_prompt = """You are a research planning agent. Create a step-by-step plan to complete the research task. Output JSON with structure: { "steps": [ {"name": "research", "description": "...", "tools": ["tavily", "arxiv"]}, {"name": "write", "description": "..."}, {"name": "edit", "description": "..."} ] }""" messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": f"Plan research for: {prompt}"} ] response = client.chat.completions.create( model=model, messages=messages, temperature=0.7 ) return json.loads(response.choices[0].message.content) def executor_agent_step(step: dict, context: dict, model: str) -> str: """Execute a single step from the plan""" step_name = step["name"] if step_name == "research": # Execute research with tools results = [] if "tavily" in step.get("tools", []): results.append(tavily_search_tool(context["prompt"])) if "arxiv" in step.get("tools", []): results.append(arxiv_search_tool(context["prompt"])) return research_agent(context["prompt"], results, model) elif step_name == "write": return writer_agent(context["research_results"], model) elif step_name == "edit": return editor_agent(context["draft_report"], model) return "" ``` ### Database Models and State Management ```python # main.py or models.py from sqlalchemy import create_engine, Column, String, Text, DateTime, Enum from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import enum from datetime import datetime import os Base = declarative_base() class TaskStatus(enum.Enum): PENDING = "pending" IN_PROGRESS = "in_progress" COMPLETED = "completed" FAILED = "failed" class ResearchTask(Base): __tablename__ = "research_tasks" task_id = Column(String, primary_key=True) prompt = Column(Text, nullable=False) model = Column(String, nullable=False) status = Column(Enum(TaskStatus), default=TaskStatus.PENDING) current_step = Column(String) report = Column(Text) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) completed_at = Column(DateTime) # Initialize database DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://app:local@127.0.0.1:5432/appdb") engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(bind=engine) # Create tables Base.metadata.create_all(bind=engine) ``` ### FastAPI Application Setup ```python # main.py from fastapi import FastAPI, BackgroundTasks, HTTPException from fastapi.templating import Jinja2Templates from fastapi.staticfiles import StaticFiles from pydantic import BaseModel import uuid import threading app = FastAPI(title="Reflective Research Agent") # Mount templates and static files templates = Jinja2Templates(directory="templates") app.mount("/static", StaticFiles(directory="static"), name="static") class ResearchRequest(BaseModel): prompt: str model: str = "openai:gpt-4o" @app.get("/") async def home(request: Request): """Render web UI""" return templates.TemplateResponse("index.html", {"request": request}) @app.post("/generate_report") async def generate_report(req: ResearchRequest, background_tasks: BackgroundTasks): """Kick off research workflow in background""" task_id = str(uuid.uuid4()) # Create task in database db = SessionLocal() task = ResearchTask( task_id=task_id, prompt=req.prompt, model=req.model, status=TaskStatus.PENDING ) db.add(task) db.commit() db.close() # Start workflow in thread thread = threading.Thread(target=run_workflow, args=(task_id, req.prompt, req.model)) thread.start() return {"task_id": task_id} def run_workflow(task_id: str, prompt: str, model: str): """Execute the full research workflow""" db = SessionLocal() task = db.query(ResearchTask).filter_by(task_id=task_id).first() try: # Update status task.status = TaskStatus.IN_PROGRESS db.commit() # Step 1: Plan plan = planner_agent(prompt, model) # Step 2: Execute plan context = {"prompt": prompt} for step in plan["steps"]: task.current_step = step["name"] db.commit() result = executor_agent_step(step, context, model) context[f"{step['name']}_results"] = result # Step 3: Store final report task.report = context.get("edit_results", "")