6.5 Framework Four: LangGraph

6.5.1 LangGraph Structure Overview

LangGraph, as an important extension of the LangChain ecosystem, represents a completely new direction in agent framework design. Unlike the "conversation"-based frameworks introduced earlier (such as AutoGen and CAMEL), LangGraph models the agent's execution flow as a State Machine and represents it as a Directed Graph. In this paradigm, the graph's Nodes represent specific computational steps (such as calling LLM, executing tools), while Edges define the transition logic from one node to another. The revolutionary aspect of this design is that it natively supports loops, making it unprecedentedly intuitive and simple to build complex agent workflows capable of iteration, reflection, and self-correction.

To understand LangGraph, we need to first grasp its three basic components.

First, is the global state (State). The entire graph's execution process revolves around a shared state object. This state is usually defined as a Python TypedDict, which can contain any information you need to track, such as conversation history, intermediate results, iteration count, etc. All nodes can read and update this central state.

from typing import TypedDict, List

# Define global state data structure
class AgentState(TypedDict):
    messages: List[str]      # Conversation history
    current_task: str        # Current task
    final_answer: str        # Final answer
    # ... any other state to track

Second, are the nodes (Nodes). Each node is a Python function that receives the current state as input and returns an updated state as output. Nodes are units that perform specific work.

# Define a "planner" node function
def planner_node(state: AgentState) -> AgentState:
    """Formulate a plan based on current task and update state."""
    current_task = state["current_task"]
    # ... call LLM to generate plan ...
    plan = f"Plan generated for task '{current_task}'..."

    # Append new message to state
    state["messages"].append(plan)
    return state

# Define an "executor" node function
def executor_node(state: AgentState) -> AgentState:
    """Execute latest plan and update state."""
    latest_plan = state["messages"][-1]
    # ... execute plan and get result ...
    result = f"Result of executing plan '{latest_plan}'..."

    state["messages"].append(result)
    return state

Finally, are the edges (Edges). Edges are responsible for connecting nodes and defining the direction of the workflow. The simplest edge is a regular edge, which specifies that the output of one node always flows to another fixed node. LangGraph's most powerful feature lies in Conditional Edges. It uses a function to judge the current state and then dynamically decides which node to jump to next. This is the key to implementing loops and complex logical branches.

def should_continue(state: AgentState) -> str:
    """Condition function: decide next route based on state."""
    # Assume if messages are less than 3, need to continue planning
    if len(state["messages"]) < 3:
        # Returned string needs to match the key defined when adding conditional edge
        return "continue_to_planner"
    else:
        state["final_answer"] = state["messages"][-1]
        return "end_workflow"

After defining state, nodes, and edges, we can assemble them into an executable workflow like building blocks.

from langgraph.graph import StateGraph, END

# Initialize a state graph and bind our defined state structure
workflow = StateGraph(AgentState)

# Add node functions to the graph
workflow.add_node("planner", planner_node)
workflow.add_node("executor", executor_node)

# Set graph entry point
workflow.set_entry_point("planner")

# Add regular edge, connecting planner and executor
workflow.add_edge("planner", "executor")

# Add conditional edge, implementing dynamic routing
workflow.add_conditional_edges(
    # Starting node
    "executor",
    # Judgment function
    should_continue,
    # Route mapping: map judgment function's return value to target node
    {
        "continue_to_planner": "planner", # If returns "continue_to_planner", jump back to planner node
        "end_workflow": END               # If returns "end_workflow", end process
    }
)

# Compile graph, generate executable application
app = workflow.compile()

# Run graph
inputs = {"current_task": "Analyze recent AI industry news", "messages": []}
for event in app.stream(inputs):
    print(event)

6.5.2 Three-Step Q&A Assistant

After understanding LangGraph's core concepts, we will consolidate what we've learned through a practical case. We will build a simplified Q&A dialogue assistant that follows a clear, fixed three-step process to answer user questions:

1. Understand: First, analyze the user's query intent.
2. Search: Then, simulate searching for information related to the intent.
3. Answer: Finally, generate the final answer based on the intent and searched information.

This case will clearly demonstrate how to define state, create nodes, and linearly connect them into a complete workflow. We will break down the code into four core steps: define state, create nodes, build graph, and run application.

(1) Define Global State

First, we need to define a global state that runs through the entire workflow. This is a shared data structure that is passed between each node of the graph, serving as the persistent context of the workflow. Each node can read data from this structure and update it.

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class SearchState(TypedDict):
    messages: Annotated[list, add_messages]
    user_query: str      # User requirement summary after LLM understanding
    search_query: str    # Optimized search query for Tavily API
    search_results: str  # Results returned by Tavily search
    final_answer: str    # Final generated answer
    step: str            # Mark current step

