Building a Research Assistant Agent with LangGraph: Step-by-Step Tutorial

What is a Research Assistant Agent?

A Research Assistant Agent is an autonomous LLM-powered system that can perform multi-step research tasks: it understands a complex question, searches for information across multiple sources, synthesizes findings, and iteratively refines its knowledge until it can produce a well-structured answer. Instead of a single prompt-response cycle, the agent plans, acts, observes, and reflects—much like a human researcher.

LangGraph is a library from the LangChain ecosystem designed specifically for building such stateful, multi-actor agent workflows. It models the agent as a directed graph where nodes are computation steps (LLM calls, tool executions, or custom functions) and edges route the flow based on conditional logic or fixed transitions. This graph-based approach gives you fine-grained control over the agent’s execution loop, making it easy to add persistence, human-in-the-loop interrupts, streaming, and complex branching logic.

Why Use LangGraph for a Research Agent?

Traditional agent frameworks often abstract the control flow behind a high-level AgentExecutor. LangGraph makes the flow explicit as a graph, offering:

• Full visibility into the agent's decision-making path.
• Customizable loops—you decide when to stop researching, how many search iterations to run, and how to combine results.
• Built-in state management across multiple turns and tools.
• Streaming and debugging support out of the box.
• Human-in-the-loop capabilities for approval or clarification steps.

In this tutorial you'll build a research agent that takes a user query, performs web searches, evaluates the gathered information, and writes a concise summary—all within a controllable graph.

Prerequisites and Setup

Install the required packages:

pip install langgraph langchain langchain-community langchain-openai tavily-python

Set up environment variables for your OpenAI and Tavily API keys:

import os
os.environ["OPENAI_API_KEY"] = "your-openai-key"
os.environ["TAVILY_API_KEY"] = "your-tavily-key"

We'll use gpt-4o-mini for cost-effective reasoning and Tavily as our search tool. The Tavily search API is optimized for AI agents and returns clean, structured results.

Imports and LLM Initialization

from typing import TypedDict, List, Optional
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.messages import HumanMessage, AIMessage
import operator
from datetime import datetime

# LLM for decision-making and summarization
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

Step 1: Define the Agent State

LangGraph revolves around a shared state object that flows through the nodes. We define a typed schema using Python's TypedDict to ensure type safety and clarity. For our research agent, the state needs to track:

• The original user query.
• A list of search queries that have been tried.
• Accumulated raw research notes (from search results).
• A synthesized intermediate answer (to decide if more research is needed).
• A final answer.
• An iteration counter to prevent infinite loops.

class ResearchState(TypedDict):
    query: str                    # original user question
    search_queries: List[str]     # queries used so far
    research_notes: str           # concatenated findings from searches
    intermediate_answer: str      # current best answer after each cycle
    final_answer: Optional[str]   # final polished answer
    iteration: int                # number of research cycles

We also define a simple reducer for the search_queries list so we can append new queries without overwriting the whole list. LangGraph supports custom reducer functions for each state key.

def append_search_query(current: List[str], new: List[str]) -> List[str]:
    return (current or []) + new

When we create the graph, we'll pass this reducer as part of the state definition.

Step 2: Set Up the Tools

The agent's main tool is a web search. We wrap Tavily in a LangChain tool for easy integration. You can also add other tools like a calculator or a document retriever.

# Initialize Tavily search tool
search_tool = TavilySearchResults(
    max_results=3,
    include_answer=False,
    include_raw_content=False,
    search_depth="advanced"
)
# Wrap in a tool list (LangGraph nodes can access tools directly)
tools = [search_tool]
# Bind tools to the LLM so it can produce tool-calls
llm_with_tools = llm.bind_tools(tools)

In LangGraph we'll call the search tool manually inside a node function, so we don't necessarily need the LLM to output tool-calls; but binding tools is useful if we want the LLM to decide which tool to use. For simplicity, we'll have the LLM generate new search queries in text form and then we execute the search ourselves.

Step 3: Create the Node Functions

Nodes are Python functions that receive the current state and return an update to the state. We'll build four nodes:

• plan_research: Decides what to search for next (or if we have enough).
• perform_search: Executes the search using Tavily.
• synthesize: Reads the accumulated research notes and produces an intermediate answer.
• finalize: Polishes the final answer and marks completion.

Node 1: plan_research

This node examines the current state and uses the LLM to decide: should we search for more information? If so, what search query should we use? It outputs a decision and a new search query (if needed). We store the decision in the state indirectly by adding to search_queries and updating research_notes.

