agentic-research-workflow

# Agentic Research Workflow > Skill by [ara.so](https://ara.so) — AI Agent Skills collection. ## Overview The Agentic Research Workflow is a FastAPI-based service that implements a reflective, multi-step research agent system. It orchestrates planning, research, writing, and editing agents that work together to generate comprehensive research reports. The system uses Postgres for state management and supports tool-calling agents that can query Tavily (web search), arXiv (academic papers), and Wikipedia. **Key capabilities:** - Multi-agent workflow orchestration (planner → research → writer → editor) - Tool-using agents with external API integration - Task state tracking and progress monitoring via REST API - Threaded, non-blocking execution - Web UI for task submission and monitoring - Single-container Docker deployment with Postgres ## Installation ### Docker Setup (Recommended) 1. **Clone and prepare environment:** ```bash git clone https://github.com/https-deeplearning-ai/agentic-ai-public.git cd agentic-ai-public ``` 2. **Create `.env` file with required API keys:** ```bash cat > .env << EOF OPENAI_API_KEY=your_openai_key TAVILY_API_KEY=your_tavily_key EOF ``` 3. **Build Docker image:** ```bash docker build -t fastapi-postgres-service . ``` 4. **Run the service:** ```bash docker run --rm -it \ -p 8000:8000 \ -p 5432:5432 \ --name fpsvc \ --env-file .env \ fastapi-postgres-service ``` The service will be available at `http://localhost:8000`. ### Local Development Setup ```bash # Install dependencies pip install -r requirements.txt # Set environment variables export DATABASE_URL="postgresql://app:local@127.0.0.1:5432/appdb" export OPENAI_API_KEY="your_openai_key" export TAVILY_API_KEY="your_tavily_key" # Start Postgres (if not using Docker) # Ensure Postgres is running and database 'appdb' exists # Run the FastAPI app uvicorn main:app --host 0.0.0.0 --port 8000 --reload ``` ## Project Structure ``` . ├── main.py # FastAPI application and endpoints ├── src/ │ ├── planning_agent.py # Planner and executor agent logic │ ├── agents.py # Research, writer, editor agents │ └── research_tools.py # Tool definitions (Tavily, arXiv, Wikipedia) ├── templates/ │ └── index.html # Web UI template ├── static/ # CSS/JS assets ├── docker/ │ └── entrypoint.sh # Docker startup script ├── requirements.txt ├── Dockerfile └── README.md ``` ## Core API Endpoints ### 1. Generate Research Report ```bash POST /generate_report ``` **Request body:** ```json { "prompt": "Large Language Models for scientific discovery", "model": "openai:gpt-4o" } ``` **Response:** ```json { "task_id": "550e8400-e29b-41d4-a716-446655440000" } ``` **Python example:** ```python import requests response = requests.post( "http://localhost:8000/generate_report", json={ "prompt": "Impact of climate change on marine ecosystems", "model": "openai:gpt-4o" } ) task_id = response.json()["task_id"] print(f"Task started: {task_id}") ``` ### 2. Check Task Progress ```bash GET /task_progress/{task_id} ``` **Response:** ```json { "task_id": "550e8400-e29b-41d4-a716-446655440000", "status": "running", "current_step": "research", "steps": [ { "step": "planning", "status": "completed", "substeps": [...] }, { "step": "research", "status": "running", "substeps": [ {"name": "tavily_search", "status": "completed"}, {"name": "arxiv_search", "status": "running"} ] } ] } ``` **Python polling example:** ```python import requests import time task_id = "550e8400-e29b-41d4-a716-446655440000" while True: response = requests.get(f"http://localhost:8000/task_progress/{task_id}") data = response.json() print(f"Status: {data['status']} - Current step: {data.get('current_step', 'N/A')}") if data['status'] in ['completed', 'failed']: break time.sleep(2) ``` ### 3. Get Final Report ```bash GET /task_status/{task_id} ``` **Response:** ```json { "task_id": "550e8400-e29b-41d4-a716-446655440000", "status": "completed", "report": "# Research Report\n\n## Introduction...", "created_at": "2025-01-15T10:30:00", "updated_at": "2025-01-15T10:35:00" } ``` **Python example:** ```python import requests task_id = "550e8400-e29b-41d4-a716-446655440000" response = requests.get(f"http://localhost:8000/task_status/{task_id}") data = response.json() if data['status'] == 'completed': print("Report generated successfully:") print(data['report']) else: print(f"Task status: {data['status']}") ``` ## Building Custom Agents ### Research Tool Implementation ```python # src/research_tools.py import requests import os def tavily_search_tool(query: str, max_results: int = 5) -> list: """ Search the web using Tavily API. Args: query: Search query string max_results: Maximum number of results to return Returns: List of search results with title, url, and snippet """ api_key = os.getenv("TAVILY_API_KEY") if not api_key: raise ValueError("TAVILY_API_KEY not set") response = requests.post( "https://api.tavily.com/search", json={ "api_key": api_key, "query": query, "max_results": max_results, "search_depth": "advanced" } ) response.raise_for_status() return response.json().get("results", []) def arxiv_search_tool(query: str, max_results: int = 5) -> list: """ Search arXiv for academic papers. Args: query: Search query max_results: Maximum papers to retrieve Returns: List of papers with title, authors, summary, and pdf_url """ import arxiv 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": [author.name for author in paper.authors], "summary": paper.summary, "pdf_url": paper.pdf_url, "published": paper.published.isoformat() }) return results def wikipedia_search_tool(query: str) -> dict: """ Search Wikipedia and return summary. Args: query: Topic to search Returns: Dictionary with title, summary, and url """ import wikipedia try: page = wikipedia.page(query, auto_suggest=True) return { "title": page.title, "summary": page.summary, "url": page.url } except wikipedia.exceptions.DisambiguationError as e: # Return first suggestion page = wikipedia.page(e.options[0]) return { "title": page.title, "summary": page.summary, "url": page.url } except wikipedia.exceptions.PageError: return {"error": f"No Wikipedia page found for '{query}'"} ``` ### Agent Implementation Pattern ```python # src/agents.py import aisuite as ai client = ai.Client() def research_agent(topic: str, model: str = "openai:gpt-4o") -> dict: """ Research agent that uses multiple tools to gather information. Args: topic: Research topic model: LLM model to use Returns: Dictionary with research findings """ from src.research_tools import ( tavily_search_tool, arxiv_search_tool, wikipedia_search_tool ) # Define available tools tools = [ { "type": "function", "function": { "name": "tavily_search", "description": "Search the web for current information", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Search query"} }, "required": ["query"] } } }, { "type": "function", "function": { "name": "arxiv_search", "description": "Search arXiv for academic papers", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Research topic"} }, "required": ["query"] } } }, { "type": "function", "function": { "name": "wikipedia_search", "description": "Get Wikipedia summary on a topic", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Topic name"} }, "required": ["query"] } } } ] messages = [ { "role": "system", "content": "You are a research assistant. Use available tools to gather comprehensive information." }, { "role": "user", "content": f"Research the following topic and provide comprehensive findings: {topic}" } ] # Initial call response = client.chat.completions.create( model=model, messages=messages, tools=tools, tool_choice="auto" ) # Handle tool calls tool_results = [] while response.choices[0].message.tool_calls: for tool_call in response.choices[0].message.tool_calls: function_name = tool_call.function.name arguments = eval(tool_call.function.arguments) # Execute tool if function_name == "tavily_search": result = tavily_search_tool(arguments["query"]) elif function_name == "arxiv_search": result = arxiv_search_tool(arguments["query"]) elif function_name == "wikipedia_search": result = wikipedia_search_tool(arguments["query"]) else: result = {"error": "Unknown tool"} tool_results.append({ "tool": function_name, "query": arguments.get("query"), "result": result }) # Add tool response to messages messages.append({ "role": "assistant", "content": None, "tool_calls": [tool_call] }) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": str(result) }) # Continue conversation response = client.chat.completions.create( model=model, messages=messages, tools=tools ) return { "findings": response.choices[0].message.content, "tool_results": tool_results } def writer_agent(research_data: dict, model: str = "openai:gpt-4o") -> str: """ Writer agent that creates a structured report from research findings. Args: