Learn LangChain - Agents & Basic LangGraph

#Introduction

In episode 12 you polished your RAG pipeline with multi-query, hybrid retrieval, and reranking — producing far higher quality context. But there's a limit that static chains can't overcome: a chain knows the exact order of its steps, yet questions don't always need the same order. Episode 13 answers this with agents — programs that decide their own next step based on state — then composes them explicitly with LangGraph. The flow of this episode: the agent loop concept, building an agent with create_agent, observing it with astream_events, breaking down StateGraph (nodes, edges, conditional edges), and closing with interrupt/resume for human-in-the-loop.

#From Chain to Agent: Adding the Ability to Act

The chains we've built so far are deterministic: prompt in, model calls, parser converts output. Agents add one new engine called the agent loop:

1. The model receives the message history and the list of available tools.
2. The model decides: answer directly, or call one or more tools.
3. If it calls a tool, the tool executes, and the result is sent back to the model as a new message.
4. Steps 2 and 3 repeat until the model gives a final answer or hits the limit (recursion limit).

The conceptual key: the model has control over the path — you don't dictate "always retrieve first". This agent is still one model, just now equipped with a think-call-rethink capability. The tool definitions are the same as in episode 8 with the @tool decorator:

Tool definitions for an agent

from langchain_core.tools import tool
 
@tool
def cek_cuaca(kota: str) -> str:
    "Mengembalikan suhu kota yang diminta."
    return f"28 derajat Celsius di {kota}"
 
@tool
def cari_berita(kueri: str) -> str:
    "Mencari berita terbaru berdasarkan kueri."
    return "Artikel demo: inflasi turun di kuartal kedua"

#Building an Agent with create_agent

Rather than hand-assembling the loop, LangChain v1 provides create_agent — a prebuilt constructor that takes a model, a list of tools, and a prompt, then returns an object ready to be invoked:

Creating a tool-calling agent

from langchain.agents import create_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Kamu asisten yang ringkas. Gunakan tool hanya bila perlu."),
    ("placeholder", "{messages}"),
])
 
agent = create_agent(
    model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    tools=[cek_cuaca, cari_berita],
    prompt=prompt,
)
result = agent.invoke({"messages": [("user", "Cuaca di Jakarta?")]})
print(result["messages"][-1].content)

Note the input: a dict with the messages key containing a list of messages — exactly like the ChatPromptTemplate from episode 4. The output is also messages, and the last message holds the final answer. Because create_agent returns an object of type Runnable, all the LCEL methods from episode 5 still apply: invoke, ainvoke, stream, even batch.

#Observing the Agent Loop with astream_events

While an agent is thinking, choosing tools, and processing results, you won't know what's happening if you only use invoke. In production (and during debugging), use astream_events — an event stream that reveals every stage: from tool execution to answer tokens:

Streaming agent events

async for event in agent.astream_events(
    {"messages": [("user", "Cuaca di Jakarta dan berita ekonomi?")]},
    version="v1",
):
    kind = event["event"]
    if kind == "on_tool_start":
        print("TOOL:", event["name"], event["data"].get("input"))
    elif kind == "on_chat_model_stream":
        chunk = event["data"]["chunk"]
        if chunk.content:
            print(chunk.content, end="")

The filters on event above separate two different things: on_tool_start tells you which tool was called and with what input, while on_chat_model_stream gives the answer tokens in real time. This is the same streaming pattern as episode 5, just applied to an agent — and the combination of the two is the raw material for a "watch the agent think" UI.

#Basic StateGraph: Nodes, Edges, START, and END

The agent loop "rolled up" by create_agent can actually be rewritten explicitly as a graph: a set of nodes (functions) connected by edges. In LangGraph, that graph is called StateGraph, and its state is described with a TypedDict. A minimal example with two nodes:

Minimal StateGraph with two nodes

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
 
class AgentState(TypedDict):
    messages: list
 
def node_a(state: AgentState):
    return {"messages": [("ai", "dari node A")]}
 
