Introduction to LangGraph: Building Stateful LLM Workflows with Nodes and Edges

 

What is LangGraph?

LangGraph is a framework for building stateful workflows and agents around LLMs.

In a typical LLM application, we may need to perform several steps. For example, the LLM may receive a question, decide that it needs a tool, call the tool, receive the result, and then generate a final answer.

LangGraph helps us represent this workflow as a graph.

A LangGraph workflow mainly consists of:

  • State — contains the information carried through the workflow.

  • Nodes — perform individual tasks, such as calling an LLM or executing a tool.

  • Edges — define how the workflow moves from one node to another.

A simple workflow can therefore look like this:

START
LLM Node
Tool Node
END

The state travels through these nodes as the workflow executes.

For example, if our state contains a messages list, the LLM can add an AIMessage to it and a tool can add a ToolMessage. The state therefore keeps track of what has happened during the workflow.

In simple terms:

LangGraph provides a way to organize an LLM application as a graph of connected steps, while maintaining the state shared between those steps.

The name LangGraph comes from this idea of representing the workflow as a graph, where nodes represent work and edges represent the flow between that work.


LangGraph version


       ┌─────────┐

       │  START           │

       └────┬────┘

            ↓

       ┌─────────┐

       │   LLM Node    │

       └────┬────┘

            ↓

       ┌──────────┐

       │   Tool Node        │

       └────┬─────┘

            ↓

          END


We have not changed anything sginificant here from our pervious blog (AI learning lang state with LangChain) we  intoduce 

This line ( builder = StateGraph(LangState)) creates a LangGraph graph builder and tells it that the graph will use LangState as its state structure.

Here, LangState defines the information that will be carried through the workflow.

At this point, the graph has not yet been fully defined. We have only created the builder. We can then add nodes and edges to define what the workflow should do and how the different steps should be connected.

We are telling LangGraph: This graph will use LangState as its state.

Then:

# Add nodes
builder.add_node("llm", call_llm)
builder.add_node("tools", call_tools)

So we have added  two nodes:

LLM node
Tool node

and three edges

# Add edges
builder.add_edge(START, "llm")
builder.add_edge("llm", "tools")
builder.add_edge("tools", END)

creates the flow:

START
  ↓
LLM
  ↓
TOOLS
  ↓
END

One important thing for learning

We deliberately made this simple, even though a production agent normally needs a conditional decision such as:

LLM
 ├── tool call → Tools → LLM
 └── final answer → END

Don't add that yet.

First let's understand this simple graph:

START → LLM → TOOLS → END

because now we can directly map what we already learned:

LangState
    ↓
StateGraph
    ↓
Nodes
    ↓
Edges
    ↓
Compiled Graph
graph = builder.compile()
↓ graph.invoke(state)

final_state = graph.invoke(initial_state)

here we are calling the initail state  and final state

print("\nInitial State:")
print(initial_state)

final_state = graph.invoke(initial_state)

print("\nFinal State:")
print(final_state)

This is the natural next step from the code you've already written.


Here is the complete code .


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

from dotenv import load_dotenv
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_core.tools import tool
from langchain_google_genai import ChatGoogleGenerativeAI


# Load environment variables
load_dotenv()


# -------------------------------------------------------------
# 1. Define Custom Python Tools
# -------------------------------------------------------------
@tool
def calculate_compound_interest(
    principal: float,
    annual_rate: float,
    years: int
) -> str:
    """Calculates compound interest."""

    amount = principal * ((1 + (annual_rate / 100)) ** years)
    interest = amount - principal

    return json.dumps({
        "principal": principal,
        "interest_earned": round(interest, 2),
        "total_amount": round(amount, 2),
        "years": years,
    })


@tool
def get_stock_price(ticker: str) -> str:
    """Fetches the mock current stock price."""

    mock_prices = {
        "AAPL": 220.50,
        "GOOGL": 175.30,
        "MSFT": 415.00
    }

    price = mock_prices.get(ticker.upper(), 100.00)

    return json.dumps({
        "ticker": ticker.upper(),
        "price_usd": price
    })


# -------------------------------------------------------------
# 2. Tools
# -------------------------------------------------------------
tools_list = [
    calculate_compound_interest,
    get_stock_price
]

tools_by_name = {
    t.name: t for t in tools_list
}


# -------------------------------------------------------------
# 3. Initialize Gemini
# -------------------------------------------------------------
llm = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash",
    temperature=0.0,
)

llm_with_tools = llm.bind_tools(tools_list)


# =============================================================
# 4. LANGSTATE
# =============================================================

class LangState(TypedDict):
    messages: list


# =============================================================
# 5. NODE 1 — LLM
# =============================================================

def call_llm(state: LangState):

    print("\n--- LLM NODE ---")

    ai_response = llm_with_tools.invoke(
        state["messages"]
    )

    return {
        "messages": [ai_response]
    }


# =============================================================
# 6. NODE 2 — TOOL
# =============================================================

def call_tools(state: LangState):

    print("\n--- TOOL NODE ---")

    last_message = state["messages"][-1]

    tool_messages = []

    for tool_call in last_message.tool_calls:

        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        tool_id = tool_call["id"]

        print("Tool:", tool_name)
        print("Arguments:", tool_args)

        selected_tool = tools_by_name[tool_name]

        tool_result = selected_tool.invoke(tool_args)

        print("Tool Result:", tool_result)

        tool_messages.append(
            ToolMessage(
                content=str(tool_result),
                tool_call_id=tool_id
            )
        )

    return {
        "messages": tool_messages
    }


# =============================================================
# 7. CREATE LANGGRAPH
# =============================================================

builder = StateGraph(LangState)


# Add nodes
builder.add_node("llm", call_llm)
builder.add_node("tools", call_tools)


# Add edges
builder.add_edge(START, "llm")
builder.add_edge("llm", "tools")
builder.add_edge("tools", END)


# Build the graph
graph = builder.compile()


# =============================================================
# 8. INITIAL LANGSTATE
# =============================================================

user_query = (
    "If I invest $5,000 at a 7% interest rate "
    "for 10 years, how much will I earn?"
)

initial_state: LangState = {
    "messages": [
        HumanMessage(content=user_query)
    ]
}


# =============================================================
# 9. RUN LANGGRAPH
# =============================================================

print("\nInitial State:")
print(initial_state)

final_state = graph.invoke(initial_state)

print("\nFinal State:")
print(final_state)

Comments

Popular posts from this blog

Aggregate function with spring data

Java Persistence API with Spring Data

NodeJS vs Java