INTRODUCTION: UNDERSTANDING LANGGRAPH AND ITS PURPOSE
When you begin working with Large Language Models, you quickly discover that single LLM calls are insufficient for complex tasks. Real applications require multiple coordinated steps, decision-making capabilities, state management across interactions, and often multiple specialized agents working together toward a common goal.
LangGraph addresses these challenges by providing a framework for building stateful, multi-agent applications with LLMs. The core insight is that complex LLM workflows can be elegantly modeled as directed graphs, where nodes represent operations or agents, and edges define the flow of information and control between them.
Consider a research assistant application. Such an assistant needs to search for information, analyze findings, synthesize results, and potentially iterate based on what it discovers. Each of these steps might involve different LLM calls with different prompts, external tool usage, and decision points about what to do next. LangGraph provides the structure to orchestrate all these components in a clear, maintainable way.
The graph-based approach offers several advantages. First, it makes your application's logic explicit and visual. You can literally draw out how your application works. Second, it provides fine-grained control over execution flow, allowing you to implement sophisticated conditional logic. Third, it manages state automatically, ensuring that information flows correctly between different parts of your application.
CORE CONCEPTS: THE FOUNDATION OF LANGGRAPH
Before writing any code, we need to understand four fundamental concepts that form the foundation of every LangGraph application. These concepts work together to create powerful, flexible LLM workflows.
The first concept is State. In LangGraph, state represents the shared information that flows through your entire application. Think of it as a living document that every part of your application can read from and write to. As your application executes, moving from one node to another, the state accumulates information, building up context and maintaining the history of what has happened.
State is crucial because it allows different parts of your application to build upon each other's work. When one agent performs a web search, it stores the results in the state. When another agent needs to analyze those results, it can access them from the state. This shared memory is what enables sophisticated multi-step reasoning.
The second concept is the Graph itself. A graph in LangGraph is the overall structure of your application. It defines all the possible paths your application can take and all the operations it can perform. The graph is composed of nodes connected by edges, forming a directed flow of execution.
The third concept is Nodes. Each node in your graph represents a discrete unit of work. A node is implemented as a Python function that receives the current state, performs some operation, and returns updates to the state. The operation could be calling an LLM, invoking an external API, performing calculations, making decisions, or any other computational task your application requires.
The fourth concept is Edges. Edges define how execution flows from one node to another. LangGraph supports two types of edges. Normal edges create unconditional connections, meaning after node A completes, node B always executes next. Conditional edges enable decision-making, allowing your application to choose different paths based on the current state.
ENVIRONMENT SETUP: PREPARING YOUR DEVELOPMENT ENVIRONMENT
To begin working with LangGraph, you need to set up your Python environment with the necessary dependencies. LangGraph requires Python version 3.9 or higher. You will also need to install LangChain, as LangGraph builds upon its foundation.
Open your terminal and execute the following command to install the required packages:
pip install langgraph langchain langchain-openai
For this tutorial, we will use OpenAI's models, so you need an OpenAI API key. You can obtain one from the OpenAI platform website. Once you have your key, set it as an environment variable. On Linux or macOS, use this command:
export OPENAI_API_KEY='your-actual-api-key-here'
On Windows, use this command instead:
set OPENAI_API_KEY=your-actual-api-key-here
Now let's verify that everything is installed correctly with a simple test:
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
print("LangGraph is successfully installed and ready to use!")
This code imports the essential components we will use throughout this tutorial. The StateGraph class is the primary tool for constructing graphs. The END constant is a special marker indicating workflow completion. The TypedDict and Annotated types from Python's typing module help us define type-safe state structures.
DEFINING STATE: THE INFORMATION BACKBONE
State definition is the first step in building any LangGraph application. The state structure determines what information your application tracks and how that information is updated as the application executes.
LangGraph uses Python's TypedDict to define state schemas. This provides type safety and makes your code more maintainable. Let's start with a simple example:
from typing import TypedDict, Annotated
import operator
class SimpleState(TypedDict):
counter: int
message: str
This SimpleState definition creates a state structure with two fields. The counter field stores an integer value, and the message field stores a string. When a node updates these fields, it simply replaces the old value with the new value.
However, LangGraph offers a more sophisticated mechanism for state updates through the Annotated type. This allows you to specify how updates should be applied. The most common pattern uses operator.add to accumulate values rather than replace them:
from typing import TypedDict, Annotated, Sequence
import operator
class ConversationState(TypedDict):
messages: Annotated[Sequence[str], operator.add]
step_count: int
current_topic: str
In this ConversationState definition, the messages field uses Annotated with operator.add. This tells LangGraph that when a node returns new messages, they should be appended to the existing messages list rather than replacing it. This is essential for maintaining conversation history.
The step_count and current_topic fields do not use Annotated, so they follow the default behavior of replacement. When a node returns a new value for step_count, it overwrites the previous value.
Let's see a more complex state definition that might be used for a research assistant:
from typing import TypedDict, Annotated, Sequence, Optional
import operator
class ResearchState(TypedDict):
# Accumulate messages throughout the conversation
messages: Annotated[Sequence[str], operator.add]
# Store search queries that have been executed
search_queries: Annotated[list[str], operator.add]
# Store search results from external sources
search_results: Annotated[list[dict], operator.add]
# Current research question being investigated
current_question: str
# Final synthesized answer
final_answer: Optional[str]
# Number of research iterations performed
iteration_count: int
This ResearchState demonstrates a realistic state structure for a multi-step research application. The messages, search_queries, and search_results fields all use operator.add to accumulate information over time. The current_question, final_answer, and iteration_count fields use replacement semantics.
Understanding how state updates work is critical. When a node function returns a dictionary, LangGraph merges that dictionary into the current state. For fields annotated with operator.add, the new values are added to existing values. For other fields, the new values replace the old values.
CREATING NODES: THE WORKHORSES OF YOUR APPLICATION
Nodes are where the actual work happens in your LangGraph application. Each node is a Python function that takes the current state as input and returns a dictionary representing updates to that state.
Let's create a simple node that increments a counter:
def increment_counter_node(state: SimpleState) -> dict:
"""
This node increments the counter in the state by one.
It demonstrates the basic pattern of reading from state
and returning an update.
"""
current_count = state.get("counter", 0)
new_count = current_count + 1
print(f"Incrementing counter from {current_count} to {new_count}")
# Return a dictionary with the updates to apply to state
return {"counter": new_count}
This increment_counter_node function demonstrates the fundamental node pattern. It receives the state, extracts the current counter value, increments it, and returns a dictionary containing the new counter value. LangGraph automatically merges this update into the state.
Now let's create a more sophisticated node that calls an LLM:
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, AIMessage
def llm_response_node(state: ConversationState) -> dict:
"""
This node calls an LLM with the current conversation history
and returns the LLM's response, which gets added to the messages.
"""
# Initialize the language model
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
# Get the current messages from state
current_messages = state.get("messages", [])
# Convert string messages to LangChain message objects
formatted_messages = []
for i, msg in enumerate(current_messages):
if i % 2 == 0:
formatted_messages.append(HumanMessage(content=msg))
else:
formatted_messages.append(AIMessage(content=msg))
# Call the LLM
response = llm.invoke(formatted_messages)
print(f"LLM responded: {response.content[:100]}...")
# Return the new message to be added to the conversation
# Because messages uses operator.add, this will be appended
return {
"messages": [response.content],
"step_count": state.get("step_count", 0) + 1
}
This llm_response_node demonstrates a more complex operation. It retrieves the conversation history from state, formats it appropriately for the LLM, invokes the LLM, and returns both the new message and an updated step count. Notice how the function returns a dictionary with updates for multiple state fields.
Let's create another node that performs a simulated web search:
def web_search_node(state: ResearchState) -> dict:
"""
This node simulates performing a web search based on
the current research question and stores the results.
"""
question = state.get("current_question", "")
print(f"Performing web search for: {question}")
# In a real application, this would call an actual search API
# For demonstration, we'll create simulated results
simulated_results = [
{
"title": f"Result 1 for {question}",
"snippet": "This is a simulated search result snippet...",
"url": "https://example.com/result1"
},
{
"title": f"Result 2 for {question}",
"snippet": "Another simulated search result snippet...",
"url": "https://example.com/result2"
}
]
# Return updates to state
# search_queries and search_results use operator.add, so these append
return {
"search_queries": [question],
"search_results": simulated_results
}
This web_search_node shows how you might integrate external tools into your LangGraph application. The node reads the current question from state, performs an operation (in this case simulated, but in reality would call an actual search API), and returns the results to be accumulated in the state.
BUILDING YOUR FIRST GRAPH: PUTTING IT ALL TOGETHER
Now that we understand state and nodes, let's build our first complete LangGraph application. We'll create a simple conversation system that takes user input, processes it through an LLM, and returns a response.
First, let's define our state:
from typing import TypedDict, Annotated, Sequence
import operator
class ChatState(TypedDict):
messages: Annotated[Sequence[str], operator.add]
conversation_active: bool
Next, let's create the nodes we'll need:
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage
def chat_node(state: ChatState) -> dict:
"""
This node processes the conversation through an LLM.
"""
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
messages = state.get("messages", [])
# Create a system message to set context
system_msg = SystemMessage(
content="You are a helpful assistant. Provide clear, concise answers."
)
# Format the conversation history
formatted_messages = [system_msg]
for i, msg in enumerate(messages):
formatted_messages.append(HumanMessage(content=msg))
# Get LLM response
response = llm.invoke(formatted_messages)
print(f"Assistant: {response.content}")
return {
"messages": [response.content]
}
Let's build the graph itself:
from langgraph.graph import StateGraph, END
def create_simple_chat_graph():
"""
Creates a simple chat graph with one LLM node.
"""
# Initialize the graph with our state type
workflow = StateGraph(ChatState)
# Add the chat node to the graph
# First argument is the node name, second is the function
workflow.add_node("chat", chat_node)
# Set the entry point - where execution begins
workflow.set_entry_point("chat")
# Add an edge from chat node to END
# This means after the chat node executes, the workflow terminates
workflow.add_edge("chat", END)
# Compile the graph into an executable application
app = workflow.compile()
return app
This create_simple_chat_graph function demonstrates the basic pattern for building graphs. We create a StateGraph instance, add our nodes, define the entry point, connect nodes with edges, and compile the graph into an executable application.
Let's use this graph:
# Create the graph application
app = create_simple_chat_graph()
# Prepare initial state with a user message
initial_state = {
"messages": ["What is LangGraph and why is it useful?"],
"conversation_active": True
}
# Execute the graph
result = app.invoke(initial_state)
# The result contains the final state after execution
print("\nFinal conversation:")
for i, msg in enumerate(result["messages"]):
role = "User" if i % 2 == 0 else "Assistant"
print(f"{role}: {msg}")
When you run this code, the graph executes the chat node, which processes the user's question through the LLM and returns a response. The final state contains both the original user message and the assistant's response.
CONDITIONAL EDGES: MAKING INTELLIGENT DECISIONS
The real power of LangGraph emerges when you add conditional logic to your graphs. Conditional edges allow your application to make decisions about which node to execute next based on the current state.
To implement conditional edges, you create a router function that examines the state and returns the name of the next node to execute. Let's build an example that demonstrates this:
from typing import TypedDict, Annotated, Sequence, Literal
import operator
class TaskState(TypedDict):
messages: Annotated[Sequence[str], operator.add]
task_type: str
task_complete: bool
result: str
Next let's create a router function:
def route_based_on_task(state: TaskState) -> Literal["math_task", "text_task", "end"]:
"""
This router function examines the state and decides which node
should execute next based on the task type and completion status.
"""
# If task is complete, end the workflow
if state.get("task_complete", False):
return "end"
# Otherwise, route based on task type
task_type = state.get("task_type", "")
if "math" in task_type.lower() or "calculate" in task_type.lower():
return "math_task"
else:
return "text_task"
This router function demonstrates decision-making logic. It checks if the task is complete, and if so, returns "end" to terminate the workflow. Otherwise, it examines the task type and routes to either a math-specialized node or a text-specialized node.
Let's create the specialized nodes:
def math_task_node(state: TaskState) -> dict:
"""
This node handles mathematical tasks.
"""
print("Processing mathematical task...")
messages = state.get("messages", [])
last_message = messages[-1] if messages else ""
# In a real application, this would use an LLM or calculation engine
result = f"Mathematical analysis of: {last_message}"
return {
"messages": [result],
"task_complete": True,
"result": result
}
def text_task_node(state: TaskState) -> dict:
"""
This node handles text-based tasks.
"""
print("Processing text task...")
messages = state.get("messages", [])
last_message = messages[-1] if messages else ""
# In a real application, this would use an LLM
result = f"Text analysis of: {last_message}"
return {
"messages": [result],
"task_complete": True,
"result": result
}
Now let's build a graph that uses conditional routing:
from langgraph.graph import StateGraph, END
def create_conditional_graph():
"""
Creates a graph with conditional routing based on task type.
"""
workflow = StateGraph(TaskState)
# Add both specialized nodes
workflow.add_node("math_task", math_task_node)
workflow.add_node("text_task", text_task_node)
# Set entry point to a router
# We need to add a node that determines the initial route
workflow.set_entry_point("math_task")
# Add conditional edges from each task node
# The router function determines where to go next
workflow.add_conditional_edges(
"math_task",
route_based_on_task,
{
"end": END,
"math_task": "math_task",
"text_task": "text_task"
}
)
workflow.add_conditional_edges(
"text_task",
route_based_on_task,
{
"end": END,
"math_task": "math_task",
"text_task": "text_task"
}
)
app = workflow.compile()
return app
The add_conditional_edges method is crucial here. It takes three arguments. First, the source node name. Second, the router function that makes the decision. Third, a mapping dictionary that maps the router's return values to actual node names or END.
Let's test this conditional graph:
app = create_conditional_graph()
# Test with a math task
math_state = {
"messages": ["Calculate the sum of 15 and 27"],
"task_type": "math calculation",
"task_complete": False,
"result": ""
}
result = app.invoke(math_state)
print(f"Math task result: {result['result']}")
# Test with a text task
text_state = {
"messages": ["Summarize the benefits of exercise"],
"task_type": "text summary",
"task_complete": False,
"result": ""
}
result = app.invoke(text_state)
print(f"Text task result: {result['result']}")
This example demonstrates how conditional edges enable your application to dynamically choose different execution paths based on the current state, making your LLM applications much more flexible and intelligent.
WORKING WITH LANGCHAIN MESSAGES: PROPER MESSAGE HANDLING
In real LangGraph applications, you'll typically work with LangChain's message types rather than plain strings. LangChain provides several message classes that represent different roles in a conversation.
Let's update our state definition to use proper message types:
from typing import TypedDict, Annotated, Sequence
from langchain.schema import BaseMessage, HumanMessage, AIMessage, SystemMessage
import operator
class ProperChatState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
iteration_count: int
The BaseMessage type is the parent class for all message types in LangChain. Using this in our state definition allows us to store any type of message (HumanMessage, AIMessage, SystemMessage, etc.) in our messages list.
Next let's create a node that properly handles these message types:
from langchain_openai import ChatOpenAI
def proper_chat_node(state: ProperChatState) -> dict:
"""
This node demonstrates proper message handling with LangChain types.
"""
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
# Get current messages from state
messages = state.get("messages", [])
# If this is the first iteration, add a system message
if state.get("iteration_count", 0) == 0:
system_message = SystemMessage(
content="You are a knowledgeable assistant specializing in "
"explaining technical concepts clearly and concisely."
)
messages = [system_message] + list(messages)
# Call the LLM with the properly formatted messages
response = llm.invoke(messages)
# The response is already an AIMessage object
print(f"Assistant response: {response.content[:100]}...")
return {
"messages": [response],
"iteration_count": state.get("iteration_count", 0) + 1
}
This proper_chat_node shows best practices for working with LangChain messages. The LLM's invoke method accepts a list of BaseMessage objects and returns an AIMessage object, which we can directly add to our state.
Let's create a helper function to make it easy to add user messages:
def add_user_message(current_state: ProperChatState, user_input: str) -> ProperChatState:
"""
Helper function to add a user message to the current state.
"""
new_message = HumanMessage(content=user_input)
# Create updated state with the new message
updated_state = current_state.copy()
updated_state["messages"] = list(current_state.get("messages", [])) + [new_message]
return updated_state
Now let's build a complete conversational graph using proper message types:
from langgraph.graph import StateGraph, END
def create_proper_chat_graph():
"""
Creates a chat graph using proper LangChain message types.
"""
workflow = StateGraph(ProperChatState)
workflow.add_node("chat", proper_chat_node)
workflow.set_entry_point("chat")
workflow.add_edge("chat", END)
app = workflow.compile()
return app
Let's use this graph in a multi-turn conversation:
app = create_proper_chat_graph()
# Initialize state with first user message
state = {
"messages": [HumanMessage(content="What is LangGraph?")],
"iteration_count": 0
}
# First turn
result = app.invoke(state)
print(f"Turn 1 - User: {result['messages'][0].content}")
print(f"Turn 1 - Assistant: {result['messages'][1].content[:200]}...")
# Add follow-up question
state = add_user_message(
result,
"Can you give me a simple example of how to use it?"
)
# Second turn
result = app.invoke(state)
print(f"\nTurn 2 - User: {result['messages'][2].content}")
print(f"Turn 2 - Assistant: {result['messages'][3].content[:200]}...")
This example demonstrates how to maintain a multi-turn conversation using proper message types. Each invocation of the graph builds upon the previous state, maintaining the full conversation history.
BUILDING A MULTI-AGENT RESEARCH SYSTEM: PRACTICAL APPLICATION
Now let's apply everything we've learned to build a practical multi-agent research system. This system will have multiple specialized agents that work together to research a topic, search for information, analyze findings, and synthesize a final answer.
First, let's define a comprehensive state for our research system:
from typing import TypedDict, Annotated, Sequence, Optional
from langchain.schema import BaseMessage
import operator
class ResearchAgentState(TypedDict):
# The original research question
question: str
# Conversation messages between agents
messages: Annotated[Sequence[BaseMessage], operator.add]
# Search queries generated by the planner
search_queries: Annotated[list[str], operator.add]
# Results from web searches
search_results: Annotated[list[dict], operator.add]
# Analysis of the search results
analysis: Annotated[list[str], operator.add]
# The final synthesized answer
final_answer: Optional[str]
# Current step in the research process
current_step: str
# Number of iterations performed
iteration_count: int
# Maximum iterations allowed
max_iterations: int
Now let's create the specialized agent nodes. First, a planner agent that generates search queries:
from langchain_openai import ChatOpenAI
from langchain.schema import SystemMessage, HumanMessage
def planner_agent_node(state: ResearchAgentState) -> dict:
"""
The planner agent analyzes the research question and generates
appropriate search queries to gather information.
"""
llm = ChatOpenAI(model="gpt-4", temperature=0.3)
question = state.get("question", "")
existing_queries = state.get("search_queries", [])
# Create a prompt for the planner
system_prompt = SystemMessage(
content="You are a research planner. Your job is to break down "
"research questions into specific, targeted search queries. "
"Generate 2-3 search queries that will help answer the question."
)
user_prompt = HumanMessage(
content=f"Research question: {question}\n\n"
f"Existing queries: {existing_queries}\n\n"
f"Generate new search queries to gather comprehensive information."
)
response = llm.invoke([system_prompt, user_prompt])
# Parse the response to extract queries (simplified for demonstration)
# In a real system, you'd use structured output or parsing
queries = [q.strip() for q in response.content.split("\n") if q.strip()]
print(f"Planner generated {len(queries)} new queries")
return {
"search_queries": queries,
"messages": [response],
"current_step": "planning_complete"
}
Next, a searcher agent that executes the search queries:
def searcher_agent_node(state: ResearchAgentState) -> dict:
"""
The searcher agent executes search queries and retrieves results.
In a real implementation, this would call actual search APIs.
"""
queries = state.get("search_queries", [])
print(f"Searcher executing {len(queries)} queries")
# Simulate search results (in reality, call actual search API)
all_results = []
for query in queries:
results = [
{
"query": query,
"title": f"Result 1 for {query}",
"snippet": f"This is detailed information about {query}. "
f"It contains relevant facts and data...",
"url": f"https://example.com/{query.replace(' ', '-')}"
},
{
"query": query,
"title": f"Result 2 for {query}",
"snippet": f"Additional information regarding {query}. "
f"This provides a different perspective...",
"url": f"https://example.org/{query.replace(' ', '-')}"
}
]
all_results.extend(results)
return {
"search_results": all_results,
"current_step": "search_complete"
}
Now an analyzer agent that processes the search results:
def analyzer_agent_node(state: ResearchAgentState) -> dict:
"""
The analyzer agent examines search results and extracts
key information relevant to the research question.
"""
llm = ChatOpenAI(model="gpt-4", temperature=0.3)
question = state.get("question", "")
results = state.get("search_results", [])
# Create analysis prompt
system_prompt = SystemMessage(
content="You are a research analyst. Analyze search results and "
"extract key information relevant to the research question. "
"Be thorough and identify important facts, patterns, and insights."
)
# Format search results for analysis
results_text = "\n\n".join([
f"Source: {r['title']}\n{r['snippet']}"
for r in results[-6:] # Analyze last 6 results
])
user_prompt = HumanMessage(
content=f"Research question: {question}\n\n"
f"Search results:\n{results_text}\n\n"
f"Provide a detailed analysis of these results."
)
response = llm.invoke([system_prompt, user_prompt])
print(f"Analyzer completed analysis: {response.content[:100]}...")
return {
"analysis": [response.content],
"messages": [response],
"current_step": "analysis_complete"
}
Finally, a synthesizer agent that creates the final answer:
def synthesizer_agent_node(state: ResearchAgentState) -> dict:
"""
The synthesizer agent combines all analyses into a
comprehensive final answer to the research question.
"""
llm = ChatOpenAI(model="gpt-4", temperature=0.5)
question = state.get("question", "")
analyses = state.get("analysis", [])
system_prompt = SystemMessage(
content="You are a research synthesizer. Your job is to combine "
"multiple analyses into a clear, comprehensive answer. "
"Provide a well-structured response that directly addresses "
"the research question."
)
# Combine all analyses
combined_analysis = "\n\n".join([
f"Analysis {i+1}:\n{analysis}"
for i, analysis in enumerate(analyses)
])
user_prompt = HumanMessage(
content=f"Research question: {question}\n\n"
f"Analyses:\n{combined_analysis}\n\n"
f"Synthesize a comprehensive final answer."
)
response = llm.invoke([system_prompt, user_prompt])
print(f"Synthesizer created final answer: {response.content[:100]}...")
return {
"final_answer": response.content,
"messages": [response],
"current_step": "synthesis_complete"
}
Now we need a router function to coordinate these agents:
from typing import Literal
def research_router(
state: ResearchAgentState
) -> Literal["planner", "searcher", "analyzer", "synthesizer", "end"]:
"""
Routes the workflow between different research agents based
on the current step and iteration count.
"""
current_step = state.get("current_step", "start")
iteration = state.get("iteration_count", 0)
max_iterations = state.get("max_iterations", 2)
# Check if we've reached maximum iterations
if iteration >= max_iterations:
# If we have analysis, synthesize; otherwise end
if state.get("analysis"):
if current_step != "synthesis_complete":
return "synthesizer"
return "end"
# Route based on current step
if current_step == "start":
return "planner"
elif current_step == "planning_complete":
return "searcher"
elif current_step == "search_complete":
return "analyzer"
elif current_step == "analysis_complete":
# Decide whether to iterate or synthesize
if iteration < max_iterations - 1:
return "planner" # Do another iteration
else:
return "synthesizer"
elif current_step == "synthesis_complete":
return "end"
return "end"
Now let's build the complete multi-agent research graph:
from langgraph.graph import StateGraph, END
def create_research_agent_graph():
"""
Creates a multi-agent research system with planner, searcher,
analyzer, and synthesizer agents working together.
"""
workflow = StateGraph(ResearchAgentState)
# Add all agent nodes
workflow.add_node("planner", planner_agent_node)
workflow.add_node("searcher", searcher_agent_node)
workflow.add_node("analyzer", analyzer_agent_node)
workflow.add_node("synthesizer", synthesizer_agent_node)
# Set entry point
workflow.set_entry_point("planner")
# Add conditional edges from each node using the router
for node_name in ["planner", "searcher", "analyzer", "synthesizer"]:
workflow.add_conditional_edges(
node_name,
research_router,
{
"planner": "planner",
"searcher": "searcher",
"analyzer": "analyzer",
"synthesizer": "synthesizer",
"end": END
}
)
# Compile the graph
app = workflow.compile()
return app
Let's use our multi-agent research system:
# Create the research agent graph
research_app = create_research_agent_graph()
# Define a research question
initial_state = {
"question": "What are the key benefits and challenges of using "
"LangGraph for building multi-agent LLM applications?",
"messages": [],
"search_queries": [],
"search_results": [],
"analysis": [],
"final_answer": None,
"current_step": "start",
"iteration_count": 0,
"max_iterations": 2
}
# Execute the research workflow
final_state = research_app.invoke(initial_state)
# Display results
print("\n" + "="*80)
print("RESEARCH COMPLETE")
print("="*80)
print(f"\nQuestion: {final_state['question']}")
print(f"\nQueries executed: {len(final_state['search_queries'])}")
print(f"Results gathered: {len(final_state['search_results'])}")
print(f"Iterations: {final_state['iteration_count']}")
print(f"\nFinal Answer:\n{final_state['final_answer']}")
This multi-agent research system demonstrates the power of LangGraph. Multiple specialized agents work together, each handling a specific aspect of the research process. The router coordinates their activities, and the shared state allows them to build upon each other's work.
PERSISTENCE AND CHECKPOINTING: SAVING YOUR WORKFLOW STATE
One of LangGraph's powerful features is the ability to persist workflow state and create checkpoints. This allows you to pause and resume workflows, implement human-in-the-loop patterns, and recover from failures.
To enable persistence, you need to provide a checkpointer when compiling your graph. LangGraph supports various checkpointer implementations. Let's use the MemorySaver for demonstration:
from langgraph.checkpoint.memory import MemorySaver
def create_persistent_chat_graph():
"""
Creates a chat graph with state persistence enabled.
"""
workflow = StateGraph(ProperChatState)
workflow.add_node("chat", proper_chat_node)
workflow.set_entry_point("chat")
workflow.add_edge("chat", END)
# Create a memory-based checkpointer
memory = MemorySaver()
# Compile with checkpointer
app = workflow.compile(checkpointer=memory)
return app
When using a persistent graph, you need to provide a thread_id to identify different conversation threads:
persistent_app = create_persistent_chat_graph()
# Configuration with thread ID
config = {"configurable": {"thread_id": "conversation-1"}}
# First message in thread
state1 = {
"messages": [HumanMessage(content="Hello, what is LangGraph?")],
"iteration_count": 0
}
result1 = persistent_app.invoke(state1, config)
print(f"Response 1: {result1['messages'][-1].content[:100]}...")
# Continue the same thread with a follow-up
state2 = {
"messages": [HumanMessage(content="Can you give me an example?")],
"iteration_count": result1["iteration_count"]
}
result2 = persistent_app.invoke(state2, config)
print(f"Response 2: {result2['messages'][-1].content[:100]}...")
# The checkpointer maintains the full conversation history
print(f"\nTotal messages in thread: {len(result2['messages'])}")
The checkpointer automatically saves the state after each node execution. This enables powerful patterns like human-in-the-loop workflows where you can pause execution, get human input, and then resume.
HUMAN-IN-THE-LOOP PATTERNS: INTERACTIVE WORKFLOWS
LangGraph makes it easy to implement human-in-the-loop patterns where human input is required at certain points in the workflow. Let's build an example where a human reviewer approves or rejects content before it's finalized.
First, let's define a state that tracks approval status:
from typing import TypedDict, Annotated, Sequence, Optional, Literal
from langchain.schema import BaseMessage
import operator
class ApprovalState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
draft_content: Optional[str]
human_feedback: Optional[str]
approval_status: Optional[Literal["pending", "approved", "rejected"]]
final_content: Optional[str]
Now let's create nodes for content generation and revision:
from langchain_openai import ChatOpenAI
from langchain.schema import SystemMessage, HumanMessage
def generate_content_node(state: ApprovalState) -> dict:
"""
Generates initial draft content based on the user's request.
"""
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
messages = state.get("messages", [])
system_prompt = SystemMessage(
content="You are a content writer. Create high-quality content "
"based on the user's request."
)
response = llm.invoke([system_prompt] + list(messages))
print(f"Generated draft content: {response.content[:100]}...")
return {
"draft_content": response.content,
"messages": [response],
"approval_status": "pending"
}
def revise_content_node(state: ApprovalState) -> dict:
"""
Revises content based on human feedback.
"""
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
draft = state.get("draft_content", "")
feedback = state.get("human_feedback", "")
system_prompt = SystemMessage(
content="You are a content editor. Revise the draft based on "
"the feedback provided."
)
user_prompt = HumanMessage(
content=f"Draft:\n{draft}\n\nFeedback:\n{feedback}\n\n"
f"Please revise the content accordingly."
)
response = llm.invoke([system_prompt, user_prompt])
print(f"Revised content: {response.content[:100]}...")
return {
"draft_content": response.content,
"messages": [response],
"approval_status": "pending"
}
def finalize_content_node(state: ApprovalState) -> dict:
"""
Finalizes approved content.
"""
draft = state.get("draft_content", "")
print("Content approved and finalized!")
return {
"final_content": draft,
"approval_status": "approved"
}
Now let's create a router that handles the approval workflow:
def approval_router(
state: ApprovalState
) -> Literal["generate", "revise", "finalize", "human_review"]:
"""
Routes based on approval status and human feedback.
"""
status = state.get("approval_status")
if status is None:
return "generate"
elif status == "pending":
return "human_review"
elif status == "rejected":
return "revise"
elif status == "approved":
return "finalize"
return "human_review"
Here's how you would build the graph with human-in-the-loop:
from langgraph.graph import StateGraph, END
def create_approval_workflow():
"""
Creates a workflow that requires human approval.
"""
workflow = StateGraph(ApprovalState)
workflow.add_node("generate", generate_content_node)
workflow.add_node("revise", revise_content_node)
workflow.add_node("finalize", finalize_content_node)
workflow.set_entry_point("generate")
# After generation, always go to human review (simulated)
workflow.add_edge("generate", "finalize")
workflow.add_edge("revise", "finalize")
workflow.add_edge("finalize", END)
app = workflow.compile()
return app
In a real implementation with checkpointing, you would pause execution before the human review step, wait for human input, and then resume with the updated state.
STREAMING OUTPUTS: REAL-TIME FEEDBACK
LangGraph supports streaming, which allows you to get real-time updates as your graph executes. This is particularly useful for long-running workflows or when you want to provide immediate feedback to users.
Here's how to use streaming:
# Create a graph (using our research agent as an example)
research_app = create_research_agent_graph()
initial_state = {
"question": "What is LangGraph?",
"messages": [],
"search_queries": [],
"search_results": [],
"analysis": [],
"final_answer": None,
"current_step": "start",
"iteration_count": 0,
"max_iterations": 1
}
# Stream the execution
print("Streaming research workflow:")
print("-" * 80)
for output in research_app.stream(initial_state):
# Each output is a dictionary with node name as key
for node_name, node_output in output.items():
print(f"\nNode '{node_name}' completed")
print(f"Current step: {node_output.get('current_step', 'N/A')}")
# You can access any part of the state here
if 'final_answer' in node_output and node_output['final_answer']:
print(f"Final answer ready: {node_output['final_answer'][:100]}...")
print("\n" + "-" * 80)
print("Workflow complete!")
The stream method yields the output of each node as it completes, allowing you to provide real-time progress updates to users or log detailed execution information.
BEST PRACTICES AND PATTERNS
As you build more complex LangGraph applications, following these best practices will help you create maintainable, efficient, and reliable systems.
First, design your state schema carefully. Your state should contain all the information that needs to flow between nodes, but avoid making it overly complex. Group related information together and use clear, descriptive field names. Use Optional types for fields that might not always be present.
Second, keep your nodes focused and single-purpose. Each node should perform one clear task. This makes your graph easier to understand, test, and debug. If a node is doing too many things, consider splitting it into multiple nodes.
Third, use meaningful node names that clearly describe what the node does. Names like "planner", "searcher", and "analyzer" are much better than "node1", "node2", and "node3". Good names make your graph self-documenting.
Fourth, implement proper error handling in your nodes. Wrap LLM calls and external API calls in try-except blocks. When an error occurs, update the state to reflect the error condition so your router can handle it appropriately.
Here's an example of a node with proper error handling:
def robust_llm_node(state: ProperChatState) -> dict:
"""
An LLM node with comprehensive error handling.
"""
try:
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
messages = state.get("messages", [])
if not messages:
raise ValueError("No messages to process")
response = llm.invoke(messages)
return {
"messages": [response],
"iteration_count": state.get("iteration_count", 0) + 1
}
except Exception as e:
print(f"Error in LLM node: {str(e)}")
# Return an error message in the state
error_message = AIMessage(
content=f"I encountered an error: {str(e)}. "
f"Please try rephrasing your question."
)
return {
"messages": [error_message],
"iteration_count": state.get("iteration_count", 0) + 1
}
Fifth, use type hints consistently throughout your code. This helps catch errors early and makes your code more maintainable. LangGraph works well with Python's type system, so take advantage of it.
Sixth, test your nodes independently before integrating them into a graph. Each node is just a Python function, so you can easily write unit tests for them:
def test_increment_counter():
"""
Test the increment counter node.
"""
test_state = {"counter": 5, "message": "test"}
result = increment_counter_node(test_state)
assert result["counter"] == 6
print("Test passed: Counter incremented correctly")
test_increment_counter()
Seventh, use logging to track execution flow. This is invaluable for debugging complex graphs:
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def logged_node(state: SimpleState) -> dict:
"""
A node that logs its execution.
"""
logger.info(f"Node executing with state: {state}")
result = {"counter": state.get("counter", 0) + 1}
logger.info(f"Node returning: {result}")
return result
Eighth, when building multi-agent systems, clearly define each agent's responsibility. Avoid overlap between agents. Each agent should have a distinct role that contributes to the overall goal.
Ninth, use conditional edges to implement retry logic and error recovery. If a node fails or produces unsatisfactory results, your router can direct execution to a retry node or an alternative path.
Tenth, for production systems, use persistent checkpointers (not just MemorySaver) to ensure state is preserved across application restarts. LangGraph supports various backend storage options for checkpointing.
ADVANCED PATTERNS: SUBGRAPHS AND COMPOSITION
As your applications grow more complex, you may want to compose multiple graphs together. LangGraph supports this through subgraphs, where one graph can be used as a node in another graph.
Let's create a simple example with a subgraph:
from langgraph.graph import StateGraph, END
# Define state for the subgraph
class SubGraphState(TypedDict):
input_value: int
output_value: int
def double_node(state: SubGraphState) -> dict:
"""
Doubles the input value.
"""
value = state.get("input_value", 0)
return {"output_value": value * 2}
def create_doubling_subgraph():
"""
Creates a simple subgraph that doubles a value.
"""
workflow = StateGraph(SubGraphState)
workflow.add_node("double", double_node)
workflow.set_entry_point("double")
workflow.add_edge("double", END)
return workflow.compile()
Now you can use this subgraph as a node in a larger graph. This pattern is useful for organizing complex workflows into modular, reusable components.
CONCLUSION: YOUR JOURNEY WITH LANGGRAPH
Congratulations! You have now learned the fundamental concepts and patterns for building sophisticated multi-agent LLM applications with LangGraph. Let's recap what we've covered.
We started by understanding what LangGraph is and why it's valuable for building complex LLM applications. We learned that LangGraph provides a graph-based framework for orchestrating multiple LLM calls, managing state, and implementing conditional logic.
We explored the four core concepts: State, which represents the shared information flowing through your application; Graphs, which define the overall structure; Nodes, which perform discrete units of work; and Edges, which control execution flow.
We learned how to define state schemas using TypedDict and how to use the Annotated type with operator.add to accumulate values rather than replace them. This is crucial for maintaining conversation history and building up context.
We created various types of nodes, from simple functions that increment counters to sophisticated agents that call LLMs, perform web searches, analyze data, and synthesize results. We saw how nodes receive state, perform operations, and return state updates.
We explored both normal edges for sequential flow and conditional edges for decision-making. We learned how to write router functions that examine state and determine the next node to execute, enabling dynamic, intelligent workflows.
We built a complete multi-agent research system with specialized agents for planning, searching, analyzing, and synthesizing information. This demonstrated how multiple agents can work together, coordinated by a router, to accomplish complex tasks.
We learned about persistence and checkpointing, which allow you to save workflow state, implement human-in-the-loop patterns, and recover from failures. We saw how to use streaming to get real-time updates as graphs execute.
Finally, we covered best practices including careful state design, focused single-purpose nodes, meaningful naming, error handling, type hints, testing, logging, and modular composition.
You now have the knowledge to build your own LangGraph applications. Start with simple graphs to get comfortable with the concepts, then gradually increase complexity as you gain confidence. Remember that the key to success with LangGraph is thinking in terms of graphs: what are the steps in your workflow, what information needs to flow between them, and what decisions need to be made along the way.
The LangGraph library continues to evolve with new features and capabilities. The patterns and concepts you've learned here provide a solid foundation that will serve you well as you explore more advanced features and build increasingly sophisticated applications.
Happy building, and may your LLM applications be stateful, intelligent, and powerful!
No comments:
Post a Comment