LangGraph in Practice

from typing import TypedDict, Annotated, List
import operator

class ResearchState(TypedDict):
    # The original research question
    question: str
    # Accumulated list of LangChain messages (auto-appended)
    messages: Annotated[list, operator.add]
    # Sourced claims collected so far
    claims: List[dict]
    # Whether the agent has enough information
    research_complete: bool
    # Number of search iterations (circuit breaker)
    search_count: int

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import ToolMessage
from langgraph.prebuilt import ToolNode
import json

llm = ChatAnthropic(model="claude-opus-4-8")
tools = [web_search, fetch_document]  # tools defined in Lesson 3
llm_with_tools = llm.bind_tools(tools)

def agent_node(state: ResearchState) -> dict:
    """
    The reasoning node: calls the LLM with current messages.
    Returns a state update containing the model's response.
    """
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# ToolNode handles tool dispatch, execution, and result formatting
# It also catches tool exceptions and formats them as tool error messages
# rather than letting them propagate as unhandled exceptions
tool_node = ToolNode(tools)

def write_report_node(state: ResearchState) -> dict:
    """
    Final node: synthesizes accumulated messages into the output report.
    Runs only when research_complete is True.
    """
    synthesis_prompt = f"""Based on your research, write a structured report
    that answers: {state['question']}
    Include: (1) at least 3 sourced claims with URLs, (2) any information gaps."""

    from langchain_core.messages import HumanMessage
    messages = state["messages"] + [HumanMessage(content=synthesis_prompt)]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import AIMessage

def should_continue(state: ResearchState) -> str:
    """
    Routing function: decides which node comes after the agent node.
    Returns the name of the next node.
    """
    messages = state["messages"]
    last = messages[-1]

    # Circuit breaker: stop after 5 searches to prevent runaway loops
    if state["search_count"] >= 5:
        return "write_report"

    # If the model called tools, execute them
    if isinstance(last, AIMessage) and last.tool_calls:
        return "tools"

    # No tool calls: agent is done researching
    return "write_report"

# Build the graph
builder = StateGraph(ResearchState)

# Add nodes
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.add_node("write_report", write_report_node)

# Entry point
builder.set_entry_point("agent")

# Conditional edge: after agent, route based on should_continue()
builder.add_conditional_edges("agent", should_continue)

# After tools run, always return to agent
builder.add_edge("tools", "agent")

# write_report is the terminal node
builder.add_edge("write_report", END)

# Compile with an in-memory checkpointer (swap for SQLite for persistence)
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# Run it
config = {"configurable": {"thread_id": "research-001"}}
result = graph.invoke(
    {
        "question": "What are the main approaches to AI agent memory?",
        "messages": [{"role": "user", "content":
                      "Research: What are the main approaches to AI agent memory?"}],
        "claims": [],
        "research_complete": False,
        "search_count": 0
    },
    config
)

# Compile with interrupt_before the tools node
graph_with_hitl = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["tools"]  # pause before every tool execution
)

# First invocation runs until it reaches the "tools" node
snapshot = graph_with_hitl.invoke(initial_state, config)
# snapshot.next == ("tools",) - graph is paused here

# Inspect the pending tool calls before approving
pending = snapshot.values["messages"][-1].tool_calls
print(f"Agent wants to call: {[tc['name'] for tc in pending]}")

# Human approves - resume by calling invoke again with same config
# LangGraph reads the saved checkpoint and continues from the interrupt
result = graph_with_hitl.invoke(None, config)

for event in graph.stream(initial_state, config, stream_mode="updates"):
    node_name = list(event.keys())[0]
    update = event[node_name]
    print(f"--- {node_name} ---")
    if "messages" in update:
        last_msg = update["messages"][-1]
        print(f"  type: {type(last_msg).__name__}")
        if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
            print(f"  tools: {[tc['name'] for tc in last_msg.tool_calls]}")