Building a Simple Graph

Beginner⏱35 minutes📖3 min readLesson 5 of 10

Building a Simple Graph

Build a simple 3-step LangGraph: input processing, transformation, and output generation. Learn the complete build-compile-invoke workflow.

In this lesson, we'll build a complete 3-step LangGraph: input → process → output. This is the foundation for every agent you'll build.

The 3-Step Pattern

Every LangGraph application follows this pattern:

Input node: Accepts and validates raw input
Process node: Performs the core logic (LLM call, computation, etc.)
Output node: Formats and returns the final result

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Step 1: Define the State

python

from typing_extensions import TypedDict
from typing import Optional

class SimpleState(TypedDict):
    input_text: str          # Raw user input
    processed_text: str      # Intermediate processed value
    output_text: str         # Final output
    error: Optional[str]     # Error message (if any)

ℹ️Note

Include an error field in your state from the start. It makes error handling much cleaner as your graph grows.

Step 2: Define the Nodes

Input Node

python

def input_node(state: SimpleState) -> dict:
    raw = state["input_text"].strip()

    if not raw:
        return {"error": "Input cannot be empty"}

    return {"input_text": raw}

Process Node

python

def process_node(state: SimpleState) -> dict:
    if state.get("error"):
        return {}  # Skip processing if there's an error

    # Simple text transformation
    processed = state["input_text"].upper()
    word_count = len(state["input_text"].split())

    return {
        "processed_text": f"[{word_count} words] {processed}"
    }

Output Node

python

def output_node(state: SimpleState) -> dict:
    if state.get("error"):
        return {"output_text": f"Error: {state['error']}"}

    return {
        "output_text": f"Result: {state['processed_text']}"
    }

💡Tip

The pattern of checking state.get("error") in each node is a basic form of error propagation. Later we'll replace this with conditional edges for cleaner routing.

Step 3: Build the Graph

python

from langgraph.graph import StateGraph, START, END

builder = StateGraph(SimpleState)

# Add nodes
builder.add_node("input", input_node)
builder.add_node("process", process_node)
builder.add_node("output", output_node)

# Add edges
builder.add_edge(START, "input")
builder.add_edge("input", "process")
builder.add_edge("process", "output")
builder.add_edge("output", END)

# Compile
app = builder.compile()

Step 4: Invoke the Graph

python

# Successful execution
result = app.invoke({
    "input_text": "hello world",
    "processed_text": "",
    "output_text": "",
    "error": None
})

print(result["output_text"])
# Result: [2 words] HELLO WORLD

# Error case
result = app.invoke({
    "input_text": "   ",
    "processed_text": "",
    "output_text": "",
    "error": None
})

print(result["output_text"])
# Error: Input cannot be empty

Step 5: Add Streaming

Streaming lets you observe each node's output as it executes:

python

for event in app.stream({
    "input_text": "langgraph is awesome",
    "processed_text": "",
    "output_text": "",
    "error": None
}):
    for node_name, state_update in event.items():
        if node_name == "__end__":
            continue

        print(f"---[{node_name}]---")
        for key, value in state_update.items():
            if value:
                print(f"  {key}: {value}")

Output:

---[input]---
  input_text: langgraph is awesome
---[process]---
  processed_text: [3 words] LANGGRAPH IS AWESOME
---[output]---
  output_text: Result: [3 words] LANGGRAPH IS AWESOME

✅Success

Streaming gives you real-time visibility into your graph's execution. Use it during development to verify each node's behavior.

Complete Working Example

python

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from typing import Optional

# 1. State
class SimpleState(TypedDict):
    input_text: str
    processed_text: str
    output_text: str
    error: Optional[str]

# 2. Nodes
def input_node(state: SimpleState) -> dict:
    raw = state["input_text"].strip()
    if not raw:
        return {"error": "Input cannot be empty"}
    return {"input_text": raw}

def process_node(state: SimpleState) -> dict:
    if state.get("error"):
        return {}
    processed = state["input_text"].upper()
    word_count = len(state["input_text"].split())
    return {"processed_text": f"[{word_count} words] {processed}"}

def output_node(state: SimpleState) -> dict:
    if state.get("error"):
        return {"output_text": f"Error: {state['error']}"}
    return {"output_text": f"Result: {state['processed_text']}"}

# 3. Graph
builder = StateGraph(SimpleState)
builder.add_node("input", input_node)
builder.add_node("process", process_node)
builder.add_node("output", output_node)
builder.add_edge(START, "input")
builder.add_edge("input", "process")
builder.add_edge("process", "output")
builder.add_edge("output", END)

app = builder.compile()

# 4. Run
result = app.invoke({
    "input_text": "hello langgraph",
    "processed_text": "",
    "output_text": "",
    "error": None
})
print(result["output_text"])
# Result: [2 words] HELLO LANGGRAPH

Adding an LLM to the Process Node

Let's upgrade the process node to use an LLM:

python

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o", temperature=0.3)

def process_with_llm(state: SimpleState) -> dict:
    if state.get("error"):
        return {}

    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a text analyzer. Analyze the given text and provide:\n"
                   "1. A summary (1 sentence)\n"
                   "2. Sentiment (positive/negative/neutral)\n"
                   "3. Key topics"),
        ("human", "{text}")
    ])

    chain = prompt | llm | StrOutputParser()
    analysis = chain.invoke({"text": state["input_text"]})

    return {"processed_text": analysis}

Now instead of simple string transformation, your process node performs AI-powered analysis.

ℹ️Note

Swapping a deterministic node for an LLM-powered node requires no changes to the graph structure. Only the node function changes. This is the power of the graph abstraction.

Adding a Loop (Preview)

Even in a simple graph, you can add a loop. Let's make the process node repeat until the text is clean:

python

from langgraph.graph import START, END, StateGraph
from typing_extensions import TypedDict

class CleanState(TypedDict):
    text: str
    cleaned: bool
    attempts: int

def clean_text(state: CleanState) -> dict:
    original = state["text"]
    cleaned = original.strip().lower()
    is_clean = cleaned == original

    return {
        "text": cleaned,
        "cleaned": is_clean,
        "attempts": state["attempts"] + 1
    }

def should_continue(state: CleanState) -> str:
    if state["cleaned"] or state["attempts"] >= 3:
        return "end"
    return "continue"

builder = StateGraph(CleanState)
builder.add_node("clean", clean_text)
builder.add_edge(START, "clean")
builder.add_conditional_edges(
    "clean",
    should_continue,
    {
        "continue": "clean",  # Loop back
        "end": END
    }
)

app = builder.compile()

result = app.invoke({"text": "  HELLO WORLD  ", "cleaned": False, "attempts": 0})
print(result["text"])      # hello world
print(result["attempts"])  # 2 (first pass cleans, second confirms)

⚠️Warning

Always have a termination condition in loops. Without the attempts >= 3 check, a bug could cause an infinite loop. Always set recursion_limit in invocation config.

Testing Your Graph

python

# Test 1: Normal input
result = app.invoke({"input_text": "Test", "processed_text": "", "output_text": "", "error": None})
assert "Error" not in result["output_text"]

# Test 2: Empty input
result = app.invoke({"input_text": "", "processed_text": "", "output_text": "", "error": None})
assert "Error" in result["output_text"]

# Test 3: Whitespace input
result = app.invoke({"input_text": "   ", "processed_text": "", "output_text": "", "error": None})
assert "Error" in result["output_text"]

💡Tip

Write tests for each node individually (pure function tests) and for the full graph (integration tests). This catches both node-level bugs and topology issues.

Common Mistakes

Mistake 1: Forgetting to handle the error case

python

def process_node(state: State) -> dict:
    # BUG: If state has an error, this still runs
    return {"result": expensive_computation(state["input"])}

# FIX: Check for errors first
def process_node(state: State) -> dict:
    if state.get("error"):
        return {}
    return {"result": expensive_computation(state["input"])}

Mistake 2: Mutating state directly

python

def bad_node(state: State) -> dict:
    state["value"] = "new"  # BUG: Don't mutate state!
    return {"value": "new"}  # CORRECT: Return updates

def good_node(state: State) -> dict:
    return {"value": "new"}  # CORRECT

Mistake 3: Missing edge to END

python

builder.add_edge("process", "output")
# BUG: No edge from output to END — graph never terminates!

builder.add_edge("output", END)  # FIX