Tutorial Lengkap LangGraph: Membangun AI Agents yang Kompleks

LangGraph adalah library dari LangChain untuk membangun stateful, multi-actor applications dengan Large Language Models (LLMs). Dengan LangGraph, Anda dapat membuat AI agents yang kompleks dengan kontrol flow yang fleksibel, state management, dan kemampuan untuk menjalankan multiple agents secara bersamaan.

Apa itu LangGraph?

LangGraph memperluas LangChain dengan menyediakan:

• Stateful graphs: Menyimpan dan mengelola state antar langkah
• Cyclic flows: Mendukung loops dan kondisional
• Human-in-the-loop: Interaksi manual di tengah workflow
• Persistence: Menyimpan state untuk resume later
• Streaming: Real-time streaming responses

• Membangun AI agents dengan multi-step reasoning
• Workflows yang membutuhkan kondisional dan loops
• Applications yang butuh human approval
• Multi-agent systems
• Complex RAG pipelines

Instalasi

pip install langgraph langchain langchain-openai

Konsep Dasar

1. Graph Components

LangGraph terdiri dari:

• State: Data yang dikelola sepanjang execution
• Nodes: Functions yang memproses dan memodifikasi state
• Edges: Koneksi antar nodes (conditional atau fixed)

2. Basic Graph Structure

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

Define state
class State(TypedDict):
    messages: list
    currentstep: str


Define nodes
def nodea(state: State) -> State:
    state["messages"].append("Processed by Node A")
    state["currentstep"] = "a"

    return state

def nodeb(state: State) -> State:
    state["messages"].append("Processed by Node B")
    state["currentstep"] = "b"

    return state

Build graph
graph = StateGraph(State)
graph.addnode("nodea", nodea)
graph.addnode("nodeb", nodeb)


Add edges
graph.addedge(START, "nodea")

graph.addedge("nodea", "nodeb")
graph.addedge("nodeb", END)

Compile
app = graph.compile()

Run
result = app.invoke({"messages": [], "currentstep": ""})

print(result)

from typing import Annotated
from typingextensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import addmessages

from langchainopenai import ChatOpenAI
from langchaincore.messages import HumanMessage, AIMessage


State dengan message history
class ChatState(TypedDict):
    messages: Annotated[list, addmessages]

Initialize LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

Chatbot node
def chatbot(state: ChatState) -> ChatState:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

Build graph
graph = StateGraph(ChatState)
graph.addnode("chatbot", chatbot)

graph.addedge(START, "chatbot")
graph.addedge("chatbot", END)


app = graph.compile()

Interactive chat
def chat(userinput: str, history: list = None):
    if history is None:
        history = []

    history.append(HumanMessage(content=userinput))

    result = app.invoke({"messages": history})

    return result["messages"]

Usage
messages = chat("Halo, siapa kamu?")
print(messages[-1].content)

messages = chat("Apa yang bisa kamu lakukan?", messages)
print(messages[-1].content)

from typing import Annotated
from typingextensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import addmessages

from langgraph.prebuilt import ToolNode, toolscondition
from langchainopenai import ChatOpenAI

from langchaincore.tools import tool

Define tools
@tool
def searchweb(query: str) -> str:

    """Search the web for information."""
    # Simulate web search
    return f"Search results for: {query}"

@tool
def calculator(expression: str) -> str:
    """Calculate a mathematical expression."""
    try:
        result = eval(expression)
        return f"Result: {result}"
    except:
        return "Error calculating expression"

@tool
def getweather(city: str) -> str:
    """Get current weather for a city."""
    # Simulate weather API
    return f"Weather in {city}: 25°C, Sunny"

tools = [searchweb, calculator, getweather]

State
class AgentState(TypedDict):
    messages: Annotated[list, addmessages]


LLM dengan tools
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llmwithtools = llm.bindtools(tools)

Agent node
def agent(state: AgentState) -> AgentState:
    response = llmwithtools.invoke(state["messages"])
    return {"messages": [response]}

Build graph
graph = StateGraph(AgentState)

Add nodes
graph.addnode("agent", agent)

graph.addnode("tools", ToolNode(tools))

Add edges
graph.addedge(START, "agent")

graph.addconditionaledges(

    "agent",
    toolscondition,  # Routes to "tools" if tool call, else END
)
graph.addedge("tools", "agent")


app = graph.compile()

Test
from langchaincore.messages import HumanMessage

result = app.invoke({
    "messages": [HumanMessage(content="What's the weather in Jakarta?")]
})
print(result["messages"][-1].content)

result = app.invoke({
    "messages": [HumanMessage(content="Calculate 25  4 + 100")]

})
print(result["messages"][-1].content)

from typing import Annotated, Literal
from typingextensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import addmessages
from langchainopenai import ChatOpenAI

from langchaincore.tools import tool
from langchaincore.messages import HumanMessage, AIMessage, ToolMessage


Tools
@tool
def searchdatabase(query: str) -> str:
    """Search internal database for information."""
    # Simulate database search
    data = {
        "producta": "Price: $100, Stock: 50",

        "productb": "Price: $200, Stock: 25",
        "usercount": "Total users: 10,000"

    }
    for key, value in data.items():
        if key in query.lower():
            return value
    return "No results found"

@tool
def sendemail(to: str, subject: str, body: str) -> str:
    """Send an email to a recipient."""
    return f"Email sent to {to} with subject: {subject}"

tools = [searchdatabase, sendemail]

State
class ReActState(TypedDict):
    messages: Annotated[list, addmessages]

    iteration: int

LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llmwithtools = llm.bindtools(tools)

Nodes
def reasoning(state: ReActState) -> ReActState:
    """Agent reasoning step"""
    response = llmwithtools.invoke(state["messages"])
    return {
        "messages": [response],
        "iteration": state.get("iteration", 0) + 1
    }

def acting(state: ReActState) -> ReActState:
    """Execute tool calls"""
    lastmessage = state["messages"][-1]

    toolresults = []

    for toolcall in lastmessage.toolcalls:

        toolname = toolcall["name"]

        toolargs = toolcall["args"]


        # Find and execute tool
        for t in tools:
            if t.name == toolname:
                result = t.invoke(toolargs)

                toolresults.append(
                    ToolMessage(content=result, toolcallid=toolcall["id"])

                )
                break

    return {"messages": toolresults}

def shouldcontinue(state: ReActState) -> Literal["acting", "end"]:

    """Determine if we should continue or end"""
    lastmessage = state["messages"][-1]

    # Check iteration limit
    if state.get("iteration", 0) >= 5:
        return "end"

    # Check if there are tool calls
    if hasattr(lastmessage, "toolcalls") and lastmessage.toolcalls:
        return "acting"

    return "end"

Build graph
graph = StateGraph(ReActState)

graph.addnode("reasoning", reasoning)

graph.addnode("acting", acting)

graph.addedge(START, "reasoning")

graph.addconditionaledges(

    "reasoning",
    shouldcontinue,
    {"acting": "acting", "end": END}
)
graph.addedge("acting", "reasoning")


app = graph.compile()

Test
result = app.invoke({
    "messages": [HumanMessage(content="Search for producta info and then send it to user@example.com")],
    "iteration": 0
})

for msg in result["messages"]:
    print(f"{type(msg).name}: {msg.content[:100]}...")

from typing import Annotated, Literal
from typingextensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import addmessages
from langchainopenai import ChatOpenAI

from langchaincore.messages import HumanMessage, AIMessage

State
class MultiAgentState(TypedDict):
    messages: Annotated[list, addmessages]

    nextagent: str

Agents
def researcher(state: MultiAgentState) -> MultiAgentState:
    """Research agent - gathers information"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    system = """You are a research agent. Your job is to gather and analyze information.
    Be thorough but concise in your findings."""

    messages = [{"role": "system", "content": system}] + [
        {"role": m.type, "content": m.content} for m in state["messages"]
    ]

    response = llm.invoke(messages)
    return {"messages": [AIMessage(content=f"[Researcher]: {response.content}")]}

def writer(state: MultiAgentState) -> MultiAgentState:
    """Writer agent - creates content"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

    system = """You are a writer agent. Your job is to create well-written content
    based on the research provided. Be creative and engaging."""

    messages = [{"role": "system", "content": system}] + [
        {"role": m.type, "content": m.content} for m in state["messages"]
    ]

    response = llm.invoke(messages)
    return {"messages": [AIMessage(content=f"[Writer]: {response.content}")]}

