LangGraph: Core Concepts

[Farhan Khan]

LangGraph is a framework for building stateful, reliable agent workflows. Instead of executing a single prompt-response cycle, LangGraph enables agents to operate as directed graphs, where each node represents a step, state is shared across the workflow, and execution can branch, pause, or persist.

This article summarizes the core concepts, prebuilt utilities, and typical applications of LangGraph.

State​

• The memory object that flows through the graph.
• Nodes read from and write updates to state.
• 
  Managed through channels that define merge strategies:
  
  • Replace → overwrite existing values.
  • Append → add items (e.g., chat history).
  • Merge → combine dicts/lists.
• 
  Acts as the single source of truth for the agent’s execution.
  

Example: How State Evolves​

Consider a simple graph with three nodes: plan → search → answer.

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

class State(TypedDict, total=False):
    question: str
    plan: str
    results: list[str]
    final_answer: str

def plan_node(state: State) -> State:
    return {"plan": f"Search for: {state['question']}"}

def search_node(state: State) -> State:
    return {"results": [f"Result about {state['plan']}"]}

def answer_node(state: State) -> State:
    return {"final_answer": f"Based on {state['results'][0]}, here is the answer."}

Initial Input

{"question": "What is LangGraph?"}

After plan node

{"question": "What is LangGraph?", "plan": "Search for: What is LangGraph?"}

After search node

{"question": "What is LangGraph?", "plan": "Search for: What is LangGraph?", "results": ["Result about Search for: What is LangGraph?"]}

After answer node

{"question": "What is LangGraph?", "plan": "Search for: What is LangGraph?", "results": ["Result about Search for: What is LangGraph?"], "final_answer": "Based on Result about Search for: What is LangGraph?, here is the answer."}

Nodes​

• 
  Nodes are the functional units of a LangGraph.
  
  
• 
  Each node is a Python function that:
  
  • Takes the current state (a dictionary-like object).
  • Returns updates to that state (usually as another dictionary).
• 
  A node can perform many different actions depending on the workflow:
  
  • Invoke an LLM to generate text, plans, or summaries.
  • Call external tools or APIs (e.g., search engine, database, calculator).
  • Execute deterministic logic (e.g., scoring, validation, formatting).
• 
  Nodes don’t overwrite the whole state by default — instead, they return partial updates that LangGraph merges into the global state using channels.
  
  
• 
  A node is not an “agent” by itself. The entire graph of nodes forms the agent.
  

Example: Simple Node​

from typing import TypedDict

class State(TypedDict, total=False):
    question: str
    plan: str

def plan_node(state: State) -> State:
    q = state["question"]
    return {"plan": f"Search online for: {q}"}

Input State

{"question": "What is LangGraph?"}

Output Update

{"plan": "Search online for: What is LangGraph?"}

After merging

{"question": "What is LangGraph?", "plan": "Search online for: What is LangGraph?"}

Example: Node Invoking an LLM​

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

def answer_node(state: State) -> State:
    response = llm.invoke(state["question"])
    return {"final_answer": response.content}

Nodes are modular steps. They can be simple (string formatting) or advanced (LLM + tools). Together, they form the workflow.

Edges​

• 
  Edges define the flow of execution between nodes.
  
  
• 
  Normal edges → fixed transitions.
  
  
• 
  Conditional edges → branching logic, using a router function.
  
  
• 
  Special markers:
  
  • START → entry point.
  • END → exit point.

Example: Normal Edges​

from langgraph.graph import StateGraph, START, END

graph = StateGraph(State)
graph.add_node("plan", plan_node)
graph.add_node("search", search_node)
graph.add_node("answer", answer_node)

graph.add_edge(START, "plan")
graph.add_edge("plan", "search")
graph.add_edge("search", "answer")
graph.add_edge("answer", END)

Example: Conditional Edges with Router​

def router(state: State) -> str:
    q = state["question"]
    if "latest" in q.lower():
        return "search"
    else:
        return "answer"

graph.add_conditional_edges(
    "plan", router, {"search": "search", "answer": "answer"}
)

• Input: "What is the capital of France?" → routed to answer.
• Input: "What are the latest news on AAPL?" → routed to search.

Streaming​

Streaming provides live feedback during execution.

Update Stream (node-level)​

for event in app.stream(inputs, config=config, stream_mode="updates"):
    print(event)

Output

{'plan': {'plan': 'Search for: What is LangGraph?'}}
{'search': {'results': ['Result about What is LangGraph?']}}
{'answer': {'final_answer': '...final text...'}}

Token Stream (LLM output)​

for chunk in app.stream(inputs, config=config, stream_mode="messages"):
    print(chunk, end="", flush=True)

• stream_mode="updates" → node updates.
• stream_mode="messages" → token stream (if LLM supports it).

Memory​

Memory in LangGraph = state + checkpointers.

Short-Term Memory (Within a Thread)​

from typing import TypedDict, List

class State(TypedDict, total=False):
    question: str
    chat_history: List[str]
    answer: str

def add_to_history(state: State) -> State:
    history = state.get("chat_history", [])
    history.append(state["question"])
    return {"chat_history": history}

Example after two turns:

{"chat_history": ["What is LangGraph?", "Explain checkpointers"], "answer": "..."}

Long-Term Memory (Across Runs)​

from langgraph.checkpoint.memory import MemorySaver

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

app.invoke({"question": "What is LangGraph?"}, config={"configurable": {"thread_id": "t1"}})
app.invoke({"question": "And what are edges?"}, config={"configurable": {"thread_id": "t1"}})

Both runs share the same thread_id → context is preserved.

Combined​

• Short-term memory = within a run.
• Long-term memory = across runs.
• Together → continuity + reliability.

Unlike LangChain, LangGraph doesn’t treat memory as a separate object — it’s baked into state + checkpointers.

Continue reading...

 

Join 𝕋𝕄𝕋 on Telegram
Channel PREVIEW: