07 Oct LangGraph Project: Build a Research Assistant with Tools

This project uses parallel work and a human review loop. You give the assistant a topic. It plans a few research questions, investigates each one at the same time using a tool, writes a report, and lets you request revisions before finishing.

What we are building

                      +-> research (Q1) -+
START -> plan --(Send)+-> research (Q2) -+-> write_report -> review --> END
                      +-> research (Q3) -+        ^            |
                                                  +-(feedback)-+

Concepts used, and where you learned them:

• Structured output for the plan: Chapter 18
• Tool and agent for research: Chapters 10 and 12
• Parallel research with Send: Chapter 19
• Reducer to collect findings: Chapter 5
• Human review with interrupt(): Chapter 15
• Revision loop with a limit: Chapter 7
• Checkpointer: Chapter 13

To avoid needing another API key, our tool uses Wikipedia’s public search API. Save everything in one file, ch22_research_assistant.py.

Step 1: Imports and setup

pip install -U requests

import operator
from typing import Annotated, Literal, TypedDict

import requests
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import SystemMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send, interrupt, Command

load_dotenv()
llm = ChatAnthropic(model="claude-sonnet-5-5")

def as_text(content):
    if isinstance(content, str):
        return content
    return "".join(b.get("text", "") for b in content if isinstance(b, dict))

Step 2: The research tool

API = "https://en.wikipedia.org/w/api.php"
HEADERS = {"User-Agent": "StudyopediaTutorial/1.0 (learning project)"}

@tool
def search_wikipedia(query: str) -> str:
    """Search Wikipedia and return the introduction of the best matching article."""
    try:
        found = requests.get(
            API,
            params={"action": "query", "list": "search", "srsearch": query,
                    "srlimit": 1, "format": "json"},
            headers=HEADERS, timeout=10,
        )
        found.raise_for_status()
        hits = found.json()["query"]["search"]
        if not hits:
            return "No results found."

        title = hits[0]["title"]
        page = requests.get(
            API,
            params={"action": "query", "prop": "extracts", "exintro": 1,
                    "explaintext": 1, "titles": title, "redirects": 1,
                    "format": "json"},
            headers=HEADERS, timeout=10,
        )
        page.raise_for_status()
        pages = page.json()["query"]["pages"]
        text = next(iter(pages.values())).get("extract", "")
        return f"Source: Wikipedia article '{title}'\n{text[:1500]}"
    except (requests.RequestException, KeyError, ValueError):
        return "Search failed. Try a different query."

Some design choices worth noting:

• The tool catches errors and returns a message, so one failed request doesn’t crash the whole graph (Chapter 10 tip)
• timeout=10 stops it from hanging forever
• Wikipedia asks API users to send a descriptive User-Agent, so change the text to your own site or project name
• The text is cut to 1500 characters to keep model calls cheap
• The result names the source article, so the report can credit it

Step 3: State and the plan

class ResearchState(TypedDict):
    topic: str
    questions: list[str]
    findings: Annotated[list[str], operator.add]   # parallel branches append here
    report: str
    feedback: str
    revisions: int

class Plan(BaseModel):
    questions: list[str] = Field(
        description="Three short, distinct research questions that together cover the topic."
    )

planner = llm.with_structured_output(Plan)

def plan(state: ResearchState):
    result = planner.invoke([
        SystemMessage("Break the topic into three research questions for an encyclopedia lookup."),
        ("human", state["topic"]),
    ])
    return {"questions": result.questions[:3]}

The operator.add reducer on findings is essential. Without it, parallel branches writing to the same key would raise an error (Chapter 19). Slicing with [:3] guards against the model returning too many questions.

Step 4: Parallel research

research_agent = create_agent(
    llm,
    tools=[search_wikipedia],
    system_prompt=(
        "You are a careful researcher. Use the search tool, at most twice, to answer the "
        "question. Write 3 to 4 sentences using ONLY what the tool returned, and name the "
        "Wikipedia article you used. If nothing useful was found, say so."
    ),
)

def fan_out(state: ResearchState):
    return [Send("research", {"question": q}) for q in state["questions"]]

def research(payload: dict):
    question = payload["question"]
    try:
        result = research_agent.invoke(
            {"messages": [("human", question)]},
            {"recursion_limit": 8},
        )
        notes = as_text(result["messages"][-1].content)
    except Exception:
        notes = "Research for this question failed."
    return {"findings": [f"Question: {question}\nNotes: {notes}"]}

How it works:

• fan_out creates one Send per question, so three questions give three parallel runs
• Each research run receives only its own payload ({“question”: …})
• Each one runs a small tool-using agent (Chapter 12) with a recursion limit, so a single question can’t loop forever
• Each result includes its question, so the order they finish in doesn’t matter
• A failed branch returns a note instead of stopping the other two

Step 5: Write, review, revise

def write_report(state: ResearchState):
    notes = "\n\n".join(state["findings"])
    prompt = (
        f"Topic: {state['topic']}\n\nResearch notes:\n{notes}\n\n"
        "Write a clear, well-organized report of about 250 words for a beginner. "
        "Use ONLY the notes above, in your own words, and end with a 'Sources' list "
        "naming the Wikipedia articles used. If the notes are thin, say what is missing."
    )
    if state["feedback"]:
        prompt += (
            f"\n\nHere is your previous draft:\n{state['report']}\n\n"
            f"Revise it to address this feedback: {state['feedback']}"
        )
    reply = llm.invoke([SystemMessage("You are a careful research writer."), ("human", prompt)])
    return {"report": as_text(reply.content)}

def review(state: ResearchState):
    answer = interrupt({
        "report": state["report"],
        "message": "Type 'ok' to accept, or write feedback for a revision.",
    })
    if answer.strip().lower() == "ok":
        return {"feedback": ""}
    return {"feedback": answer, "revisions": state["revisions"] + 1}

def after_review(state: ResearchState) -> Literal["write_report", "__end__"]:
    if state["feedback"] and state["revisions"] < 2:
        return "write_report"
    return END

• write_report tells the model to use only the notes. That reduces invented facts, and putting “in your own words” in the prompt keeps the report from copying source text.
• review pauses with interrupt(). The human either accepts (ok) or returns feedback.
• after_review loops back for a revision, but only twice at most (Chapter 7’s exit-condition rule), so an endless feedback loop can’t happen.
• The pause is at the start of review, so there are no side effects before it (Chapter 15’s golden rule).

Step 6: Build the graph

builder = StateGraph(ResearchState)

builder.add_node("plan", plan)
builder.add_node("research", research)
builder.add_node("write_report", write_report)
builder.add_node("review", review)

builder.add_edge(START, "plan")
builder.add_conditional_edges("plan", fan_out, ["research"])
builder.add_edge("research", "write_report")
builder.add_edge("write_report", "review")
builder.add_conditional_edges("review", after_review)

graph = builder.compile(checkpointer=InMemorySaver())

Here, write_report runs once, after all parallel research branches finish. The checkpointer is required for the review pause.

Step 7: Run it

config = {
    "configurable": {"thread_id": "research-1"},
    "recursion_limit": 25,
    "max_concurrency": 3,
}

topic = input("Research topic: ")

result = graph.invoke(
    {"topic": topic, "questions": [], "findings": [],
     "report": "", "feedback": "", "revisions": 0},
    config,
)

while "__interrupt__" in result:
    request = result["__interrupt__"][0].value
    print("\n--- DRAFT REPORT ---\n")
    print(request["report"])
    answer = input(f"\n{request['message']}\n> ")
    result = graph.invoke(Command(resume=answer), config)

print("\n--- FINAL REPORT ---\n")
print(result["report"])

Sample session (wording will differ):

Research topic: How do vaccines work?

--- DRAFT REPORT ---

Vaccines train the immune system to recognize a germ without causing the full
illness. They contain a harmless piece or weakened form of a pathogen...

Sources: Vaccine; Immune system; Herd immunity

Type 'ok' to accept, or write feedback for a revision.
> make it simpler and add a short example

