Mastering LangGraph: Building Intelligent Order Management Workflows

Introduction

In this comprehensive guide, we will explore LangGraph—a robust library designed for orchestrating complex, multi-step workflows with Large Language Models (LLMs). We will apply it to a practical e-commerce use case: determining whether to place or cancel an order based on a user’s query.

By the end of this tutorial, you will understand how to:

• Set up LangGraph in a Python environment.
• Load and manage data (e.g., inventory and customer records).
• Define nodes to structure your workflow.
• Build a graph of interconnected nodes with conditional branching.
• Visualize and test the workflow.

We will walk through each step in detail, making it accessible to beginners and useful for those seeking to develop dynamic, intelligent workflows using LLMs. A dataset link is also provided for hands-on experimentation.

---

Table of Contents

1. What Is LangGraph?
2. The Problem Statement: Order Management
3. Environment Setup and Imports
4. Data Loading and State Definition
5. Creating Tools and Integrating LLMs
6. Defining Workflow Nodes
7. Constructing the Workflow Graph
8. Visualizing and Testing the Workflow
9. Conclusion

---

1. What Is LangGraph?

LangGraph is a library that brings a graph-based approach to LangChain workflows. Traditional pipelines follow a linear progression, but real-world tasks often involve branching logic, loops (e.g., retrying failed steps), or human intervention.

Key Features:

• Nodes: Represent individual tasks or functions (e.g., checking inventory, computing shipping costs).
• Edges: Define the flow between nodes, enabling conditional execution.
• Shared State: Nodes can update a global state object dynamically.
• Tool Integration: Supports external tools that LLMs can call.
• Human-in-the-Loop: Allows for manual intervention in workflows when required.

---

2. The Problem Statement: Order Management

The workflow needs to handle two types of user queries:

1. Placing an Order
   • Check item availability.
   • Compute shipping costs.
   • Simulate payment processing.
2. Canceling an Order
   • Extract the order_id from the query.
   • Mark the order as canceled.

Since these operations require decision-making, we will use LangGraph to implement a structured, conditional workflow:

• If a user wants to place an order, we move through inventory checks, shipping computation, and payment processing.
• If a user wants to cancel an order, we extract the order_id and trigger the cancellation process.

---

3. Environment Setup and Imports

import os
import pandas as pd
import random
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.runnables.graph import MermaidDrawMethod
from IPython.display import display, Image
from typing import Literal, Dict, TypedDict

# Load environment variables
os.environ["OPENAI_API_KEY"] = ""

Explanation of Key Imports:

• tool: Converts functions into tools accessible by the LLM.
• ChatOpenAI: Provides access to OpenAI’s GPT models.
• StateGraph: Manages the workflow graph.
• MermaidDrawMethod: Generates a visual representation of the workflow.

---

4. Data Loading and State Definition

Load Inventory and Customer Data

inventory_df = pd.read_csv("inventory.csv")
customers_df = pd.read_csv("customers.csv")

inventory = inventory_df.set_index("item_id").T.to_dict()
customers = customers_df.set_index("customer_id").T.to_dict()

Define the Workflow State

class State(TypedDict):
    query: str
    category: str
    next_node: str
    item_id: str
    order_status: str
    cost: str
    payment_status: str
    location: str
    quantity: int

---

5. Creating Tools and Integrating LLMs

Define the Order Cancellation Tool

@tool
def cancel_order(query: str) -> dict:
    """Simulate order cancellation."""
    order_id = llm.with_structured_output(method='json_mode').invoke(
        f'Extract order_id from the following text in JSON format: {query}'
    )["order_id"]
    
    if not order_id:
        return {"error": "Missing 'order_id'."}
    
    return {"order_status": "Order canceled"}

Initialize LLM and Bind Tools

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [cancel_order]
tool_node = ToolNode(tools)

---

6. Defining Workflow Nodes

Query Categorization

def categorize_query(state: MessagesState) -> MessagesState:
    """Categorize user query into PlaceOrder or CancelOrder."""
    prompt = ChatPromptTemplate.from_template(
        "Categorize user query into PlaceOrder or CancelOrder: {state}"
    )
    
    chain = prompt | ChatOpenAI(temperature=0)
    category = chain.invoke({"state": state}).content
    
    return {"query": state, "category": category}

Check Inventory

def check_inventory(state: MessagesState) -> MessagesState:
    """Check if the requested item is in stock."""
    item_id = inventory.get(state["item_id"], {}).get("stock", 0)
    return {"order_status": "In Stock" if item_id > 0 else "Out of Stock"}

Compute Shipping Costs

def compute_shipping(state: MessagesState) -> MessagesState:
    """Calculate shipping costs based on location and item weight."""
    rates = {"local": 5, "domestic": 10, "international": 20}
    cost = inventory[state["item_id"]]["weight"] * state["quantity"] * rates.get(state["location"], 10)
    return {"cost": f"${cost:.2f}"}

Process Payment

def process_payment(state: State) -> State:
    """Simulate payment processing."""
    payment_outcome = random.choice(["Success", "Failed"])
    return {"payment_status": payment_outcome}

---

7. Constructing the Workflow Graph

workflow = StateGraph(MessagesState)
workflow.add_node("RouteQuery", categorize_query)
workflow.add_node("CheckInventory", check_inventory)
workflow.add_node("ComputeShipping", compute_shipping)
workflow.add_node("ProcessPayment", process_payment)
workflow.add_conditional_edges("RouteQuery", lambda s: "CheckInventory" if s["category"] == "PlaceOrder" else "CancelOrder")
workflow.add_edge("CheckInventory", "ComputeShipping")
workflow.add_edge("ComputeShipping", "ProcessPayment")
workflow.add_edge("ProcessPayment", END)

---

8. Visualizing and Testing the Workflow

agent = workflow.compile()
mermaid_graph = agent.get_graph()
display(Image(mermaid_graph.draw_mermaid_png()))