def node_b(state: AgentState):
    return {"messages": [("ai", "dari node B")]}
 
graph = StateGraph(AgentState)
graph.add_node("a", node_a)
graph.add_node("b", node_b)
graph.add_edge(START, "a")
graph.add_edge("a", "b")
graph.add_edge("b", END)
 
app = graph.compile()
result = app.invoke({"messages": []})

Each node is a function that receives the state and returns a partial update — LangGraph merges it into the main state. START and END are special nodes for entering and exiting the graph. compile() turns the graph into something executable, and the invoke result contains the complete state after all nodes finish. This is the core that distinguishes LangGraph from a chain: explicit state flowing between nodes, not a silent pipe.

#Conditional Edges: Choosing a Path Based on State

A static graph is already useful, but LangGraph's power is in conditional edges — edges that choose a destination node based on the state contents. This is what makes dynamic routing like "if the search returns nothing, generate directly; if not, summarize" possible:

Conditional edge with a router function

def router(state: AgentState):
    if not state.get("documents"):
        return "generate"
    return "summarize"
 
graph = StateGraph(AgentState)
graph.add_node("retrieve", node_retrieve)
graph.add_node("summarize", node_summarize)
graph.add_node("generate", node_generate)
graph.add_edge(START, "retrieve")
graph.add_conditional_edges(
    "retrieve",
    router,
    {"generate": "generate", "summarize": "summarize"},
)
graph.add_edge("summarize", "generate")
graph.add_edge("generate", END)

add_conditional_edges takes three arguments: the source node, a function that returns the destination name, and a mapping of names to nodes. The router function reads the state and returns a destination key — LangGraph then follows the mapping to find the right node. With this pattern, you can replicate the agent loop logic exactly: a tools node executes tool calls, then a conditional edge decides whether to return to the model or go to the end node.

#Interrupt & Resume: Human-in-the-Loop

Sometimes an agent must not run unsupervised: payment confirmation, running destructive commands, or approving sensitive tools. LangGraph provides interrupt — a node that pauses execution midway and waits for a human decision. The requirement: the graph must have a checkpointer so the state is saved while paused.

Interrupt for human approval

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
 
def execute_sensitive(state: AgentState):
    action = interrupt({"action": "hapus-file", "target": "/tmp/demo"})
    return {"messages": [("ai", f"aksi disetujui: {action}")]}
 
graph.add_node("execute", execute_sensitive)
graph.add_edge("execute", END)
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)
 
config = {"configurable": {"thread_id": "thread-1"}}
app.invoke({"messages": []}, config=config)

When execution reaches interrupt, it stops and its status is saved under the thread_id. To continue, send a Command with a resume value:

Resuming after an interrupt

app.invoke(Command(resume="setuju"), config=config)

The value "setuju" above becomes the return value of interrupt inside the execute node. The thread_id links all calls within the same conversation — a concept you already know from the checkpoints in episode 7, now used for genuine human-in-the-loop.

#Conclusion

This episode marked the conceptual leap from static pipelines to systems that determine their own path: understanding the agent loop as a cycle of tool calls and feedback, building an agent with create_agent, monitoring its process with astream_events, composing logic explicitly with StateGraph and conditional edges, and pausing it with interrupt/Command for human-in-the-loop. The agent is now no longer just a "chain with tools", but a stateful graph that can be steered, paused, and resumed.

Key takeaways:

• Agent loop = model decides, tool executes, result sent back, repeating until a final answer.
• create_agent combines prompt, model, and tools into a single Runnable object supporting all LCEL methods.
• astream_events exposes every agent stage: tool calls and answer tokens.
• StateGraph composes nodes and edges; conditional edges make routing depend on state.
• Interrupt pauses the graph at a specific point and resumes with Command; a checkpointer is required.

Episode 13 built the foundation of a single-agent graph. In episode 14 we level it up: Deep Agents & Multi-Agent Patterns — automatically spawned subagents, async subagents, multi-modal tools for PDF/audio/video, and state backends like StateBackend and StoreBackend. See you there!