We created the SearchState TypedDict, defining a clear data schema for the state object. A key design is the inclusion of both user_query and search_query fields. This allows the agent to first optimize the user's natural language question into refined keywords more suitable for search engines, thereby significantly improving the quality of search results.

(2) Define Workflow Nodes

After defining the state structure, the next step is to create the various nodes that make up our workflow. In LangGraph, each node is a Python function that performs a specific task. These functions receive the current state object as input and return a dictionary containing updated fields.

Before defining nodes, we first complete the project initialization setup, including loading environment variables and instantiating the large language model.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from tavily import TavilyClient

# Load environment variables from .env file
load_dotenv()

# Initialize model
# We will use this llm instance to drive the intelligence of all nodes
llm = ChatOpenAI(
    model=os.getenv("LLM_MODEL_ID", "gpt-4o-mini"),
    api_key=os.getenv("LLM_API_KEY"),
    base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1"),
    temperature=0.7
)
# Initialize Tavily client
tavily_client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))

Now, let's create the three core nodes one by one.

(1) Understand and Query Node

This node is the first step of the workflow. Its responsibility is to understand user intent and generate an optimized search query for it.

def understand_query_node(state: SearchState) -> dict:
    """Step 1: Understand user query and generate search keywords"""
    user_message = state["messages"][-1].content

    understand_prompt = f"""Analyze the user's query: "{user_message}"
Please complete two tasks:
1. Concisely summarize what the user wants to know
2. Generate keywords most suitable for search engines (Chinese or English, must be precise)

Format:
Understanding: [User requirement summary]
Search terms: [Best search keywords]"""

    response = llm.invoke([SystemMessage(content=understand_prompt)])
    response_text = response.content

    # Parse LLM's output, extract search keywords
    search_query = user_message # Default to using original query
    if "Search terms:" in response_text or "搜索词：" in response_text:
        if "Search terms:" in response_text:
            search_query = response_text.split("Search terms:")[1].strip()
        else:
            search_query = response_text.split("搜索词：")[1].strip()

    return {
        "user_query": response_text,
        "search_query": search_query,
        "step": "understood",
        "messages": [AIMessage(content=f"I will search for you: {search_query}")]
    }

This node uses a structured prompt to require the LLM to simultaneously complete two tasks: "intent understanding" and "keyword generation," and updates the parsed dedicated search keywords to the state's search_query field, preparing for the next step of precise search.

(2) Search Node

This node is responsible for executing the agent's "tool usage" capability. It will call the Tavily API for real internet search and has basic error handling functionality.

def tavily_search_node(state: SearchState) -> dict:
    """Step 2: Use Tavily API for real search"""
    search_query = state["search_query"]
    try:
        print(f"🔍 Searching: {search_query}")
        response = tavily_client.search(
            query=search_query, search_depth="basic", max_results=5, include_answer=True
        )
        # ... (process and format search results) ...
        search_results = ... # Formatted result string

        return {
            "search_results": search_results,
            "step": "searched",
            "messages": [AIMessage(content="✅ Search completed! Organizing answer...")]
        }
    except Exception as e:
        # ... (handle error) ...
        return {
            "search_results": f"Search failed: {e}",
            "step": "search_failed",
            "messages": [AIMessage(content="❌ Search encountered a problem...")]
        }

This node initiates a real API call through tavily_client.search. It is wrapped in a try...except block to catch possible exceptions. If the search fails, it updates the step state to "search_failed", which will be used by the next node to trigger a fallback plan.

(3) Answer Node

The final answer node can choose different answering strategies based on whether the previous search was successful, possessing a certain degree of flexibility.

def generate_answer_node(state: SearchState) -> dict:
    """Step 3: Generate final answer based on search results"""
    if state["step"] == "search_failed":
        # If search failed, execute fallback strategy, answer based on LLM's own knowledge
        fallback_prompt = f"Search API is temporarily unavailable, please answer the user's question based on your knowledge:\nUser question: {state['user_query']}"
        response = llm.invoke([SystemMessage(content=fallback_prompt)])
    else:
        # Search successful, generate answer based on search results
        answer_prompt = f"""Provide a complete and accurate answer to the user based on the following search results:
User question: {state['user_query']}
Search results:\n{state['search_results']}
Please synthesize the search results and provide an accurate, useful answer..."""
        response = llm.invoke([SystemMessage(content=answer_prompt)])

    return {
        "final_answer": response.content,
        "step": "completed",
        "messages": [AIMessage(content=response.content)]
    }

This node executes conditional logic by checking the value of state["step"]. If the search fails, it will use the LLM's internal knowledge to answer and inform the user of the situation. If the search succeeds, it will use a prompt containing real-time search results to generate a timely and evidence-based answer.

(4) Build Graph

We connect all nodes together.

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

def create_search_assistant():
    workflow = StateGraph(SearchState)

    # Add nodes
    workflow.add_node("understand", understand_query_node)
    workflow.add_node("search", tavily_search_node)
    workflow.add_node("answer", generate_answer_node)

    # Set linear process
    workflow.add_edge(START, "understand")
    workflow.add_edge("understand", "search")
    workflow.add_edge("search", "answer")
    workflow.add_edge("answer", END)

    # Compile graph
    memory = InMemorySaver()
    app = workflow.compile(checkpointer=memory)
    return app

(5) Running Case Demonstration

After running this script, you can ask some questions that require real-time information, such as the case in our first chapter: I'm going to Beijing tomorrow, what's the weather like? Are there suitable attractions?

You will see the terminal clearly display the agent's "thinking" process:

🔍 Intelligent Search Assistant Started!
I will use Tavily API to search for the latest and most accurate information for you
Supports various questions: news, technology, knowledge Q&A, etc.
(Enter 'quit' to exit)

🤔 What would you like to know: I'm going to Beijing tomorrow, what's the weather like? Are there suitable attractions?

============================================================
🧠 Understanding phase: I understand your needs: Understanding: The user wants to know about tomorrow's weather in Beijing and suitable attraction recommendations.
Search terms: Beijing tomorrow weather attraction recommendations Beijing weather tomorrow attractions
🔍 Searching: Beijing tomorrow weather attraction recommendations Beijing weather tomorrow attractions
🔍 Search phase: ✅ Search completed! Found relevant information, organizing answer for you...

💡 Final Answer:
Tomorrow (September 17, 2025) Beijing's weather forecast shows it is expected to be cloudy, with temperatures ranging from 17°C (62°F) to 25°C (77°F). This mild weather is very suitable for outdoor activities.

### Suitable Attraction Recommendations:
1. **Great Wall**: As one of China's most famous historical sites, the Great Wall is a must-visit. You can choose popular sections like Badaling or Mutianyu for your tour.

2. **Forbidden City**: The Forbidden City was the imperial palace of the Ming and Qing dynasties, with rich history and culture, suitable for tourists interested in Chinese history.

3. **Tiananmen Square**: This is one of China's symbols, with many important buildings and monuments on the square, suitable for taking photos.

4. **Summer Palace**: A very beautiful royal garden, suitable for strolling and enjoying natural scenery, especially the lakes and ancient buildings.

5. **798 Art District**: If you're interested in modern art, the 798 Art District is a place that integrates art, culture, and creativity, suitable for exploration and photography.

### Tips:
- Since tomorrow's weather is good, it's recommended to plan your travel route in advance and prepare some water and snacks to maintain sufficient energy during the tour.
- Since weather changes may affect the tour experience, it's recommended to check real-time weather updates.

Hope this information helps you arrange a pleasant Beijing trip! If you need more information about attractions or travel advice, feel free to ask anytime.

============================================================

🤔 What would you like to know:

And it is a continuously interactive assistant, you can continue to ask questions.

6.5.3 Analysis of LangGraph's Advantages and Limitations

Any technical framework has its specific applicable scenarios and design trade-offs. In this section, we will objectively analyze LangGraph's core advantages and the limitations it may face in practical applications.

(1) Advantages

• As shown in our intelligent search assistant case, LangGraph explicitly defines a complete real-time Q&A process as a "flowchart" composed of states, nodes, and edges. The greatest advantage of this design is high controllability and predictability. Developers can precisely plan every step of the agent's behavior, which is crucial for building production-level applications that require high reliability and auditability. Its most powerful feature lies in native support for cycles. Through conditional edges, we can easily build "reflection-correction" loops. For example, in our case, if the search fails, we can design a path to fall back to a backup plan. This is key to building agents capable of self-optimization and fault tolerance.

• In addition, since each node is an independent Python function, this brings high modularity. At the same time, inserting a node waiting for human review in the process becomes very straightforward, providing a solid foundation for implementing reliable "Human-in-the-loop" collaboration.

(2) Limitations

• Compared to conversation-based frameworks, LangGraph requires developers to write more boilerplate code. Defining states, nodes, edges, and a series of operations makes the development process more cumbersome for simple tasks. Developers need to think more about "how to control the process (how)" rather than just "what to do (what)". Since the workflow is predefined, LangGraph's behavior is controllable but also lacks the dynamic, "emergent" interaction of conversational agents. Its strength lies in executing a determined, reliable process, rather than simulating open-ended, unpredictable social collaboration.

• The debugging process also presents challenges. Although the process is clearer than conversation history, problems may occur at multiple points: logical errors within a node, mutations in state data passed between nodes, or mistakes in edge transition condition judgments. This requires developers to have a global understanding of the entire graph's operating mechanism.