LangGraph: Core Concepts

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.

1️⃣ 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."}

---

2️⃣ 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.

---

3️⃣ 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.

---

4️⃣ 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).

---

5️⃣ 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.