Composing Complex LLM Workflows: Advanced LangChain Patterns
When chains and agents aren't enough—when you need conditional branching, iterative loops, human feedback integration, or coordinated multi-agent reasoning—use LangGraph. LangGraph is LangChain's orchestration layer for stateful workflows, enabling complex patterns like ReAct loops, plan-and-execute, and hierarchical agent teams. It's the natural evolution from simple chains to production-grade LLM systems.
I started with chains, outgrew them, hacked together orchestration logic, then discovered LangGraph solved all the problems with a cleaner abstraction. Suddenly, complex workflows were readable and testable.
Introduction to LangGraph
LangGraph builds state machines where nodes are LLM calls and edges are transitions. State persists across steps, enabling loops and branching:
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
# Define state schema
class State(TypedDict):
input: str
reasoning: str
action: str
observation: str
# Create graph
graph = StateGraph(State)
# Define nodes (functions that process state)
def think_node(state: State) -> State:
# Generate reasoning
state["reasoning"] = "Let me think about this..."
return state
def act_node(state: State) -> State:
# Decide on action
state["action"] = "Search for information"
return state
def observe_node(state: State) -> State:
# Observe result
state["observation"] = "Found relevant documents"
return state
# Add nodes to graph
graph.add_node("think", think_node)
graph.add_node("act", act_node)
graph.add_node("observe", observe_node)
# Add edges (transitions)
graph.add_edge("think", "act")
graph.add_edge("act", "observe")
graph.add_edge("observe", END)
# Set entry point
graph.set_entry_point("think")
# Compile and run
runnable = graph.compile()
result = runnable.invoke({"input": "What is async/await?"})
This defines a linear workflow: think → act → observe. State flows through each node, accumulating information.
Conditional Branching with Router Nodes
Add logic to decide which node to go to next:
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
class State(TypedDict):
question: str
needs_retrieval: str
answer: str
def classify_node(state: State) -> State:
"""Decide if we need to retrieve documents."""
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_template(
"Is this a factual question requiring retrieval? Answer YES or NO: {question}"
)
response = (prompt | model | StrOutputParser()).invoke(
{"question": state["question"]}
)
state["needs_retrieval"] = "YES" if "YES" in response.upper() else "NO"
return state
def retrieve_and_answer(state: State) -> State:
"""Retrieve documents and answer."""
# Retrieve + generate
state["answer"] = "Answer based on retrieved documents..."
return state
def direct_answer(state: State) -> State:
"""Answer without retrieval."""
state["answer"] = "Answer from model knowledge..."
return state
def route_retrieval(state: State) -> str:
"""Router function: decide next node."""
if state["needs_retrieval"] == "YES":
return "retrieve"
else:
return "direct"
# Build graph
graph = StateGraph(State)
graph.add_node("classify", classify_node)
graph.add_node("retrieve", retrieve_and_answer)
graph.add_node("direct", direct_answer)
graph.add_edge("classify", "route")
graph.add_conditional_edges(
"classify",
route_retrieval,
{"retrieve": "retrieve", "direct": "direct"}
)
graph.add_edge("retrieve", END)
graph.add_edge("direct", END)
graph.set_entry_point("classify")
runnable = graph.compile()
result = runnable.invoke({"question": "What is Python?"})
The route_retrieval function returns a string determining which node to visit next. This enables dynamic workflows.
Looping and Iteration with LangGraph
Implement ReAct-style loops:
from langgraph.graph import StateGraph
class State(TypedDict):
question: str
thoughts: list[str]
actions: list[str]
observations: list[str]
final_answer: str
iterations: int
def think_step(state: State) -> State:
"""Generate thought."""
state["thoughts"].append("I need to search for information...")
return state
def act_step(state: State) -> State:
"""Execute action."""
state["actions"].append("Search the web")
state["observations"].append("Found relevant results")
return state
def should_continue(state: State) -> str:
"""Decide to continue or finish."""
state["iterations"] += 1
if state["iterations"] >= 3 or "final answer" in state["observations"][-1].lower():
return "finish"
else:
return "continue"
def finish(state: State) -> State:
"""Generate final answer."""
state["final_answer"] = "Based on my reasoning..."
return state
graph = StateGraph(State)
graph.add_node("think", think_step)
graph.add_node("act", act_step)
graph.add_node("finish", finish)
graph.add_edge("think", "act")
graph.add_conditional_edges(
"act",
should_continue,
{"continue": "think", "finish": "finish"}
)
graph.add_edge("finish", END)
graph.set_entry_point("think")
runnable = graph.compile()
result = runnable.invoke({
"question": "How do async/await work?",
"thoughts": [],
"actions": [],
"observations": [],
"final_answer": "",
"iterations": 0
})
This loop iterates (think → act → check) until it reaches a stopping condition.
Human-in-the-Loop: Pausing for Input
Pause execution for human review or correction:
from langgraph.graph import StateGraph, END, Interrupt
class State(TypedDict):
draft_response: str
human_feedback: str
final_response: str
def generate_draft(state: State) -> State:
"""Generate a draft response."""
state["draft_response"] = "This is my draft answer..."
return state
def review_with_human(state: State) -> State:
"""Pause and wait for human feedback."""
# In production, this integrates with your API/UI
raise Interrupt("Please review the draft and provide feedback.")
def revise(state: State) -> State:
"""Revise based on feedback."""
state["final_response"] = f"Revised answer incorporating: {state['human_feedback']}"
return state
graph = StateGraph(State)
graph.add_node("generate", generate_draft)
graph.add_node("review", review_with_human)
graph.add_node("revise", revise)
graph.add_edge("generate", "review")
graph.add_edge("review", "revise")
graph.add_edge("revise", END)
graph.set_entry_point("generate")
runnable = graph.compile()
# Execution
try:
result = runnable.invoke({"draft_response": "", "human_feedback": "", "final_response": ""})
except Interrupt as interrupt:
# Handle the pause—get feedback, then resume
print(f"Paused: {interrupt}")
# Resume with: runnable.invoke(..., checkpoint_id=..., feedback=...)
Interrupts pause the graph, return control to the user/API, and resume when feedback is provided.
Multi-Agent Patterns: Hierarchical Teams
Coordinate multiple agents toward a common goal:
from langgraph.graph import StateGraph
class State(TypedDict):
task: str
planner_output: str
executor_output: str
reviewer_output: str
final_output: str
model = ChatOpenAI(model="gpt-4o-mini")
def planner_node(state: State) -> State:
"""Plan the approach."""
prompt = ChatPromptTemplate.from_template(
"Create a plan to solve: {task}"
)
state["planner_output"] = (prompt | model | StrOutputParser()).invoke(
{"task": state["task"]}
)
return state
def executor_node(state: State) -> State:
"""Execute the plan."""
prompt = ChatPromptTemplate.from_template(
"Execute this plan: {plan}"
)
state["executor_output"] = (prompt | model | StrOutputParser()).invoke(
{"plan": state["planner_output"]}
)
return state
def reviewer_node(state: State) -> State:
"""Review and improve."""
prompt = ChatPromptTemplate.from_template(
"Review this output and suggest improvements: {output}"
)
state["reviewer_output"] = (prompt | model | StrOutputParser()).invoke(
{"output": state["executor_output"]}
)
return state
def finalize_node(state: State) -> State:
"""Combine outputs into final response."""
state["final_output"] = f"""
Plan: {state["planner_output"]}
Execution: {state["executor_output"]}
Review: {state["reviewer_output"]}
"""
return state
graph = StateGraph(State)
graph.add_node("plan", planner_node)
graph.add_node("execute", executor_node)
graph.add_node("review", reviewer_node)
graph.add_node("finalize", finalize_node)
graph.add_edge("plan", "execute")
graph.add_edge("execute", "review")
graph.add_edge("review", "finalize")
graph.add_edge("finalize", END)
graph.set_entry_point("plan")
runnable = graph.compile()
result = runnable.invoke({
"task": "Design a caching strategy for a web app",
"planner_output": "",
"executor_output": "",
"reviewer_output": "",
"final_output": ""
})
Three agents (planner, executor, reviewer) collaborate on the same task.
Pattern Comparison: Chains vs. Agents vs. LangGraph
| Pattern | Use Case | State | Loops | Branching |
|---|---|---|---|---|
| Chain | Sequential steps | No | No | No |
| Agent | Autonomous tool-calling | Implicit | Yes | Implicit (routing) |
| LangGraph | Complex workflows | Explicit | Yes | Explicit |
Choose chains for simple sequences, agents for tool-calling, LangGraph for everything else.
Key Takeaways
- LangGraph builds state machines where nodes are LLM calls and edges are transitions
- State persists across steps, enabling complex stateful workflows
- Conditional edges route execution based on node output
- Loops implement ReAct and iterative reasoning patterns
- Interrupts pause execution for human feedback
- Multi-agent patterns coordinate teams toward shared goals
- LangGraph is the natural evolution when chains and agents reach their limits
Frequently Asked Questions
When should I use LangGraph instead of chains?
When you need loops, conditional branching, or explicit state management. Chains are simpler; use them for linear workflows. Switch to LangGraph when logic becomes complex.
Can I convert an agent to LangGraph?
Yes. The underlying logic is the same—the difference is explicit state management. LangGraph gives you more control and visibility.
How do I debug LangGraph workflows?
Use runnable.get_graph().visualize() to see the graph structure. Inspect state at each step. Set breakpoints in nodes or use langchain.debug = True.
Can LangGraph integrate with external systems?
Yes. Nodes can call APIs, databases, or external tools. State is just a dict, so it carries any data through the workflow.
What's the maximum complexity LangGraph can handle?
LangGraph scales to dozens of nodes and complex branching. For very complex systems (100+ nodes), consider modular subgraphs or external orchestration (Airflow, Prefect).