Build a LangGraph Research Agent That Searches, Reads, and Cites

A stateful LangGraph loop that plans queries, calls Serply's search API, fetches full pages through the scraper, and stops when it has enough to answer.

Profile picture of Serply
Serply
A directed graph with plan, search, read, and answer nodes

A single search call is not research. Research is a loop: ask something, look at what comes back, notice what’s missing, ask a better question. Tool-calling agents built on a plain while loop tend to do this badly — they either stop after the first result set or spiral for twenty calls with no sense of when they’re done.

LangGraph is a good fit here precisely because it makes the loop explicit. You define nodes, you define edges, and the “should I keep going?” decision becomes a conditional edge you can actually reason about instead of a prompt you hope the model respects.

The shape of the graph

Four nodes:

  • plan — turn the user’s question into one or more concrete search queries
  • search — hit Serply’s Google Search endpoint for each query
  • read — pull full page text for the most promising results
  • answer — write the final response with citations

The interesting edge is between read and plan: if the gathered material doesn’t cover the question, we go back and plan again with what we’ve learned.

State

from typing import Annotated, TypedDict
import operator


class ResearchState(TypedDict):
    question: str
    queries: list[str]
    results: Annotated[list[dict], operator.add]
    documents: Annotated[list[dict], operator.add]
    iterations: int

Using operator.add on results and documents means nodes append rather than overwrite, which is what you want when the loop runs more than once.

The Serply calls

Two functions, both thin. Search first:

import os
import requests
from urllib.parse import quote_plus

API_KEY = os.environ["SERPLY_API_KEY"]
HEADERS = {"X-Api-Key": API_KEY}


def serply_search(query: str, num: int = 10) -> list[dict]:
    """Google Search via Serply. Returns the organic results array."""
    q = quote_plus(query)
    url = f"https://api.serply.io/v1/search/q={q}&num={num}"
    resp = requests.get(url, headers={**HEADERS, "X-Proxy-Location": "US"}, timeout=30)
    resp.raise_for_status()
    return resp.json().get("results", [])

Note the URL shape — Serply embeds the Google-style query string in the path segment, so it’s /v1/search/q=..., not /v1/search?q=.... Each result comes back with title, link, description, position, and a metadata object that usually carries display_url.

Then the reader, which is where most agents fall down. A search snippet is two lines; answering a real question usually needs the page:

def serply_read(url: str) -> str:
    """Fetch a URL as markdown through Serply's scraper."""
    resp = requests.post(
        "https://api.serply.io/v1/request",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={"url": url, "response_type": "markdown"},
        timeout=60,
    )
    resp.raise_for_status()
    return resp.text

The markdown variant returns the converted text directly in the body — not JSON — so read .text, not .json(). If you ask for "response_type": "full" instead, you get a JSON object with the raw HTML under a data key.

The nodes

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

llm = ChatOpenAI(model="gpt-4o", temperature=0)


def plan_node(state: ResearchState) -> dict:
    known = "\n".join(d["url"] for d in state.get("documents", []))
    prompt = (
        f"Question: {state['question']}\n\n"
        f"Already read:\n{known or '(nothing yet)'}\n\n"
        "Write up to 3 Google queries that would fill the remaining gaps. "
        "One per line, no numbering."
    )
    out = llm.invoke([SystemMessage(content="You plan web research."),
                      HumanMessage(content=prompt)])
    queries = [line.strip() for line in out.content.splitlines() if line.strip()][:3]
    return {"queries": queries, "iterations": state.get("iterations", 0) + 1}


def search_node(state: ResearchState) -> dict:
    hits = []
    for q in state["queries"]:
        hits.extend(serply_search(q, num=10))
    return {"results": hits}


def read_node(state: ResearchState) -> dict:
    seen = {d["url"] for d in state.get("documents", [])}
    fresh = [r for r in state["results"] if r.get("link") and r["link"] not in seen]
    docs = []
    for r in fresh[:4]:
        try:
            docs.append({
                "url": r["link"],
                "title": r.get("title", ""),
                "text": serply_read(r["link"])[:12000],
            })
        except requests.HTTPError:
            continue
    return {"documents": docs}

Capping at four pages per iteration keeps latency and cost predictable. Truncating to 12k characters keeps the context window from filling up with boilerplate.

The conditional edge

def should_continue(state: ResearchState) -> str:
    if state["iterations"] >= 3:
        return "answer"
    if len(state["documents"]) >= 6:
        return "answer"
    return "plan"

Hard caps, not a model judgment call. You can add an LLM-based sufficiency check on top, but always keep a numeric ceiling underneath it — that’s what stops a bad day from turning into a hundred-dollar API bill.

Wiring it up

from langgraph.graph import StateGraph, START, END

builder = StateGraph(ResearchState)
builder.add_node("plan", plan_node)
builder.add_node("search", search_node)
builder.add_node("read", read_node)
builder.add_node("answer", answer_node)

builder.add_edge(START, "plan")
builder.add_edge("plan", "search")
builder.add_edge("search", "read")
builder.add_conditional_edges("read", should_continue,
                              {"plan": "plan", "answer": "answer"})
builder.add_edge("answer", END)

graph = builder.compile()

result = graph.invoke({
    "question": "What changed in EU AI Act enforcement timelines this year?",
    "iterations": 0,
})
print(result["answer"])

The answer node

def answer_node(state: ResearchState) -> dict:
    sources = "\n\n".join(
        f"[{i+1}] {d['title']} — {d['url']}\n{d['text'][:3000]}"
        for i, d in enumerate(state["documents"])
    )
    prompt = (
        f"Question: {state['question']}\n\nSources:\n{sources}\n\n"
        "Answer using only these sources. Cite with [n]. "
        "If the sources don't answer it, say so."
    )
    out = llm.invoke([SystemMessage(content="You write cited research answers."),
                      HumanMessage(content=prompt)])
    return {"answer": out.content}

That last instruction matters more than it looks. Without it the model will happily fall back on training data and present it in the same voice as the cited material, which is exactly the failure mode you built this graph to avoid.

What to tune

The num=10 on search is doing real work — dropping to 5 roughly halves your read volume downstream, since the reader only ever looks at the top few links. If your questions are geographically specific, set X-Proxy-Location to match (the endpoint accepts US, GB, DE, JP, AU, and about ten others); a query about local regulations returns visibly different results from a US exit node than a German one.

Watch for 429 responses under parallel load. The response carries x-ratelimit-requests-remaining, so you can back off before you hit the wall rather than after.