def reviewer(state: MultiAgentState) -> MultiAgentState:
    """Reviewer agent - reviews and improves content"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    system = """You are a reviewer agent. Your job is to review content and provide
    constructive feedback or improvements."""

    messages = [{"role": "system", "content": system}] + [
        {"role": m.type, "content": m.content} for m in state["messages"]
    ]

    response = llm.invoke(messages)
    return {"messages": [AIMessage(content=f"[Reviewer]: {response.content}")]}

def supervisor(state: MultiAgentState) -> MultiAgentState:
    """Supervisor - decides which agent to call next"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    system = """You are a supervisor managing a team of agents:
researcher: gathers information
writer: creates content
reviewer: reviews content

    Based on the conversation, decide which agent should work next.
    Respond with ONLY one of: researcher, writer, reviewer, or FINISH"""

    messages = [{"role": "system", "content": system}] + [
        {"role": m.type, "content": m.content} for m in state["messages"]
    ]

    response = llm.invoke(messages)
    nextagent = response.content.strip().lower()


    return {"nextagent": nextagent}


def routesupervisor(state: MultiAgentState) -> Literal["researcher", "writer", "reviewer", "end"]:
    """Route based on supervisor decision"""
    nextagent = state.get("nextagent", "").lower()

    if "finish" in nextagent:

        return "end"
    elif "researcher" in nextagent:
        return "researcher"
    elif "writer" in nextagent:

        return "writer"
    elif "reviewer" in nextagent:
        return "reviewer"
    else:
        return "end"

Build graph
graph = StateGraph(MultiAgentState)

graph.addnode("supervisor", supervisor)

graph.addnode("researcher", researcher)
graph.addnode("writer", writer)

graph.addnode("reviewer", reviewer)

graph.addedge(START, "supervisor")

graph.addconditionaledges(

    "supervisor",
    routesupervisor,
    {
        "researcher": "researcher",
        "writer": "writer",
        "reviewer": "reviewer",
        "end": END
    }
)
graph.addedge("researcher", "supervisor")

graph.addedge("writer", "supervisor")
graph.addedge("reviewer", "supervisor")


app = graph.compile()

Test
result = app.invoke({
    "messages": [HumanMessage(content="Write a blog post about AI in healthcare")],
    "nextagent": ""
})

for msg in result["messages"]:
    print(f"\n{msg.content}\n{'='50}")

Human-in-the-Loop

1. Interrupt untuk Approval

from typing import Annotated
from typingextensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import addmessages
from langgraph.checkpoint.memory import MemorySaver
from langchainopenai import ChatOpenAI

from langchaincore.messages import HumanMessage

State
class ApprovalState(TypedDict):
    messages: Annotated[list, addmessages]

    draft: str
    approved: bool

Nodes
def createdraft(state: ApprovalState) -> ApprovalState:
    llm = ChatOpenAI(model="gpt-4o-mini")
    response = llm.invoke(state["messages"])
    return {"draft": response.content, "approved": False}

def processapproval(state: ApprovalState) -> ApprovalState:

    # This node processes after human approval
    return {"messages": [AIMessage(content=f"Final approved content: {state['draft']}")]}

def checkapproval(state: ApprovalState) -> str:
    if state.get("approved", False):
        return "process"
    return "wait"

Build graph
graph = StateGraph(ApprovalState)

graph.addnode("createdraft", createdraft)

graph.addnode("processapproval", processapproval)

graph.addedge(START, "createdraft")
graph.addconditionaledges(
    "createdraft",

    checkapproval,
    {"process": "processapproval", "wait": END}

)
graph.addedge("processapproval", END)


Compile dengan checkpointer untuk persistence
memory = MemorySaver()
app = graph.compile(checkpointer=memory, interruptbefore=["processapproval"])


Usage dengan human approval
config = {"configurable": {"threadid": "1"}}

Step 1: Create draft
result = app.invoke(
    {"messages": [HumanMessage(content="Write an email to client about project delay")]},
    config
)
print("Draft created:", result["draft"])

Step 2: Human reviews and approves
In real app, this would be UI interaction
print("\n[Human approves the draft]\n")

Step 3: Continue with approval
result = app.invoke(
    {"approved": True},
    config
)
print("Final:", result["messages"][-1].content)

Persistence dan Checkpointing

1. Memory Saver (In-Memory)

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = graph.compile(checkpointer=memory)

Run dengan threadid

config = {"configurable": {"threadid": "user123"}}

result = app.invoke({"messages": [HumanMessage(content="Hello")]}, config)

Continue conversation
result = app.invoke({"messages": [HumanMessage(content="What did I say?")]}, config)

from langgraph.checkpoint.sqlite import SqliteSaver

Create SQLite checkpointer
with SqliteSaver.fromconnstring(":memory:") as checkpointer:

    app = graph.compile(checkpointer=checkpointer)

    config = {"configurable": {"threadid": "session1"}}

    result = app.invoke({"messages": [HumanMessage(content="Hello")]}, config)

from langgraph.checkpoint.postgres import PostgresSaver

connectionstring = "postgresql://user:password@localhost/langgraph"

with PostgresSaver.fromconnstring(connectionstring) as checkpointer:

    app = graph.compile(checkpointer=checkpointer)

    config = {"configurable": {"threadid": "persistentsession"}}

    result = app.invoke({"messages": [HumanMessage(content="Hello")]}, config)

from langchaincore.messages import HumanMessage

Stream all events
for event in app.stream({"messages": [HumanMessage(content="Tell me a story")]}):
    print(event)
    print("---")

2. Stream dengan Mode

# Stream values (state after each node)
for state in app.stream(
    {"messages": [HumanMessage(content="Hello")]},
    streammode="values"

):
    print(state["messages"][-1].content)

Stream updates (only changes)
for update in app.stream(
    {"messages": [HumanMessage(content="Hello")]},
    streammode="updates"
):
    print(update)

3. Async Streaming

import asyncio

async def streamresponse():

    async for event in app.astream(
        {"messages": [HumanMessage(content="Hello")]}
    ):
        print(event)

asyncio.run(streamresponse())

RAG dengan LangGraph

from typing import Annotated, List
from typingextensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langchainopenai import ChatOpenAI, OpenAIEmbeddings
from langchaincommunity.vectorstores import FAISS

from langchaincore.messages import HumanMessage

State
class RAGState(TypedDict):
    question: str
    context: List[str]
    answer: str

Setup vector store
embeddings = OpenAIEmbeddings()
documents = [
    "LangGraph is a library for building stateful AI applications.",
    "It supports cyclic flows and human-in-the-loop interactions.",
    "LangGraph extends LangChain with graph-based workflows.",
]
vectorstore = FAISS.fromtexts(documents, embeddings)

retriever = vectorstore.asretriever(k=2)

Nodes
def retrieve(state: RAGState) -> RAGState:
    """Retrieve relevant documents"""
    docs = retriever.invoke(state["question"])
    return {"context": [doc.pagecontent for doc in docs]}


def generate(state: RAGState) -> RAGState:
    """Generate answer based on context"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    context = "\n".join(state["context"])
    prompt = f"""Answer the question based on the context.

Context:
{context}

Question: {state["question"]}

Answer:"""

    response = llm.invoke(prompt)
    return {"answer": response.content}

