agentic-ai-research-agent

# Agentic AI Research Agent > Skill by [ara.so](https://ara.so) — AI Agent Skills collection. ## Overview The Agentic AI Research Agent is a FastAPI-based service that implements a reflective, multi-step research workflow. It uses planning agents to break down research tasks, executes specialized agents (research, writer, editor) with external tools (Tavily search, arXiv papers, Wikipedia), and stores task state/results in Postgres. The system provides real-time progress tracking and generates comprehensive research reports. **Key Features:** - Multi-step agent planning and execution - Tool-using agents with Tavily, arXiv, Wikipedia integration - Postgres-backed task state and result storage - Real-time progress tracking via REST API - Web UI for launching research tasks - Single-container Docker deployment with embedded Postgres ## Installation ### Prerequisites - Docker (Desktop or Engine) - OpenAI API key - Tavily API key (for web search) ### Environment Setup Create a `.env` file in the project root: ```bash # Required API keys OPENAI_API_KEY=sk-... TAVILY_API_KEY=tvly-... # Optional: Override database settings # POSTGRES_USER=app # POSTGRES_PASSWORD=local # POSTGRES_DB=appdb # DATABASE_URL=postgresql://app:local@127.0.0.1:5432/appdb ``` ### Docker Build and Run ```bash # Build the image docker build -t fastapi-postgres-service . # Run the container (foreground with logs) docker run --rm -it \ -p 8000:8000 \ -p 5432:5432 \ --name fpsvc \ --env-file .env \ fastapi-postgres-service # Run in detached mode docker run -d \ -p 8000:8000 \ -p 5432:5432 \ --name fpsvc \ --env-file .env \ fastapi-postgres-service ``` ### Access Points - Web UI: http://localhost:8000/ - API Docs: http://localhost:8000/docs - Database: `postgresql://app:local@localhost:5432/appdb` ## Project Structure ``` . ├─ main.py # FastAPI app with endpoints ├─ src/ │ ├─ planning_agent.py # Planner and executor agents │ ├─ agents.py # Research, writer, editor agents │ └─ research_tools.py # Tavily, arXiv, Wikipedia tools ├─ templates/ │ └─ index.html # Web UI template ├─ static/ # CSS/JS assets ├─ docker/ │ └─ entrypoint.sh # Postgres + Uvicorn startup ├─ requirements.txt ├─ Dockerfile └─ README.md ``` ## Core API Endpoints ### Generate Research Report ```python # POST /generate_report import requests 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 ID: {task_id}") ``` ### Poll Task Progress ```python # GET /task_progress/{task_id} import requests import time def poll_progress(task_id): while True: response = requests.get(f"http://localhost:8000/task_progress/{task_id}") data = response.json() print(f"Status: {data['status']}") print(f"Current step: {data.get('current_step', 'N/A')}") print(f"Progress: {data.get('progress_pct', 0)}%") if data["status"] in ["completed", "failed"]: break time.sleep(2) return data progress = poll_progress(task_id) ``` ### Get Final Report ```python # GET /task_status/{task_id} import requests response = requests.get(f"http://localhost:8000/task_status/{task_id}") result = response.json() if result["status"] == "completed": print("Final Report:") print(result["final_output"]) else: print(f"Task failed: {result.get('error')}") ``` ## Building Custom Agents ### Creating a Research Tool ```python # src/research_tools.py import os import requests def tavily_search_tool(query: str, max_results: int = 5) -> dict: """Search the web using Tavily API.""" api_key = os.getenv("TAVILY_API_KEY") if not api_key: return {"error": "TAVILY_API_KEY not set"} url = "https://api.tavily.com/search" payload = { "api_key": api_key, "query": query, "max_results": max_results } try: response = requests.post(url, json=payload, timeout=30) response.raise_for_status() return response.json() except Exception as e: return {"error": str(e)} def arxiv_search_tool(query: str, max_results: int = 5) -> list: """Search arXiv for research papers.""" import arxiv try: search = arxiv.Search( query=query, max_results=max_results, sort_by=arxiv.SortCriterion.Relevance ) results = [] for paper in search.results(): results.append({ "title": paper.title, "authors": [a.name for a in paper.authors], "summary": paper.summary, "published": paper.published.isoformat(), "pdf_url": paper.pdf_url }) return results except Exception as e: return [{"error": str(e)}] def wikipedia_search_tool(query: str) -> dict: """Search Wikipedia and return summary.""" import wikipedia try: # Search for the topic search_results = wikipedia.search(query, results=3) if not search_results: return {"error": "No results found"} # Get the first page page = wikipedia.page(search_results[0], auto_suggest=False) return { "title": page.title, "summary": page.summary, "url": page.url } except wikipedia.exceptions.DisambiguationError as e: return {"error": f"Disambiguation: {e.options[:5]}"} except Exception as e: return {"error": str(e)} ``` ### Implementing a Research Agent ```python # src/agents.py import os from src.research_tools import tavily_search_tool, arxiv_search_tool, wikipedia_search_tool def research_agent(query: str, tools: list = None) -> dict: """ Research agent that gathers information using multiple tools. """ if tools is None: tools = ["tavily", "arxiv", "wikipedia"] results = { "query": query, "sources": [] } # Use Tavily for web search if "tavily" in tools: tavily_results = tavily_search_tool(query, max_results=5) if "error" not in tavily_results: results["sources"].append({ "tool": "tavily", "data": tavily_results }) # Use arXiv for academic papers if "arxiv" in tools: arxiv_results = arxiv_search_tool(query, max_results=5) results["sources"].append({ "tool": "arxiv", "data": arxiv_results }) # Use Wikipedia for general knowledge if "wikipedia" in tools: wiki_results = wikipedia_search_tool(query) results["sources"].append({ "tool": "wikipedia", "data": wiki_results }) return results def writer_agent(research_data: dict, style: str = "academic") -> str: """ Writer agent that creates content from research data. """ # This would typically use an LLM to synthesize the research # into a coherent report. Example structure: import aisuite as ai client = ai.Client() prompt = f""" Based on the following research data, write a {style} report: {research_data} Structure the report with: 1. Executive Summary 2. Key Findings 3. Detailed Analysis 4. Conclusions """ messages = [{"role": "user", "content": prompt}] response = client.chat.completions.create( model="openai:gpt-4o", messages=messages ) return response.choices[0].message.content def editor_agent(draft: str, feedback: str = None) -> str: """ Editor agent that refines and improves the draft. """ import aisuite as ai client = ai.Client() prompt = f""" Edit and improve the following draft. Focus on: - Clarity and conciseness - Logical flow - Grammar and style - Factual accuracy {f'Specific feedback to address: {feedback}' if feedback else ''} Draft: {draft} """ messages = [{"role": "user", "content": prompt}] response = client.chat.completions.create( model="openai:gpt-4o", messages=messages ) return response.choices[0].message.content ``` ### Planning Agent Implementation ```python # src/planning_agent.py import json from typing import List, Dict def planner_agent(user_prompt: str, model: str = "openai:gpt-4o") -> List[Dict]: """ Plans a multi-step research workflow based on user prompt. Returns a list of steps with agent assignments and parameters. """ import aisuite as ai client = ai.Client() planning_prompt = f""" Create a step-by-step research plan for the following task: "{user_prompt}" Break it into concrete steps. For each step, specify: - step_id: unique identifier - agent: which agent to use (research_agent, writer_agent, editor_agent) - action: brief description - parameters: dict of parameters for the agent - dependencies: list of step_ids this step depends on Return ONLY a JSON array of steps. """ messages = [{"role": "user", "content": planning_prompt}] response = client.chat.completions.create( model=model, messages=messages, temperature=0.3 ) # Parse the plan plan_text = response.choices[0].message.content try: # Extract JSON from markdown code blocks if present if "```json" in plan_text: plan_text = plan_text.split("```json")[1].split("```")[0] elif "```" in plan_text: plan_text = plan_text.split("```")[1].split("```")[0] plan = json.loads(plan_text.strip()) return plan except json.JSONDecodeError: # Fallback: create a simple default plan return [ { "step_id": "research", "agent": "research_agent", "action": "Gather information", "parameters": {"query": user_prompt, "tools": ["tavily", "arxiv", "wikipedia"]}, "dependencies": [] }, { "step_id": "write", "agent": "writer_agent", "action": "Write initial draft", "parameters": {"style": "academic"}, "dependencies": ["research"] }, { "step_id": "edit", "agent": "editor_agent", "action": "Edit and refine", "parameters": {}, "dependencies": ["write"] } ] def executor_agent_step(step: Dict, step_results: Dict) -> any: """ Execute a single step in the plan. Uses step_results to access outputs from dependency steps. """ from src.agents import research_agent, writer_agent, editor_agent agent_name = step["agent"] parameters = step.get("parameters", {}) dependencies = step.get("dependencies", []) # Inject dependency results into parameters for dep_id in dependencies: if dep_id in step_results: parameters[f"{dep_id}_output"] = step_results[dep_id] # Execute the appropriate agent if agent_name == "research_agent": return research_agent(**parameters) elif agent_name == "writer_agent": # Use research output if available research_data = parameters.get("research_output", {}) return writer_agent(research_data, style=parameters.get("style", "academic")) elif agent_name == "editor_agent": # Use writer output if available draft = parameters.get("write_output", "") return editor_agent(draft, feedback=parameters.get("feedback")) else: raise ValueError(f"Unknown agent: {agent_name}") ``` ## Database Models and Task Management ### SQLAlchemy Models ```python # main.py or models.py from sqlalchemy import Column, String, Text, DateTime, Float, JSON, create_engine