From ReAct to StateGraph: Building a Production-Ready Research Agent with LangGraph and Firecrawl

Tuan Do — Thu, 06 Aug 2026 04:26:54 +0000

The Problem with the Classic Agent Loop

Most AI agent tutorials show you the same pattern. You write a while True: loop. The AI calls a tool, gets a result, and calls another tool, again and again, until it decides to stop. It looks simple. It looks clean. But this pattern breaks easily once you try to use it for real.

Here is what that classic loop looks like in code:

# The classic "ReAct loop" — simple to write, risky to run
def react_agent(query: str, max_steps: int = 50):
    messages = [{"role": "user", "content": query}]

    while True:  # <-- the problem starts here
        response = llm.chat(messages, tools=available_tools)

        if response.tool_calls:
            for call in response.tool_calls:
                result = execute_tool(call)
                messages.append({"role": "tool", "content": result})
                # ^ this list just keeps growing, with no limit
        else:
            return response.content
            # ^ this is the ONLY way to stop —
            #   and it only works if the AI *chooses* to stop

This code will run fine the first few times you test it. But once you use it for real tasks, three problems show up.

Problem 1: The Loop Never Ends

Look closely at the loop. The only way it stops is if the AI decides, on its own, to stop calling tools. That's not a real safety rule — it's just a hope that the AI will behave well. If a task is unclear, or a tool keeps returning empty results, the AI can easily keep calling tools forever. Many people add a max_steps limit to fix this. But a simple step limit doesn't really solve the problem — it just stops the damage a bit later. The agent still doesn't know when it should actually stop.

Problem 2: The Context Grows Too Big

Every single tool result gets added to the messages list, and nothing is ever removed. There's no cleanup, no summary, no filter for what's actually still useful. After many steps, this list is full of long, messy text — much of it no longer relevant. This causes two problems: the agent gets slower and more expensive to run (because the AI has to read more text every time), and the AI's answers actually get worse, because too much irrelevant text makes it harder for the model to focus on what matters.

Problem 3: You Lose Control of What the Agent Knows

There is no clear, organized record of "what has the agent learned so far." Everything is mixed together inside one long chat history. This creates real problems when you try to use the agent in production: Can you save its progress and continue later if it crashes? Can you check what it knew at a certain step? Can you test just one part of it on its own? With a single messy chat list, the answer is usually no. You end up debugging a long conversation instead of debugging a program.

These are not rare edge cases. They happen by default, the moment you give the agent a real task instead of a simple demo. The fix isn't a smarter loop — it's a different design altogether: a graph, where the agent's knowledge is stored in clear, organized data, and where you — not the AI — decide when to stop.

Graph Architecture with LangGraph + Firecrawl

LangGraph lets you build an agent as a state machine instead of a loop. Instead of "keep chatting until the AI feels like stopping," you define clear steps (called nodes) and clear rules for what happens next (called edges). This one change fixes all three problems above.

We'll also use Firecrawl to search the web and turn pages into clean, readable text — which helps a lot with Problem 2 (too much messy text).

The State: `ResearchState`

Instead of one long chat history, we define exactly what information our agent is allowed to hold at any time:

from typing import TypedDict, List, Annotated
import operator

class ResearchState(TypedDict):
    topic: str
    search_queries: List[str]
    scraped_content: Annotated[List[dict], operator.add]
    completeness_score: float
    completeness_feedback: str
    retry_count: int
    final_report: str

A few things to notice:

scraped_content uses Annotated[List[dict], operator.add]. This tells LangGraph: "when a new source is found, add it to the list — don't erase what was already there." This replaces the messy, never-ending chat list with something clear and organized.
retry_count and completeness_score are simple fields you can check. This means the "should we stop yet?" decision is based on real, visible data — not hidden inside the AI's mind.
Nothing here is raw chat text. Every field answers one clear question about the research task. That means you can look at it, log it, save it, and test it. ### The Four Steps (Nodes)

We split the agent into four small, simple steps:

plan_query — turns the topic (and, if needed, what's still missing) into real search queries.
firecrawl_search — runs those searches with Firecrawl, gets clean text from web pages, and saves the results.
evaluate_completeness — checks the sources found so far and decides: is this enough information, or not?
write_report — writes the final report using only the sources we collected. Each step only reads and updates the shared state — nothing more. This is what fixes Problem 3: every step is small, clear, and easy to test on its own.

The Rule That Stops the Loop

The most important fix for Problem 1 is this one small function:

def should_continue_research(state: ResearchState) -> str:
    if state["completeness_score"] >= 0.75:
        return "write_report"
    if state["retry_count"] >= MAX_RETRIES:
        return "write_report"  # stop and write with what we have
    return "plan_query"

Notice what this does. The agent doesn't stop because the AI "feels done." It stops because of a clear, checkable rule based on real data. And even if the score never gets high enough, there's a hard limit (retry_count >= MAX_RETRIES) that always kicks in. The loop cannot run forever. That's not a hope anymore — it's a guarantee.

Step-by-Step Setup and Real Test

1. Set Up Your Environment

python -m venv venv
source venv/bin/activate   # on Windows: venv\Scripts\activate

pip install langgraph openai firecrawl-py python-dotenv

Create a file named .env:

OPENAI_API_KEY=sk-...
FIRECRAWL_API_KEY=fc-...

2. The Full Code

Save this as research_agent.py. It includes every step and the full graph.

import os
from typing import TypedDict, List, Annotated
import operator

from dotenv import load_dotenv
from langgraph.graph import StateGraph, END
from firecrawl import FirecrawlApp
from openai import OpenAI

load_dotenv()

# ---------------------------------------------------------------------------
# Setup
# ---------------------------------------------------------------------------
firecrawl = FirecrawlApp(api_key=os.environ["FIRECRAWL_API_KEY"])
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

MODEL = "gpt-4o-mini"
MAX_RETRIES = 3
COMPLETENESS_THRESHOLD = 0.75


# ---------------------------------------------------------------------------
# State: what the agent is allowed to know
# ---------------------------------------------------------------------------
class ResearchState(TypedDict):
    topic: str
    search_queries: List[str]
    scraped_content: Annotated[List[dict], operator.add]
    completeness_score: float
    completeness_feedback: str
    retry_count: int
    final_report: str


# ---------------------------------------------------------------------------
# Step 1: Plan Query
# ---------------------------------------------------------------------------
def plan_query_node(state: ResearchState) -> dict:
    topic = state["topic"]
    retry_count = state.get("retry_count", 0)

    if retry_count == 0:
        prompt = (
            f"Generate 3 diverse, specific web search queries to research "
            f"the topic: '{topic}'. Return them as a numbered list, nothing else."
        )
    else:
        gaps = state.get("completeness_feedback", "")
        prompt = (
            f"We are researching: '{topic}'.\n"
            f"Here is what's still missing from our research: {gaps}\n"
            f"Generate 2 new, more specific search queries to fill these gaps. "
            f"Return them as a numbered list, nothing else."
        )

    response = client.chat.completions.create(
        model=MODEL,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,
    )

    raw = response.choices[0].message.content or ""
    queries = [
        line.split(".", 1)[-1].strip()
        for line in raw.strip().split("\n")
        if line.strip()
    ]

    print(f"[plan_query] queries: {queries}")
    return {"search_queries": queries}


# ---------------------------------------------------------------------------
# Step 2: Firecrawl Search & Scrape
# ---------------------------------------------------------------------------
def firecrawl_search_node(state: ResearchState) -> dict:
    new_content = []

    for query in state["search_queries"]:
        try:
            results = firecrawl.search(
                query,
                params={"limit": 3, "scrapeOptions": {"formats": ["markdown"]}},
            )
            for r in results.get("data", []):
                new_content.append(
                    {
                        "query": query,
                        "url": r.get("url"),
                        "title": r.get("title"),
                        # Keep only the first part of each page —
                        # this is what keeps the text from getting too big.
                        "content": (r.get("markdown") or "")[:4000],
                    }
                )
        except Exception as e:
            print(f"[firecrawl_search] query '{query}' failed: {e}")
            continue

    print(f"[firecrawl_search] gathered {len(new_content)} sources")
    return {"scraped_content": new_content}


# ---------------------------------------------------------------------------
# Step 3: Evaluate Completeness
# ---------------------------------------------------------------------------
def evaluate_completeness_node(state: ResearchState) -> dict:
    topic = state["topic"]
    sources = state["scraped_content"]

    sources_summary = "\n\n".join(
        f"[{i + 1}] {c['title']} ({c['url']})\n{c['content'][:500]}..."
        for i, c in enumerate(sources)
    )

    prompt = (
        f"Topic being researched: '{topic}'\n\n"
        f"We have gathered {len(sources)} sources so far:\n{sources_summary}\n\n"
        f"On a scale from 0.0 to 1.0, how completely do these sources cover "
        f"the topic? Respond in this exact format:\n"
        f"SCORE: <number>\n"
        f"GAPS: <one short sentence describing what's missing, or \"none\">"
    )

    response = client.chat.completions.create(
        model=MODEL,
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )

    text = response.choices[0].message.content or ""
    score_line = next((l for l in text.split("\n") if l.startswith("SCORE:")), "SCORE: 0.5")
    gaps_line = next((l for l in text.split("\n") if l.startswith("GAPS:")), "GAPS: none")

    score = float(score_line.replace("SCORE:", "").strip())
    gaps = gaps_line.replace("GAPS:", "").strip()
    retry_count = state.get("retry_count", 0) + 1

    print(f"[evaluate_completeness] score={score} retry_count={retry_count} gaps='{gaps}'")

    return {
        "completeness_score": score,
        "completeness_feedback": gaps,
        "retry_count": retry_count,
    }


# ---------------------------------------------------------------------------
# Step 4: Write Report
# ---------------------------------------------------------------------------
def write_report_node(state: ResearchState) -> dict:
    topic = state["topic"]
    sources_text = "\n\n---\n\n".join(
        f"Source: {c['title']} ({c['url']})\n{c['content']}"
        for c in state["scraped_content"]
    )

    prompt = (
        f"Write a well-structured research report on: '{topic}'\n\n"
        f"Base your report strictly on the following sources. Cite sources "
        f"by their URL where relevant.\n\n{sources_text}\n\n"
        f"The report should have an introduction, 3-4 body sections with "
        f"clear headers, and a conclusion."
    )

    response = client.chat.completions.create(
        model=MODEL,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.4,
    )

    print("[write_report] report generated")
    return {"final_report": response.choices[0].message.content}


# ---------------------------------------------------------------------------
# The rule that decides: keep researching, or write the report?
# ---------------------------------------------------------------------------
def should_continue_research(state: ResearchState) -> str:
    if state["completeness_score"] >= COMPLETENESS_THRESHOLD:
        return "write_report"
    if state["retry_count"] >= MAX_RETRIES:
        return "write_report"  # stop and write with what we have
    return "plan_query"


# ---------------------------------------------------------------------------
# Build and connect the graph
# ---------------------------------------------------------------------------
def build_graph():
    graph = StateGraph(ResearchState)

    graph.add_node("plan_query", plan_query_node)
    graph.add_node("firecrawl_search", firecrawl_search_node)
    graph.add_node("evaluate_completeness", evaluate_completeness_node)
    graph.add_node("write_report", write_report_node)

    graph.set_entry_point("plan_query")

    graph.add_edge("plan_query", "firecrawl_search")
    graph.add_edge("firecrawl_search", "evaluate_completeness")

    graph.add_conditional_edges(
        "evaluate_completeness",
        should_continue_research,
        {
            "plan_query": "plan_query",
            "write_report": "write_report",
        },
    )

    graph.add_edge("write_report", END)

    return graph.compile()


# ---------------------------------------------------------------------------
# Run it
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    app = build_graph()

    initial_state: ResearchState = {
        "topic": "The impact of AI coding agents on software engineering productivity in 2026",
        "search_queries": [],
        "scraped_content": [],
        "completeness_score": 0.0,
        "completeness_feedback": "",
        "retry_count": 0,
        "final_report": "",
    }

    result = app.invoke(initial_state)

    print("\n" + "=" * 60)
    print("FINAL REPORT")
    print("=" * 60)
    print(result["final_report"])

3. Run the Agent

python research_agent.py

You will see the agent move through each step in order: plan_query → firecrawl_search → evaluate_completeness. Then it will either go back to plan_query with better search terms, or move on to write_report. Unlike the old while True: loop, every one of these steps is something you can see, log, and check — before it ever reaches a real user.

4. What You Can Add Next

Once this basic version is working, here are a few simple next steps:

Save progress as you go. LangGraph supports tools like SqliteSaver so if the agent crashes, it can pick up where it left off, instead of starting over. This solves Problem 3 for good.
Show each step's result on screen as it happens. Since every step returns clean, organized data, this is much easier than trying to stream a messy chat history.
Use a simpler check instead of an AI check for evaluate_completeness (for example: "did we find at least 5 sources from different websites?"). This can be faster and cheaper than always asking the AI. The main idea stays the same no matter which tools you use: an agent that has to guess when to stop is risky. An agent whose stopping point is a clear, visible rule inside a graph is something you can actually trust and ship.

Deploying Under the Moonlight

Tuan Do — Sat, 18 Apr 2026 14:25:37 +0000

There's something magical about a calm night walk that clears your mind. I was out, just enjoying the evening breeze and the peaceful streets.

Suddenly, I realized a small but frustrating bug was bothering my personal project. Users were being forced to re-login every time my server restarted. Instead of rushing home to my computer to fix it, I pulled out my phone. I didn't want to break the peaceful mood of my walk.

I opened GitHub on my mobile browser to investigate. I quickly found the culprit: my user session storage was too temporary. Every server blip wiped it clean.

Instead of manually rewriting complex storage logic right there on the sidewalk, I used a modern shortcut. I engaged an AI agent built right into the platform.

I simply gave it a natural language instruction: “Maintain user login sessions for at least 1 day.”

Within seconds, the AI understood the problem and generated the exact changes needed to switch to stable storage. I reviewed the AI’s work on my small screen, merged the pull request, and triggered the entire automated deployment workflow with a single tap.

I put my phone back in my pocket and continued my stroll, letting the background automation do the rest.

Moments later, my phone started buzzing with notifications from my Telegram deployment bot. First, "Images Pushed Successfully," and then a second later, "Restart Successful!"

Seeing that "Completed successfully 🚀" message while my feet were still hitting the pavement felt incredible. It proved that we no longer need to be chained to a physical desk to be productive. We are living in an era where technology—powered by mobile access and seamless AI assistance—has made the entire world our office. You truly can bring an idea all the way to production, even while taking a casual evening stroll.

DEV Community: Tuan Do