Build graph
graph = StateGraph(RAGState)

graph.addnode("retrieve", retrieve)
graph.addnode("generate", generate)


graph.addedge(START, "retrieve")
graph.addedge("retrieve", "generate")

graph.addedge("generate", END)

app = graph.compile()

Test
result = app.invoke({"question": "What is LangGraph?"})
print(f"Answer: {result['answer']}")

Best Practices

1. Error Handling

from langgraph.graph import StateGraph

def safenode(state):

    try:
        # Your logic here
        result = riskyoperation()
        return {"result": result, "error": None}
    except Exception as e:
        return {"result": None, "error": str(e)}

def errorrouter(state):

    if state.get("error"):
        return "handleerror"
    return "continue"

2. Timeout dan Retry

import asyncio
from tenacity import retry, stopafterattempt, waitexponential


@retry(stop=stopafterattempt(3), wait=waitexponential(multiplier=1, min=4, max=10))
def nodewithretry(state):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

3. Logging

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(name)

def logged_node(state):
    logger.info(f"Processing state: {state}")
    result = process(state)
    logger.info(f"Result: {result}")
    return result

Kesimpulan

LangGraph menyediakan framework yang powerful untuk membangun AI agents yang kompleks dengan:

Key takeaways:

• Gunakan TypedDict untuk type-safe state
• Leverage conditional edges untuk dynamic routing
• Implement checkpointing untuk production
• Use streaming untuk real-time responses
• Handle errors gracefully dengan fallbacks