--- FINAL REPORT ---

(a revised version, written more simply, with an example)

Behind the scenes: plan produced three questions, three research agents ran in parallel, write_report combined their notes, and your feedback triggered one revision loop.

Watching it work

To see the steps, stream updates while it runs. Streaming with interrupt() also delivers the pause, so here is a simple version for the first run only:

for step in graph.stream(
    {"topic": "How do vaccines work?", "questions": [], "findings": [],
     "report": "", "feedback": "", "revisions": 0},
    {"configurable": {"thread_id": "research-2"}, "recursion_limit": 25},
    stream_mode="updates",
):
    print(list(step.keys()))

Sample output (order of the middle lines may vary):

['plan']
['research']
['research']
['research']
['write_report']
['__interrupt__']

Three research lines confirm the fan-out. __interrupt__ shows the graph waiting for your review.

Test the tool by itself

print(search_wikipedia.invoke({"query": "photosynthesis"})[:300])

If this works but the graph fails, the problem is elsewhere. If it fails, check your internet connection or the User-Agent.

Security and quality notes

• Treat fetched text as untrusted. A web page can contain sentences like “ignore your instructions”. Our design limits the damage: the researcher can only read, the writer has no tools, and a human reviews the result. Keep it that way, and don’t add powerful tools (sending email, writing files) to the research agent.
• Check facts before publishing. A model can misread or overstate its notes. Treat the report as a draft, and verify important claims against the sources it lists.
• Respect content licenses. Wikipedia text is licensed with attribution requirements, and we ask the model to rewrite in its own words and list sources. If you publish reports on your site, credit your sources properly, and read the license of any source you use.
• Be polite to APIs. Keep the max_concurrency limit and timeouts, and don’t hammer a public service with hundreds of requests.
• Watch costs. One run makes a planning call, several agent calls (each with possible tool rounds), a writing call, and possibly revisions. The recursion limits keep this bounded.

Ideas to extend it

• Swap Wikipedia for a proper web search tool (many providers offer LangChain integrations, so check their current docs for setup)
• Save the accepted report to a Markdown file
• Store finished reports in a Store (Chapter 14), so the assistant can reuse past research
• Turn the researcher into a subgraph (Chapter 17) with its own steps, such as search, then read, then summarize
• Let the user choose the number of questions
• Trace runs in LangSmith to see exactly what each researcher searched (Chapter 20)
• Studyopedia idea: generate study notes plus quiz questions for a topic

Common mistakes

• No reducer on findings: parallel writes would raise an InvalidUpdateError
• Forgetting the third argument in add_conditional_edges(“plan”, fan_out, [“research”]): LangGraph can’t see the Send target
• Reading the whole state inside research: it receives only the payload you sent
• No exit limit on the revision loop: keep revisions < 2 or similar
• No timeout on web requests: a slow site could hang your app
• Letting the writer add outside facts: keep “use ONLY the notes” in the prompt
• Changing the thread_id on resume: the paused review can’t be found
• Trusting the report without checking sources

Summary

• You built a research assistant that plans, researches in parallel, writes, and revises
• Send created one research run per question, and a reducer merged their findings
• A read-only tool plus human review keeps the design safe
• Loops have limits, tools have timeouts, and errors are handled inside nodes
• The pattern (plan, fan out, combine, review) works for many tasks

Practice

1. Run it on three topics, and check each report’s sources against the real articles
2. Change the plan to five questions, and set max_concurrency to 2
3. Give feedback twice, and confirm that a third round is not requested
4. Add a save_report node after review that writes the accepted report to a .md file

---

If you liked the tutorial, spread the word and share the link and our website, Studyopedia, with others.

---

For Videos, Join Our YouTube Channel: Join Now

---

Read More:

• Generative AI Tutorial
• AI Ethics
• Machine Learning Tutorial
• Deep Learning Tutorial
• Ollama Tutorial
• Retrieval Augmented Generation (RAG) Tutorial
• ChatGPT Tutorial
• Microsoft Copilot Tutorial