YouTip

LangGraph is a **low-level Agent orchestration framework** developed by the LangChain team, designed for building stateful, long-running AI workflows. Unlike traditional linear LLM call chains, LangGraph models workflows as a **Directed Graph**: * **Node**: Functions that execute specific operations (e.g., calling LLM, executing tools, processing data) * **Edge**: Defines the flow paths between nodes, supporting conditional branching * **State**: Data that is shared and passed throughout the workflow Open source address: **[https://github.com/langchain-ai/langgraph](https://github.com/langchain-ai/langgraph)**. !(#) > Imagine you are conducting a symphony orchestra: a traditional LLM Chain is like playing a piece from beginning to end, only sequentially; while LangGraph is like a conductor who can adjust the playing order at any time based on the live audience's reaction, repeat a movement, or jump to a specific section. It gives AI workflows the "wisdom of conducting" — capable of looping, branching, and backtracking, truly achieving complex autonomous decision-making. ### Why Choose LangGraph? The following table compares the main differences between traditional LLM Chain and LangGraph: | Feature | Traditional LLM Chain | LangGraph | | --- | --- | --- | | **Workflow Structure** | Linear, unidirectional execution | Graph structure, supports loops | | **State Management** | Manual management required | Built-in state persistence | | **Conditional Routing** | Complex to implement | Natively supported | | **Human-in-the-loop** | Requires additional development | Built-in interrupt support | | **Multi-Agent Coordination** | Difficult to implement | First-class support | | **Debugging Tools** | Limited | LangGraph Studio | ### Applicable Scenarios * **Chatbots**: Requires memory of multi-turn conversation context * **Autonomous Agents**: Capable of planning, using tools, and iterative thinking * **Multi-Agent Systems**: Multiple AIs collaborating to complete complex tasks * **Approval Workflows**: Automated processes requiring human review * **Research Assistants**: Multi-step reasoning and information retrieval required * * * ## Core Concepts Before writing code, let's understand the three core concepts of LangGraph. ### Graph Graph is the blueprint of the entire workflow, defining the complete logical structure of the Agent. It consists of Nodes and Edges: StateGraph |-- Nodes | |-- node_a | |-- node_b | +-- node_c +-- Edges |-- START -> node_a |-- node_a -> node_b (conditional edge) |-- node_a -> node_c (conditional edge) +-- node_b -> END ### State State is the **shared data structure** that runs through the entire graph. Each node can read and update State, and the updated State is passed to the next node. from typing import TypedDict, Annotatedfrom langgraph.graph import add_messages class MyState(TypedDict): messages: Annotated[list, add_messages] # message list (auto-append) user_name: str # user name step_count: int # step count **Important:** `Annotated[list, add_messages]` indicates that this field uses `add_messages` as the reducer — new messages are appended to the list rather than overwriting it. This is the core mechanism of LangGraph's state management. ### Nodes Nodes are ordinary Python functions that receive the current State and return the updated State (partial fields). def my_node(state: MyState) -> dict: # read state messages = state # execute operation... result = "processing result" # return updated fields (no need to return all fields) return {"messages": [{"role": "ai", "content": result}]} ### Edges Edges define how nodes flow to each other: * **Normal Edge**: Fixed path, `node_a -> node_b` * **Conditional Edge**: Dynamic routing based on State, `node_a -> node_b or node_c` * **Start Edge**: `START -> first node` * **End Edge**: `some node -> END` * * * ## Environment Setup ### Install Dependencies Use domestic mirror to install LangGraph and related dependencies: # create virtual environment (recommended) python -m venv venv source venv/bin/activate # macOS/Linux# venvScriptsactivate # Windows# install LangGraph and LangChain pip install langgraph langchain langchain-openai python-dotenv -i https://mirrors.aliyun.com/pypi/simple/# optional: install development tools pip install langgraph-cli jupyter -i https://mirrors.aliyun.com/pypi/simple/ ### Configure API Key Create `.env` file in the project root directory, configuring both OpenAI and DeepSeek: # .env file content# OpenAI configuration (for overseas users) OPENAI_API_KEY=sk-xxx OPENAI_BASE_URL=https://api.openai.com/v1# DeepSeek configuration (recommended for domestic users) DEEPSEEK_API_KEY=sk-xxx DEEPSEEK_BASE_URL=https://api.deepseek.com DEEPSEEK_MODEL=deepseek-v4-pro **Tip:** The examples in this article support both OpenAI and DeepSeek. DeepSeek is recommended for domestic users, with API address configured as `https://api.deepseek.com` (without /v1), LangChain will automatically append the path. DeepSeek API Key can be created at [https://platform.deepseek.com/api_keys](https://platform.deepseek.com/api_keys). ### Verify Installation import langgraph print(f"LangGraph version: {langgraph.__version__}") * * * ## First LangGraph Program Let's start with the simplest example — a linear workflow with only two nodes. ## Example from langgraph.graph import StateGraph, START, END from typing import TypedDict # Step 1: Define State class SimpleState(TypedDict): message: str processed: bool # Step 2: Define node functions def greet_node(state: SimpleState) ->dict: """Greeting node: generate greeting""" print(f" Received message: {state['message']}") return{"message": f"Hello! {state['message']}"} def process_node(state: SimpleState) ->dict: """Processing node: mark as processed""" print(f" Processing message: {state['message']}") return{"processed": True} # Step 3: Build graph builder = StateGraph(SimpleState) # Add nodes builder.add_node("greet", greet_node) builder.add_node("process", process_node) # Add edges builder.add_edge(START,"greet") builder.add_edge("greet","process") builder.add_edge("process", END) # Step 4: Compile graph graph = builder.compile() # Step 5: Run result = graph.invoke({ "message": "World", "processed": False }) print(f"n Final result: {result}") Run result: Received message: World Processing message: Hello! WorldFinal result: {'message': 'Hello! World', 'processed': True} ### Visualize Graph Structure In Jupyter Notebook, you can directly visualize the graph structure: # Visualize in Jupyter Notebookfrom IPython.display import ImageImage(graph.get_graph().draw_mermaid_png())# Or print Mermaid formatprint(graph.get_graph().draw_mermaid()) * * * ## State Management ### Define State Using TypedDict from typing import TypedDict, Annotated, Optionalfrom langgraph.graph.message import add_messages class AgentState(TypedDict): # message history (add_messages reducer auto-appends instead of overwriting) messages: Annotated[list, add_messages] # normal fields (directly overwritten) user_id: str session_id: str # optional fields error: Optional # counter (using operator.add as reducer) retry_count: Annotated[int, lambda x, y: x + y] ### Define State Using Pydantic (Recommended for Production) from pydantic import BaseModel, Fieldfrom typing import Annotatedfrom langgraph.graph.message import add_messages class ProductionState(BaseModel): messages: Annotated[list, add_messages] = Field(default_factory=list) user_id: str = "" confidence_score: float = 0.0 class Config: arbitrary_types_allowed = True ### MessagesState (Built-in Shortcut State) LangGraph provides a built-in `MessagesState`, specifically designed for conversation scenarios: from langgraph.graph import MessagesState# MessagesState is equivalent to:# class MessagesState(TypedDict):# messages: Annotated[list, add_messages]# Use directly, no need to customize builder = StateGraph(MessagesState) * * * ## Nodes ### Normal Function Node def simple_node(state: AgentState) -> dict: # read state last_message = state # execute operation response = f"Received: {last_message.content}" # return partial state update return { "messages": [{"role": "assistant", "content": response}] } ### LLM Call Node ## Example import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage load_dotenv() # Use DeepSeek model llm = ChatOpenAI( model=os.getenv('DEEPSEEK_MODEL','deepseek-v4-pro'), openai_api_key=os.getenv('DEEPSEEK_API_KEY'), openai_api_base=os.getenv('DEEPSEEK_BASE_URL','https://api.deepseek.com'), temperature=0 ) def llm_node(state: dict) ->dict: """Node that calls LLM""" system_prompt = SystemMessage(content="You are a helpful assistant.") # Merge system prompt with conversation history messages = + state # Call LLM response = llm.invoke(messages) return{"messages": } ### Async Node import asyncio async def async_node(state: AgentState) -> dict: """Async node, suitable for I/O-intensive operations""" # simulate async operation (e.g., API call, database query) await asyncio.sleep(0.1) result = await some_async_api_call(state.content) return {"messages": [{"role": "assistant", "content": result}]}# Use async graph result = await graph.ainvoke({"messages": [...]}) ### Using Class as Node class RouterNode: def __init__(self, llm, system_prompt: str): self.llm = llm self.system_prompt = system_prompt def __call__(self, state: AgentState) -> dict: """Class instance can be used as node""" messages = [ SystemMessage(content=self.system_prompt), *state ] response = self.llm.invoke(messages) return {"messages": }# Add class node router = RouterNode(llm, "You are a professional routing assistant.") builder.add_node("router", router) * * * ## Edges and Conditional Routing ### Normal Edge # Fixed path: after node_a completes, always execute node_b builder.add_edge("node_a", "node_b")# End: after node_a completes, graph ends builder.add_edge("node_a", END) ### Conditional Edge Conditional edges are the core feature of LangGraph, dynamically deciding the next step based on current State. def route_after_llm(state: AgentState) -> str: """ Routing function: decide which path to take based on LLM's latest output Return value must be a registered node name or END """ last_message = state # If LLM requests to use tools if hasattr(last_message, "tool_calls") and last_message.tool_calls: return "tools" # Otherwise end return END# Add conditional edge builder.add_conditional_edges( "llm", # source node route_after_llm, # routing function { "tools": "tool_executor", # when "tools" is returned -> tool_executor node END: END # when END is returned -> end }) ### Parallel Execution (Fan-out) # Parallel fork from one node to multiple nodes builder.add_edge("start_node", "branch_a") builder.add_edge("start_node", "branch_b") builder.add_edge("start_node", "branch_c")# Multiple nodes converge to one node (Fan-in