def plan_research(state: ResearchState) -> dict:
    """
    Decide next search query or whether to move to final answer.
    """
    prompt = f"""You are a research planner. Given the user query and the intermediate answer so far, decide if more research is needed. 
If more research is needed, suggest ONE specific new search query to fill gaps. If the information is sufficient, output "FINALIZE".

User query: {state['query']}
Current intermediate answer: {state['intermediate_answer']}
Research notes so far: {state['research_notes'][:2000]}  # truncate for prompt length
Iteration: {state['iteration']}/5 (max 5)

Respond in JSON format: {{"action": "search" or "finalize", "search_query": "..." if action is search}}"""

    response = llm.invoke([HumanMessage(content=prompt)])
    content = response.content.strip()
    
    # Simple parsing (in production use a structured output parser)
    import json
    try:
        # Extract JSON if wrapped in markdown
        if "json" in content:
            content = content.split("json")[1].split("")[0].strip()
        decision = json.loads(content)
    except Exception:
        # Fallback: assume finalize
        decision = {"action": "finalize"}
    
    if decision.get("action") == "search" and decision.get("search_query"):
        new_query = decision["search_query"]
        return {
            "search_queries": [new_query],  # append using reducer
            "iteration": state["iteration"] + 1,
            "intermediate_answer": state.get("intermediate_answer", "")
        }
    else:
        # Signal to move to finalize
        return {
            "iteration": state["iteration"] + 1,
            "intermediate_answer": state.get("intermediate_answer", ""),
            "_next": "finalize"  # custom key we'll use for routing
        }

Note: For brevity we use a custom _next key; a cleaner approach is to return a specific state update and let conditional edges read the state. We'll use the conditional edge approach below.

Node 2: perform_search

This node takes the latest search query (the last element in search_queries) and runs the Tavily search. It appends the raw results to research_notes.

def perform_search(state: ResearchState) -> dict:
    """
    Execute the most recent search query and append results to research_notes.
    """
    if not state["search_queries"]:
        return {}  # nothing to search
    
    latest_query = state["search_queries"][-1]
    print(f"Searching for: {latest_query}")
    
    # Call Tavily via the tool's internal method
    results = search_tool.invoke({"query": latest_query})
    # Format results as text
    formatted = "\n".join(
        [f"Source: {res['url']}\nContent: {res['content']}" for res in results]
    )
    
    new_notes = state.get("research_notes", "") + f"\n--- Search: {latest_query} ---\n{formatted}\n"
    
    return {
        "research_notes": new_notes,
        # Keep other fields unchanged
    }

Node 3: synthesize

