Graph Basics: Nodes, Edges & State

Every LangGraph workflow is built from three fundamental concepts: Nodes, Edges, and State. Think of it like building with LEGO - once you understand these three pieces, you can construct any workflow.

The Three Core Concepts

Understanding State

State is the central data store for your workflow. It's a typed dictionary that represents everything the workflow needs to know. Each node can read from and update this state.

from typing import TypedDict, Literal
from typing_extensions import Annotated

class EcommerceState(TypedDict):
    # Customer information
    customer_id: str
    customer_tier: Literal["standard", "premium", "vip"]

    # Order details
    order_id: str
    items: list[dict]
    subtotal: float
    discount: float
    total: float

    # Workflow tracking
    status: str
    messages: list[str]  # Audit trail

Understanding Nodes

A Node is a Python function that:

• Receives the current state
• Performs some action (API call, LLM call, computation)
• Returns an updated state (partial updates are merged)

from langgraph.graph import StateGraph, START, END

# Node 1: Calculate subtotal
def calculate_subtotal(state: EcommerceState) -> EcommerceState:
    subtotal = sum(item["price"] * item["qty"] for item in state["items"])
    return {
        "subtotal": subtotal,
        "messages": state["messages"] + [f"Subtotal calculated: ${subtotal:.2f}"]
    }

# Node 2: Apply discount based on customer tier
def apply_discount(state: EcommerceState) -> EcommerceState:
    discount_rates = {"standard": 0, "premium": 0.10, "vip": 0.20}
    discount = state["subtotal"] * discount_rates[state["customer_tier"]]
    return {
        "discount": discount,
        "total": state["subtotal"] - discount,
        "messages": state["messages"] + [f"Discount applied: ${discount:.2f}"]
    }

# Node 3: Finalize order
def finalize_order(state: EcommerceState) -> EcommerceState:
    return {
        "status": "completed",
        "messages": state["messages"] + [f"Order finalized. Total: ${state['total']:.2f}"]
    }

Understanding Edges

Edges define how state flows from one node to another. There are three types:

1. Normal Edges (Static)

Always flow from Node A to Node B:

workflow = StateGraph(EcommerceState)

workflow.add_node("calculate", calculate_subtotal)
workflow.add_node("discount", apply_discount)
workflow.add_node("finalize", finalize_order)

# Static edges - always follow this path
workflow.add_edge(START, "calculate")
workflow.add_edge("calculate", "discount")
workflow.add_edge("discount", "finalize")
workflow.add_edge("finalize", END)

This creates a linear flow:

START → calculate → discount → finalize → END

2. Conditional Edges (Dynamic)

Route to different nodes based on state:

def route_by_order_value(state: EcommerceState) -> str:
    """Route high-value orders for manual review"""
    if state["total"] > 500:
        return "manual_review"
    else:
        return "auto_approve"

# Add conditional edge
workflow.add_conditional_edges(
    "discount",  # From this node
    route_by_order_value,  # Use this function to decide
    {
        "manual_review": "review_node",  # If returns "manual_review", go here
        "auto_approve": "finalize"       # If returns "auto_approve", go here
    }
)

3. START and END Edges

Special edges that mark where the graph begins and ends:

from langgraph.graph import START, END

workflow.add_edge(START, "first_node")  # Entry point
workflow.add_edge("last_node", END)     # Exit point

Complete E-commerce Example

Let's build a complete order processing workflow with conditional routing:

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Literal

class OrderState(TypedDict):
    customer_id: str
    customer_tier: Literal["standard", "premium", "vip"]
    items: list[dict]
    subtotal: float
    discount: float
    total: float
    payment_method: Literal["card", "cod"]  # Cash on delivery
    status: str
    requires_review: bool

# Node functions
def validate_order(state: OrderState) -> dict:
    """Validate order has items and customer exists"""
    if not state["items"]:
        return {"status": "error: no items"}
    return {"status": "validated"}

def calculate_total(state: OrderState) -> dict:
    """Calculate subtotal and apply tier-based discount"""
    subtotal = sum(item["price"] * item["qty"] for item in state["items"])
    discount_rates = {"standard": 0, "premium": 0.10, "vip": 0.20}
    discount = subtotal * discount_rates.get(state["customer_tier"], 0)
    total = subtotal - discount

    return {
        "subtotal": subtotal,
        "discount": discount,
        "total": total,
        "requires_review": total > 500
    }

def process_card_payment(state: OrderState) -> dict:
    """Process credit card payment"""
    # In reality, call Stripe/PayPal API here
    return {"status": "payment_processed"}

def process_cod(state: OrderState) -> dict:
    """Set up cash on delivery"""
    return {"status": "cod_scheduled"}

def manual_review(state: OrderState) -> dict:
    """Flag for manual review"""
    return {"status": "pending_review"}

def fulfill_order(state: OrderState) -> dict:
    """Send to fulfillment"""
    return {"status": "fulfilled"}

# Routing functions
def route_payment(state: OrderState) -> str:
    """Route based on payment method"""
    return "card" if state["payment_method"] == "card" else "cod"

def route_review(state: OrderState) -> str:
    """Route high-value orders for review"""
    return "review" if state["requires_review"] else "fulfill"

# Build the graph
workflow = StateGraph(OrderState)

# Add all nodes
workflow.add_node("validate", validate_order)
workflow.add_node("calculate", calculate_total)
workflow.add_node("card_payment", process_card_payment)
workflow.add_node("cod_payment", process_cod)
workflow.add_node("review", manual_review)
workflow.add_node("fulfill", fulfill_order)

# Add edges
workflow.add_edge(START, "validate")
workflow.add_edge("validate", "calculate")

# Conditional: route based on payment method
workflow.add_conditional_edges(
    "calculate",
    route_payment,
    {"card": "card_payment", "cod": "cod_payment"}
)

# Both payment paths lead to review check
workflow.add_conditional_edges(
    "card_payment",
    route_review,
    {"review": "review", "fulfill": "fulfill"}
)
workflow.add_conditional_edges(
    "cod_payment",
    route_review,
    {"review": "review", "fulfill": "fulfill"}
)

workflow.add_edge("review", "fulfill")
workflow.add_edge("fulfill", END)

# Compile
app = workflow.compile()

Running Your Graph

# Test with a VIP customer, high-value card order
result = app.invoke({
    "customer_id": "cust_456",
    "customer_tier": "vip",
    "items": [
        {"sku": "LAPTOP-PRO", "price": 999.99, "qty": 1},
        {"sku": "MOUSE-WL", "price": 49.99, "qty": 2}
    ],
    "subtotal": 0,
    "discount": 0,
    "total": 0,
    "payment_method": "card",
    "status": "pending",
    "requires_review": False
})

print(f"Final status: {result['status']}")
print(f"Total after VIP discount: ${result['total']:.2f}")
# Output:
# Final status: fulfilled
# Total after VIP discount: $879.98 (20% VIP discount applied!)

Visualizing Your Graph

LangGraph can generate visual diagrams of your workflow:

# Generate Mermaid diagram
print(app.get_graph().draw_mermaid())

This produces a flowchart you can visualize:

         ┌─────────┐
         │  START  │
         └────┬────┘
              ▼
         ┌─────────┐
         │validate │
         └────┬────┘
              ▼
         ┌─────────┐
         │calculate│
         └────┬────┘
        ╱            ╲
       ▼              ▼
┌───────────┐  ┌───────────┐
│card_payment│ │cod_payment│
└─────┬─────┘  └─────┬─────┘
      └──────┬───────┘
             ▼
      ┌────────────┐
      │review/fulfill│
      └──────┬─────┘
             ▼
         ┌─────┐
         │ END │
         └─────┘

Edge Types Summary

Key Takeaways

Next up: State Management - Advanced patterns for managing conversation history, order context, and session data in e-commerce.