Langgraph Multi Agent Application
This article is part one of a learning-focused series where I build a multi-agent system using LangGraph and Claude.
The goal of this project is to understand how multiple specialized agents can perform real-world operational tasks in a structured and scalable way.
1. Kubernetes Cluster Operations
The system has access to Kubernetes clusters and can answer queries related to cluster state and resources, such as:
- Listing Kubernetes resources (pods, deployments, services, etc.)
- Querying namespaces and workloads
- Providing high-level cluster information
These queries are handled through a dedicated Kubernetes-aware agent.
2. Redis Key Management
The system can interact with Redis and perform basic key management operations, including:
- Listing Redis keys
- Deleting specific Redis keys
This capability is implemented through a separate Redis-focused agent, ensuring clear separation of responsibilities.
3. Application Release Information
The system can provide information related to application releases, such as:
- Current release.
- Release metadata
- Historical release details (if available).
Reducers:
A Reducer in LangGraph is a function that takes in multiple inputs, like messages, states, or results from different parts of your graph, and reduces them into a single, unified output.
If that sounds abstract, think of it like this:
“You have three agents each giving different answers to the same question. Instead of picking one randomly, you can use a Reducer to combine their responses into a single, well-rounded answer.”
In programming terms, Reducers are similar to the reduce() function you may have seen in JavaScript or Python. They take a list of things and boil them down into one. In LangGraph, this means they can:
- Merge messages from multiple nodes.
- Summarize outputs from different agents.
- Keep your graph’s state clean and consistent.
Why does this matter? Because as your workflow grows, you’ll often have multiple branches running at once — and eventually, you’ll need a way to bring everything back together. Reducers make that process simple and predictable.
So, in short:
A Reducer helps your LangGraph workflow make sense of many outputs by turning them into one clear, useful result.
Reducers are key to understanding how updates from nodes are applied to the State. Each key in the State has its own independent reducer function. If no reducer function is explicitly specified then it is assumed that all updates to that key should override it. There are a few different types of reducers, starting with the default type of reducer:
Default reducer
These two examples show how to use the default reducer:
from typing_extensions import TypedDict
class State(TypedDict):
foo: int
bar: list[str]In this example, no reducer functions are specified for any key. Let’s assume the input to the graph is: {"foo": 1, "bar": ["hi"]}. Let’s then assume the first Node returns {"foo": 2}. This is treated as an update to the state. Notice that the Node does not need to return the whole State schema - just an update. After applying this update, the State would then be {"foo": 2, "bar": ["hi"]}. If the second node returns {"bar": ["bye"]} then the State would then be {"foo": 2, "bar": ["bye"]}
from typing import Annotated
from typing_extensions import TypedDict
from operator import add
class State(TypedDict):
foo: int
bar: Annotated[list[str], add]In this example, we’ve used the Annotated type to specify a reducer function (operator.add) for the second key (bar). Note that the first key remains unchanged. Let’s assume the input to the graph is {"foo": 1, "bar": ["hi"]}. Let’s then assume the first Node returns {"foo": 2}. This is treated as an update to the state. Notice that the Node does not need to return the whole State schema - just an update. After applying this update, the State would then be {"foo": 2, "bar": ["hi"]}. If the second node returns {"bar": ["bye"]} then the State would then be {"foo": 2, "bar": ["hi", "bye"]}. Notice here that the bar key is updated by adding the two lists together.
Default Reducer (Overwrite): If you don’t specify a reducer for a state key, any new update from a node simply replaces the old value.
Accumulator Reducers: These combine new values with existing ones.
add: Adds new values (e.g., for a running total).concat: Appends to lists or strings.
Custom Reducers: You can define your own functions for complex logic, like appending to a list while ensuring uniqueness or handling specific data structures.
Specialized Reducers:
add_messages: A built-in workhorse for managing conversation history, appending messages while correctly handling updates to existing ones by ID.- Annotated: This Python type hint pairs a data type with a reducer, instructing LangGraph how to apply updates, crucial for stateful agents
add_messages:
add_messages is a built-in LangGraph reducer that updates your agent’s message history by adding new messages to the existing state.
That’s it. No magic, just a clean way to keep track of every message in a conversation so your agent remembers what’s already happened.
You can think of it as your agent’s note-taker.
Every time someone speaks — whether it’s the user, the assistant, or a system message, add_messages records that message in order so the context stays consistent.
In LangGraph, a message is just a small dictionary that looks like this:
{"role": "user", "content": "Hello!"}When you use add_messages, it takes your current list of messages and appends the new one:
Before:
messages = [{"role": "user", "content": "Hi!"}]
After add_messages("How are you?"):
[
{"role": "user", "content": "Hi!"},
{"role": "user", "content": "How are you?"}
]This small reducer is what helps your LangGraph agent maintain the flow of a conversation. Without it, each new message would feel like starting from scratch.
Next, we’ll look at how add_messages actually works and what happens behind the scenes when it updates your graph’s state.
How It Works:
Now that you know what add_messages does, let’s see how it works behind the scenes.
When you call add_messages, two things happen:
- It looks at the current state of your LangGraph agent.
- It appends new messages to the existing message list.
You give it a state and a message or list of messages, and it returns a new state with those messages added in.
Nothing fancy, just clean state management that makes conversation tracking reliable.
Here’s a quick example:
from langgraph.reducers import add_messages
state = {"messages": []}
new_state = add_messages(
state,
{"role": "user", "content": "Hello, AI!"}
)
print(new_state)Output:
{'messages': [{'role': 'user', 'content': 'Hello, AI!'}]}
If you add another message, it simply builds on the previous one:
next_state = add_messages(
new_state,
{"role": "assistant", "content": "Hi there! How can I help you?"}
)
print(next_state)
Output:
{
'messages': [
{'role': 'user', 'content': 'Hello, AI!'},
{'role': 'assistant', 'content': 'Hi there! How can I help you?'}
]
}add_messages always appends new entries while keeping the existing conversation intact.
This makes it perfect for agents that need to remember context across multiple turns.
update_dialog_stack
def update_dialog_stack(left: list[str], right: Optional[str]) -> list[str]:
if right is None:
return left
if right == "pop":
return left[:-1]
return left + [right]update_dialog_stack is reducer function for managing the dialog state stack.
This function is called automatically by LangGraph whenever a node updates the dialog_state field. It implements a stack-based state management system.
Parameters:
left: Current dialog_state stack (e.g., [“primary”, “k8s_agent”])
right: New update to apply:
. None: Keep stack unchanged
. pop: Remove last item from stack (exit current agent)
. String (e.g., “k8s_agent”): Add to stack (enter new agent)
Returns:
Updated dialog_state stack (list of strings)
if right is None:
# Case 1: No update - keep current stack unchanged
return left
if right == "pop":
# Case 2: Pop from stack - exit current agent, return to previous
return left[:-1]
# Case 3: Push to stack - enter new agent, add to stack
return left + [right]Example 1: Simple Query “List all pods in hotels namespace”
Get Syed Muhammad Hassan’s stories in your inbox
Join Medium for free to get updates from this writer.
Step 1: Initial State
state = {
'messages': [HumanMessage("list all pods in hotels namespace")],
'dialog_state': [] # ← Empty = primary agent active
}Step 2: Primary Agent Processes
# primary_assistant node processes query
# LLM decides to call ToGetKubernetesInfo tool
# Node returns:
{
"messages": [AIMessage(tool_calls=[ToGetKubernetesInfo(...)])]
# No dialog_state update
}
# LangGraph calls:
update_dialog_stack([], None)
# Returns: [] (unchanged - still primary agent)
`dialog_state = []` (primary agent still active)Step 3: Entry Node Pushes to Stack
# enter_k8s_info node executes
# LangGraph calls:
update_dialog_stack([], "k8s_agent")
# left = [] (current stack)
# right = "k8s_agent" (new agent to enter)
# Returns: ["k8s_agent"] (pushed to stack)
`dialog_state = ["k8s_agent"]` (Kubernetes agent now active)Step 4: Kubernetes Agent Processes
# k8s_agent node processes query
# Calls list_k8s_pods tool
# Node returns:
{
"messages": [AIMessage(tool_calls=[list_k8s_pods(...)])]
# No dialog_state update
}
# LangGraph calls:
update_dialog_stack(["k8s_agent"], None)
# Returns: ["k8s_agent"] (unchanged - still k8s_agent)
`dialog_state = ["k8s_agent"]` (still active)Step 5: Tool Execution
# update_k8s_tools node executes tool
# Returns tool results
# No dialog_state update
# LangGraph calls:
update_dialog_stack(["k8s_agent"], None)
# Returns: ["k8s_agent"] (unchanged)
```
**Result:** `dialog_state = ["k8s_agent"]` (still active)Step 6: Final Response
# k8s_agent generates final response
# No dialog_state update
# Routes to END
# LangGraph calls:
update_dialog_stack(["k8s_agent"], None)
# Returns: ["k8s_agent"] (unchanged)
```
**Final State:** `dialog_state = ["k8s_agent"]` (for tracking purposes)Example 2: Agent Completes Task “CompleteOrEscalate” (Normal Flow)
User asks “list all pods in hotels namespace”, Kubernetes agent completes the task and calls `CompleteOrEscalate` to return control to primary agent
Agents call it when:
- Task is fully completed (e.g., “I have fully completed the task.”)
- User wants to switch tasks (e.g., “User changed their mind about the current task.”)
- Agent needs to return control for any reason
Step 1: Agent Completes Task and Calls CompleteOrEscalate
# k8s_agent has successfully listed pods and provided answer
# Agent calls CompleteOrEscalate tool to return control
# RouteUpdater detects CompleteOrEscalate and routes to "leave_skill" node
# Current state:
dialog_state = ["k8s_agent"]Step 2: RouteUpdater Detects CompleteOrEscalate
# RouteUpdater.route_update_info (helper.py line 34)
did_cancel = any(tc["name"] == CompleteOrEscalate.__name__ for tc in tool_calls)
# did_cancel = True
if did_cancel:
return "leave_skill" # Routes to leave_skill nodeStep 3: Pop Dialog State
# pop_dialog_state node executes (helper.py line 174)
# Returns: {"dialog_state": "pop"}
# LangGraph calls:
update_dialog_stack(["k8s_agent"], "pop")
# left = ["k8s_agent"] (current stack)
# right = "pop" (exit command)
# Returns: [] (popped from stack)
```
**Result:** `dialog_state = []` (back to primary agent)Step 4: Return to Primary Agent
# leave_skill node has edge to primary_assistant
# Primary agent receives full conversation context
# Can provide final response or handle new queries
# State after pop:
dialog_state = [] # ← Primary agent active againMessages:
Messages are the fundamental unit of context for models in LangChain. They represent the input and output of models, carrying both the content and metadata needed to represent the state of a conversation when interacting with an LLM. Messages are objects that contain:
- Role — Identifies the message type (e.g.
system,user) - Content — Represents the actual content of the message (like text, images, audio, documents, etc.)
- Metadata — Optional fields such as response information, message IDs, and token usage
LangChain provides a standard message type that works across all model providers, ensuring consistent behavior regardless of the model being called.
Message types
- System message — Tells the model how to behave and provide context for interactions
- Human message — Represents user input and interactions with the model
- AI message — Responses generated by the model, including text content, tool calls, and metadata
- Tool message — Represents the outputs of tool calls
System message
A SystemMessage represent an initial set of instructions that primes the model’s behavior. You can use a system message to set the tone, define the model’s role, and establish guidelines for responses.
system_msg = SystemMessage("You are a helpful coding assistant.")
messages = [
system_msg,
HumanMessage("How do I create a REST API?")
]
response = model.invoke(messages)from langchain.messages import SystemMessage, HumanMessage
system_msg = SystemMessage("""
You are a senior Python developer with expertise in web frameworks.
Always provide code examples and explain your reasoning.
Be concise but thorough in your explanations.
""")
messages = [
system_msg,
HumanMessage("How do I create a REST API?")
]
response = model.invoke(messages)Human message
A HumanMessage represents user input and interactions. They can contain text, images, audio, files, and any other amount of multimodal content.
response = model.invoke([
HumanMessage("What is machine learning?")
])AI message
An AIMessage represents the output of a model invocation. They can include multimodal data, tool calls, and provider-specific metadata that you can later access.
response = model.invoke("Explain AI")
print(type(response)) # <class 'langchain.messages.AIMessage'>AIMessage objects are returned by the model when calling it, which contains all of the associated metadata in the response. Providers weigh/contextualize types of messages differently, which means it is sometimes helpful to manually create a new AIMessage object and insert it into the message history as if it came from the model.
from langchain.messages import AIMessage, SystemMessage, HumanMessage
# Create an AI message manually (e.g., for conversation history)
ai_msg = AIMessage("I'd be happy to help you with that question!")
# Add to conversation history
messages = [
SystemMessage("You are a helpful assistant"),
HumanMessage("Can you help me?"),
ai_msg, # Insert as if it came from the model
HumanMessage("Great! What's 2+2?")
]
response = model.invoke(messages)State:
State in LangGraph is a way to maintain and track information as an AI system processes data. Think of it as the system’s memory, allowing it to remember and update information as it moves through different stages of a workflow, or graph.
A shared data structure that holds the application’s current snapshot, accessible and updatable by all agents.
This basic state definition creates a simple structure that can hold a single integer value, though it could be any value. While it may seem elementary, this type of state can be incredibly useful in many scenarios, such as:
- Tracking the number of turns in a conversation
- Counting the occurrences of a specific event
- Maintaining a simple score, metric, or AI output
from typing import List, Dict, Optional, Literal, Annotated, Any
from typing_extensions import TypedDict
from langgraph.graph.message import AnyMessage, add_messages
def update_dialog_stack(left: list[str], right: Optional[str]) -> list[str]:
if right is None:
return left
if right == "pop":
return left[:-1]
return left + [right]
class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
dialog_state: Annotated[
list[
Literal[
"primary",
"k8s_agent",
"redis_agent",
"app_deployment_info",
]
],
update_dialog_stack,
]
agent_results: Dict[str, Any]
current_query: Optional[str]
metadata: Dict[str, Any]
agents_invoked: List[str]
final_response: Optional[str]This state maintains a list of messages. The Annotated type provides additional metadata that LangGraph uses for special handling of message lists (Tuples). This structure is particularly useful for:
- Chatbots that need to remember conversation history
- AI assistants that maintain context over multiple interactions
- Systems that need to track both quantitative (count) and qualitative (messages) data
References:
https://dzone.com/articles/langchain-langgraph-langflow-langsmith-ai-guide
