This content originally appeared on DEV Community and was authored by 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.
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).
- Takes the current
-
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.
This content originally appeared on DEV Community and was authored by FARHAN KHAN