Capstone — Build an Autonomous Research Agent

research_agent/
├── agent.py          # LangGraph state machine
├── tools.py          # Search, lookup, and analysis tools
├── persistence.py    # SQLite conversation history
├── reporter.py       # Report formatting
├── main.py           # Entry point
└── requirements.txt

# script_id: day_048_capstone_autonomous_research_agent/research_agent
# state.py
from typing import TypedDict, List, Optional, Annotated
from langgraph.graph.message import add_messages


class ResearchState(TypedDict):
    """The complete state of the research agent."""
    # The research topic
    topic: str

    # Conversation messages (managed by LangGraph)
    messages: Annotated[list, add_messages]

    # Structured research data
    search_queries: List[str]        # Queries we've run
    findings: List[dict]             # Raw findings from searches
    synthesis: str                   # Running synthesis of findings
    identified_gaps: List[str]       # Gaps identified in research

    # Control flow
    iteration_count: int
    max_iterations: int
    research_complete: bool

    # Output
    final_report: Optional[str]
    session_id: str

# script_id: day_048_capstone_autonomous_research_agent/research_agent
# tools.py
import json
import random
from datetime import datetime
from langchain_core.tools import tool


@tool
def web_search(query: str, num_results: int = 5) -> str:
    """
    Search the web for information on a topic.
    Returns a JSON list of results with title, url, and snippet.
    
    Args:
        query: The search query
        num_results: Number of results to return (1-10)
    """
    # In production: replace with Tavily, SerpAPI, or Brave Search API
    # from tavily import TavilyClient
    # client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
    # results = client.search(query, max_results=num_results)

    # Mock implementation for development
    mock_results = [
        {
            "title": f"Research Result: {query}",
            "url": f"https://example.com/article-{i}",
            "snippet": f"This article discusses {query} in depth, covering key aspects including implementation details, best practices, and real-world applications. Published {datetime.now().strftime('%B %Y')}.",
            "published_date": datetime.now().isoformat(),
        }
        for i in range(min(num_results, 5))
    ]

    return json.dumps(mock_results, indent=2)


@tool
def fetch_article(url: str) -> str:
    """
    Fetch and extract the full text of an article from a URL.
    Use this when a search snippet is not detailed enough.
    
    Args:
        url: The URL of the article to fetch
    """
    # In production: use requests + BeautifulSoup or a content extraction API
    # import requests
    # from bs4 import BeautifulSoup
    # response = requests.get(url, timeout=10)
    # soup = BeautifulSoup(response.content, 'html.parser')
    # return soup.get_text(separator='\n', strip=True)[:5000]

    return f"[Mock article content for {url}]\n\nThis is detailed content about the topic. It covers background, current state, key challenges, and future directions. The article cites several studies and expert opinions."


@tool
def analyze_findings(findings_json: str) -> str:
    """
    Analyze a collection of research findings to identify:
    - Key themes
    - Consensus points
    - Contradictions
    - Information gaps
    
    Args:
        findings_json: JSON string of findings list
    """
    findings = json.loads(findings_json)
    return json.dumps({
        "finding_count": len(findings),
        "key_themes": ["theme 1", "theme 2"],  # LLM will derive real themes
        "has_contradictions": False,
        "suggested_gaps": ["Consider researching X", "More data needed on Y"],
    })


@tool
def save_note(note: str, category: str = "general") -> str:
    """
    Save an important note or finding for inclusion in the final report.
    Use this to bookmark key insights as you research.
    
    Args:
        note: The note to save
        category: Category label (e.g., 'key_finding', 'statistic', 'quote')
    """
    # In a real system, persist this to a database
    timestamp = datetime.now().isoformat()
    return f"Note saved at {timestamp}: [{category}] {note[:100]}..."


RESEARCH_TOOLS = [web_search, fetch_article, analyze_findings, save_note]

# script_id: day_048_capstone_autonomous_research_agent/research_agent
# agent.py
import json
import os
from typing import Literal
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

from state import ResearchState
from tools import RESEARCH_TOOLS

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
llm_with_tools = llm.bind_tools(RESEARCH_TOOLS)

RESEARCH_SYSTEM_PROMPT = """You are an expert research agent. Your job is to thoroughly research a topic and produce a comprehensive report.

Research Strategy:
1. Start by planning 3-5 specific search queries that will cover the topic from different angles
2. Execute searches and analyze the results carefully
3. Identify gaps in your knowledge and search for more specific information
4. Synthesize findings into coherent insights
5. When you have sufficient information (typically 5-10 searches), produce a final report

Report Structure:
- Executive Summary (2-3 sentences)
- Background & Context
- Key Findings (bullet points)
- Analysis & Implications
- Open Questions / Areas for Further Research
- Sources Consulted

Important:
- Use tools actively — don't try to answer from memory
- Track what you've searched and avoid redundant queries
- When you have enough information, signal completion by including "RESEARCH_COMPLETE" in your response
- Stay focused on the topic — don't go down rabbit holes"""


def should_continue(state: ResearchState) -> Literal["tools", "end"]:
    """Routing logic: continue with tools or end?"""
    messages = state["messages"]
    last_message = messages[-1]

    # Check max iterations
    if state["iteration_count"] >= state["max_iterations"]:
        return "end"

    # Check if research is marked complete
    if state.get("research_complete"):
        return "end"

    # Check if the last message has tool calls
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"

    # Check if the assistant signaled completion
    if isinstance(last_message, AIMessage) and "RESEARCH_COMPLETE" in (last_message.content or ""):
        return "end"

    # Default: ask the agent to continue
    return "tools"


def research_node(state: ResearchState) -> dict:
    """Main agent node: think, plan, and decide what tools to call."""
    iteration = state.get("iteration_count", 0)

    # Build context message about current progress
    progress_context = f"""
Current research progress:
- Topic: {state['topic']}
- Iteration: {iteration + 1}/{state['max_iterations']}
- Searches completed: {len(state.get('search_queries', []))}
- Findings collected: {len(state.get('findings', []))}

{'IMPORTANT: You are running low on iterations. If you have enough information, produce the final report now.' if iteration >= state['max_iterations'] - 3 else ''}
"""

    messages = [
        SystemMessage(content=RESEARCH_SYSTEM_PROMPT + "\n\n" + progress_context),
        *state["messages"],
    ]

    response = llm_with_tools.invoke(messages)

    # Check if research is complete
    research_complete = "RESEARCH_COMPLETE" in (response.content or "")

    return {
        "messages": [response],
        "iteration_count": iteration + 1,
        "research_complete": research_complete,
    }


def report_node(state: ResearchState) -> dict:
    """Generate the final structured report."""
    all_tool_results = []
    for msg in state["messages"]:
        if hasattr(msg, "type") and msg.type == "tool":
            all_tool_results.append(msg.content)

    report_prompt = f"""Based on all the research conducted, produce a final comprehensive report on: {state['topic']}

Research findings collected:
{chr(10).join(all_tool_results[:10])}  

Format the report with clear sections:
1. Executive Summary
2. Background & Context  
3. Key Findings
4. Analysis & Implications
5. Open Questions
6. Sources Consulted

Be thorough but concise. This is the deliverable."""

    response = llm.invoke([
        SystemMessage(content="You are an expert research analyst producing a final report."),
        HumanMessage(content=report_prompt),
    ])

    return {
        "messages": [response],
        "final_report": response.content,
    }


def build_graph() -> StateGraph:
    """Assemble the LangGraph state machine."""
    graph = StateGraph(ResearchState)

    # Add nodes
    graph.add_node("researcher", research_node)
    graph.add_node("tools", ToolNode(RESEARCH_TOOLS))
    graph.add_node("reporter", report_node)

    # Set entry point
    graph.set_entry_point("researcher")

    # Conditional routing from researcher
    graph.add_conditional_edges(
        "researcher",
        should_continue,
        {
            "tools": "tools",
            "end": "reporter",
        },
    )

    # After tools, always go back to researcher
    graph.add_edge("tools", "researcher")

    # Reporter is the final node
    graph.add_edge("reporter", END)

    return graph.compile()

# script_id: day_048_capstone_autonomous_research_agent/research_agent
# persistence.py
import sqlite3
import json
from datetime import datetime
from pathlib import Path

DB_PATH = Path("./research_sessions.db")


def init_db():
    """Initialize the SQLite database."""
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS sessions (
            session_id TEXT PRIMARY KEY,
            topic TEXT NOT NULL,
            status TEXT DEFAULT 'in_progress',
            created_at TEXT,
            updated_at TEXT,
            final_report TEXT,
            iteration_count INTEGER DEFAULT 0,
            metadata TEXT DEFAULT '{}'
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS messages (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            session_id TEXT NOT NULL,
            role TEXT NOT NULL,
            content TEXT,
            tool_calls TEXT,
            created_at TEXT,
            FOREIGN KEY (session_id) REFERENCES sessions(session_id)
        )
    """)
    conn.commit()
    conn.close()


def save_session(session_id: str, topic: str, status: str = "in_progress",
                 final_report: str = None, iteration_count: int = 0):
    """Save or update a research session."""
    conn = sqlite3.connect(DB_PATH)
    now = datetime.now().isoformat()
    conn.execute("""
        INSERT INTO sessions (session_id, topic, status, created_at, updated_at, final_report, iteration_count)
        VALUES (?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(session_id) DO UPDATE SET
            status=excluded.status,
            updated_at=excluded.updated_at,
            final_report=excluded.final_report,
            iteration_count=excluded.iteration_count
    """, (session_id, topic, status, now, now, final_report, iteration_count))
    conn.commit()
    conn.close()


def load_session(session_id: str) -> dict | None:
    """Load a research session by ID."""
    conn = sqlite3.connect(DB_PATH)
    row = conn.execute(
        "SELECT * FROM sessions WHERE session_id = ?", (session_id,)
    ).fetchone()
    conn.close()

    if not row:
        return None

    columns = ["session_id", "topic", "status", "created_at", "updated_at",
               "final_report", "iteration_count", "metadata"]
    return dict(zip(columns, row))


def list_sessions() -> list:
    """List all research sessions."""
    conn = sqlite3.connect(DB_PATH)
    rows = conn.execute(
        "SELECT session_id, topic, status, created_at, iteration_count FROM sessions ORDER BY created_at DESC"
    ).fetchall()
    conn.close()
    return [{"session_id": r[0], "topic": r[1], "status": r[2], "created_at": r[3], "iterations": r[4]} for r in rows]

# script_id: day_048_capstone_autonomous_research_agent/research_agent
# main.py
import uuid
from langchain_core.messages import HumanMessage
from agent import build_graph
from persistence import init_db, save_session, list_sessions
from state import ResearchState

init_db()


def run_research(topic: str, max_iterations: int = 15) -> str:
    """
    Run a full research session on a topic.
    Returns the final report as a string.
    """
    session_id = str(uuid.uuid4())[:8]
    print(f"\nStarting research session {session_id}")
    print(f"Topic: {topic}")
    print(f"Max iterations: {max_iterations}")
    print("-" * 60)

    graph = build_graph()

    initial_state: ResearchState = {
        "topic": topic,
        "messages": [HumanMessage(content=f"Please research this topic thoroughly: {topic}")],
        "search_queries": [],
        "findings": [],
        "synthesis": "",
        "identified_gaps": [],
        "iteration_count": 0,
        "max_iterations": max_iterations,
        "research_complete": False,
        "final_report": None,
        "session_id": session_id,
    }

    # Save initial session
    save_session(session_id, topic, status="in_progress")

    # Run the graph with streaming to see progress
    final_state = None
    # Use `stream_mode="debug"` during development for detailed execution traces
    # showing every node entry/exit and state change.
    for event in graph.stream(initial_state, stream_mode="values"):
        final_state = event
        iteration = event.get("iteration_count", 0)
        last_message = event["messages"][-1] if event["messages"] else None

        if last_message and hasattr(last_message, "tool_calls") and last_message.tool_calls:
            for tc in last_message.tool_calls:
                print(f"  [Iteration {iteration}] Tool: {tc['name']}({list(tc['args'].keys())})")

    # Save completed session
    report = final_state.get("final_report", "No report generated")
    save_session(
        session_id,
        topic,
        status="completed",
        final_report=report,
        iteration_count=final_state.get("iteration_count", 0),
    )

    print(f"\nSession {session_id} complete after {final_state.get('iteration_count', 0)} iterations.")
    return report


if __name__ == "__main__":
    # Example research topics
    topics = [
        "The current state of AI agent frameworks in 2025",
        "Best practices for RAG system evaluation",
    ]

    report = run_research(topics[0], max_iterations=12)

    print("\n" + "=" * 60)
    print("FINAL REPORT")
    print("=" * 60)
    print(report)

    print("\n" + "=" * 60)
    print("ALL SESSIONS")
    print("=" * 60)
    for session in list_sessions():
        print(f"  {session['session_id']}: {session['topic'][:50]}... ({session['status']}, {session['iterations']} iterations)")

# requirements.txt
openai>=1.30.0
langchain>=0.2.0
langchain-openai>=0.1.0
langgraph>=0.1.0

pip install -r requirements.txt
export OPENAI_API_KEY="sk-..."

python main.py

Starting research session a3f9b2c1
Topic: The current state of AI agent frameworks in 2025
Max iterations: 12
------------------------------------------------------------
  [Iteration 1] Tool: web_search(['query'])
  [Iteration 2] Tool: web_search(['query'])
  [Iteration 3] Tool: fetch_article(['url'])
  [Iteration 4] Tool: save_note(['note', 'category'])
  ...
  [Iteration 9] Tool: analyze_findings(['findings_json'])

Session a3f9b2c1 complete after 11 iterations.

============================================================
FINAL REPORT
============================================================
Executive Summary
...