This node reads the accumulated research notes and the user query, then generates an intermediate answer. It also reflects on whether the answer is satisfactory (we'll let the planner decide next steps).

def synthesize(state: ResearchState) -> dict:
    """
    Create or update the intermediate answer based on research notes.
    """
    prompt = f"""You are a research synthesizer. Based on the research notes below, write a concise intermediate answer to the user's query.
Include key facts and sources. If you don't have enough information, state what's missing clearly.

User query: {state['query']}
Research notes: {state['research_notes'][:3000]}

Intermediate answer:"""

    response = llm.invoke([HumanMessage(content=prompt)])
    return {
        "intermediate_answer": response.content.strip(),
        # Keep other fields unchanged
    }

Node 4: finalize

Produces the final polished answer and sets the final_answer key. We can also mark completion by setting a flag (or just returning __end__ in the graph).

def finalize(state: ResearchState) -> dict:
    """
    Polish the intermediate answer into a final, well-formatted response.
    """
    prompt = f"""You are a research editor. Turn the following intermediate answer into a final, polished response for the user. 
Add a summary, structure it clearly, and cite sources where possible.

User query: {state['query']}
Intermediate answer: {state['intermediate_answer']}

Final answer:"""

    response = llm.invoke([HumanMessage(content=prompt)])
    return {
        "final_answer": response.content.strip(),
        # Optionally clear intermediate to free memory
        "intermediate_answer": "",
    }

Step 4: Build the Graph

Now we wire the nodes together using StateGraph. We define the graph structure, add nodes, and set up conditional edges that route based on the state.

from langgraph.graph import StateGraph, END

# Create graph with our state schema and custom reducer for search_queries
graph = StateGraph(ResearchState)

# Add nodes
graph.add_node("plan", plan_research)
graph.add_node("search", perform_search)
graph.add_node("synthesize", synthesize)
graph.add_node("finalize", finalize)

Define the flow:

• Start at plan.
• From plan, conditional edge: if we need to search, go to search; else go to finalize.
• After search, go to synthesize.
• After synthesize, go back to plan to re-evaluate.
• finalize goes to END.

The conditional routing uses a function that reads the state and returns the next node name. We'll check if search_queries has grown (i.e., the planner added a new query) or if the planner set a flag. A robust method: in plan_research, we return an extra key _next_action (e.g., "search" or "finalize") and the router reads it.

Let's adjust plan_research to return an explicit routing key:

def plan_research(state: ResearchState) -> dict:
    # ... same logic ...
    if decision.get("action") == "search" and decision.get("search_query"):
        new_query = decision["search_query"]
        return {
            "search_queries": [new_query],
            "iteration": state["iteration"] + 1,
            "intermediate_answer": state.get("intermediate_answer", ""),
            "next": "search"   # routing key
        }
    else:
        return {
            "iteration": state["iteration"] + 1,
            "intermediate_answer": state.get("intermediate_answer", ""),
            "next": "finalize"
        }

Now we define the router function:

def route_after_plan(state: ResearchState) -> str:
    # Check the 'next' key we set in plan_research
    if state.get("next") == "search":
        return "search"
    return "finalize"

Add edges and conditional edges:

# Start edge (entry point)
graph.set_entry_point("plan")

# After plan, conditional routing
graph.add_conditional_edges(
    "plan",
    route_after_plan,
    {
        "search": "search",
        "finalize": "finalize"
    }
)

# After search, go to synthesize
graph.add_edge("search", "synthesize")

# After synthesize, loop back to plan for re-evaluation
graph.add_edge("synthesize", "plan")

# After finalize, end the run
graph.add_edge("finalize", END)

Now compile the graph. This step checks the graph structure and prepares it for execution.

app = graph.compile()

Optionally, you can visualize the graph (if running in a notebook):

from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

Step 5: Run the Agent

Invoke the compiled graph with an initial state. We'll provide the user query and initialize empty fields.

initial_state = {
    "query": "What are the latest breakthroughs in solid-state batteries as of 2025?",
    "search_queries": [],
    "research_notes": "",
    "intermediate_answer": "",
    "final_answer": None,
    "iteration": 0,
    "next": ""
}

# Run synchronously (for streaming, use app.stream)
result = app.invoke(initial_state)

print("FINAL ANSWER:")
print(result["final_answer"])

For streaming updates (to see the agent's progress in real time), use:

for event in app.stream(initial_state):
    for key, value in event.items():
        # Each node's output is yielded as {node_name: state_update}
        if key == "finalize":
            print("\n--- FINAL ANSWER ---")
            print(value.get("final_answer", ""))
        else:
            print(f"Node '{key}' updated state: {list(value.keys())}")

The agent will run up to 5 iterations (enforced in plan_research), performing search, synthesis, and planning loops until the planner decides to finalize or the max iteration is reached. You can adjust the max iterations or add a timeout.

Best Practices for Research Agents in LangGraph

• Keep state minimal but descriptive – only store what nodes need. Avoid bloating the state with large objects; truncate or offload to external storage if necessary.
• Use structured outputs – For the planner's decision, use LangChain's with_structured_output or parse JSON carefully to avoid parsing errors in production.
• Add a human-in-the-loop step – Before finalizing, you can insert an interrupt node that pauses execution and asks for user approval. LangGraph's interrupt and checkpointer make this easy.
• Implement a maximum iteration guard – Always cap the number of research cycles to prevent infinite loops.
• Log and debug with LangGraph's built-in tracing – Use langgraph.debug or integrate with LangSmith for observability.
• Handle tool errors gracefully – Wrap tool calls in try/except and return informative error messages as research notes so the synthesizer can adapt.
• Use a checkpointer for persistence – MemorySaver or SqliteSaver allows you to resume interrupted runs and supports human-in-the-loop workflows.

Adding Persistence and Interrupt

To make the agent resumable and support human approval, wrap the graph with a checkpointer and use interrupt before finalize.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app_with_checkpointer = graph.compile(checkpointer=checkpointer, interrupt_before=["finalize"])

# Run with a thread_id for persistence
config = {"configurable": {"thread_id": "research-1"}}
app_with_checkpointer.invoke(initial_state, config)

# After interrupt, you can inspect the state and resume
state = app_with_checkpointer.get_state(config)
print(state.values["intermediate_answer"])
# Resume with approval
app_with_checkpointer.invoke(None, config)  # continues execution

Conclusion

You've built a fully functional research assistant agent using LangGraph. The graph-based approach gave you explicit control over the research loop: planning, searching, synthesizing, and finalizing. By defining a clear state schema and routing logic, you can easily extend this agent with additional tools (e.g., PDF readers, calculators) or more sophisticated decision-making. LangGraph's composable architecture makes it straightforward to add persistence, streaming, and human-in-the-loop breaks, turning a simple prototype into a production-ready research co-pilot. Start from this foundation, iterate on the prompts, and watch your agent become a reliable digital researcher.