Hierarchical Multi-Agent Supervisor Swarms in LangGraph: Building Router-Worker Systems with Redis State Persistence in Python

python3 -m venv .venv
source .venv/bin/activate
pip install langchain langchain-openai langgraph langgraph-checkpoint-redis redis pydantic tavily-python

export OPENAI_API_KEY="your-openai-api-key"
export TAVILY_API_KEY="your-tavily-api-key"
export REDIS_URL="redis://localhost:6379"

import operator
from typing import Annotated, List, Literal, Sequence, TypedDict
from pydantic import BaseModel, Field
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

# Global State shared between the Supervisor and Workers
class GlobalState(TypedDict):
 """
 Top-level state managed by the Supervisor.

 Attributes:
 messages: Full conversation history and executive worker summaries.
 next_worker: The next worker node to execute, or 'FINISH'.
 task_summary: Running synthesis of completed sub-goals.
 iteration_count: Safety circuit breaker preventing infinite routing loops.
 """
 messages: Annotated[Sequence[BaseMessage], operator.add]
 next_worker: str
 task_summary: str
 iteration_count: int

# Supervisor Router Schema for Structured Output
class RouterSchema(BaseModel):
 """Routing decision determining the next worker or termination."""
 next_worker: Literal["Researcher", "Coder", "FINISH"] = Field(
 description="The next specialized worker to invoke. Select 'FINISH' when all tasks are complete."
 )
 instructions: str = Field(
 description="Specific instructions and bounded context for the selected worker."
 )
 progress_notes: str = Field(
 description="Brief summary of progress made so far toward the global objective."
 )

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

supervisor_llm = ChatOpenAI(model="gpt-4o", temperature=0)

supervisor_prompt = ChatPromptTemplate.from_messages([
 ("system", (
 "You are the Lead Technical Supervisor orchestrating a multi-agent engineering team.\n"
 "Your available workers are:\n"
 "- 'Researcher': Gathers facts, API specifications, and architectural documentation.\n"
 "- 'Coder': Writes, tests, and executes clean Python scripts based on gathered specifications.\n\n"
 "Analyze the current conversation history and task summary.\n"
 "If more research is needed, route to 'Researcher'.\n"
 "If research is complete and code needs to be written, route to 'Coder'.\n"
 "If all user requirements are fully satisfied and verified, route to 'FINISH'.\n"
 "Never delegate tasks that have already been completed."
 )),
 ("human", "Conversation History:\n{messages}\n\nCurrent Task Summary:\n{task_summary}")
])

supervisor_chain = supervisor_prompt | supervisor_llm.with_structured_output(RouterSchema)

def supervisor_node(state: GlobalState) -> dict:
 """Evaluates task progress and determines the next delegation step."""
 iteration = state.get("iteration_count", 0)

 # Circuit Breaker: Enforce maximum routing iterations
 if iteration >= 10:
 return {
 "next_worker": "FINISH",
 "messages": [AIMessage(content="[Supervisor Guard] Maximum iteration limit reached. Terminating swarm.")]
 }

 decision: RouterSchema = supervisor_chain.invoke({
 "messages": state["messages"],
 "task_summary": state.get("task_summary", "Initial request received.")
 })

 return {
 "next_worker": decision.next_worker,
 "task_summary": decision.progress_notes,
 "iteration_count": iteration + 1,
 "messages": [AIMessage(content=f"[Supervisor] Routing to {decision.next_worker}: {decision.instructions}")]
 }

from langchain_community.tools.tavily_search import TavilySearchResults

worker_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
search_tool = TavilySearchResults(max_results=3)

def researcher_worker(state: GlobalState) -> dict:
 """
 Executes information gathering in an isolated context.
 Filters raw search outputs to prevent global context pollution.
 """
 last_message = state["messages"][-1].content

 # Extract query from supervisor instructions
 research_prompt = ChatPromptTemplate.from_messages([
 ("system", "You are an expert technical researcher. Formulate a search query, extract factual findings, and return a concise technical briefing."),
 ("human", "Supervisor Instructions: {instructions}")
 ])

 # Execute targeted research
 search_query = worker_llm.invoke(f"Extract a concise search query from: {last_message}").content
 raw_results = search_tool.invoke({"query": search_query})

 # Synthesize clean findings
 synthesis_prompt = ChatPromptTemplate.from_messages([
 ("system", "Synthesize the following search results into a clean, factual summary for a software engineer. Exclude boilerplate."),
 ("human", "Results:\n{results}\n\nObjective: {objective}")
 ])
 summary_chain = synthesis_prompt | worker_llm
 briefing = summary_chain.invoke({"results": str(raw_results), "objective": last_message})

 # Return structured summary to the global state
 return {
 "messages": [AIMessage(content=f"[Researcher Output]\n{briefing.content}")]
 }

def coder_worker(state: GlobalState) -> dict:
 """
 Generates and verifies production code based on researcher briefings.
 """
 history_text = "\n".join([f"{m.type}: {m.content}" for m in state["messages"]])

 coding_prompt = ChatPromptTemplate.from_messages([
 ("system", (
 "You are a Senior Python Systems Engineer.\n"
 "Review the research briefings provided by the Researcher.\n"
 "Write production-grade, typed Python code satisfying the user's requirements.\n"
 "Include error handling, clear docstrings, and zero placeholder comments."
 )),
 ("human", "Context and Requirements:\n{history}")
 ])

 coder_chain = coding_prompt | ChatOpenAI(model="gpt-4o", temperature=0.1)
 code_artifact = coder_chain.invoke({"history": history_text})

 return {
 "messages": [AIMessage(content=f"[Coder Output]\n{code_artifact.content}")]
 }

from langgraph.graph import StateGraph, END

# Initialize Parent Graph
workflow = StateGraph(GlobalState)

# Add Nodes
workflow.add_node("Supervisor", supervisor_node)
workflow.add_node("Researcher", researcher_worker)
workflow.add_node("Coder", coder_worker)

# Set Entry Point
workflow.set_entry_point("Supervisor")

# Conditional Router Function
def route_next(state: GlobalState) -> str:
 """Routes execution based on the Supervisor's decision."""
 next_node = state["next_worker"]
 if next_node == "FINISH":
 return END
 return next_node

# Add Conditional Edges from Supervisor to Workers
workflow.add_conditional_edges(
 "Supervisor",
 route_next,
 {
 "Researcher": "Researcher",
 "Coder": "Coder",
 END: END
 }
)

# Workers always report back to the Supervisor for verification
workflow.add_edge("Researcher", "Supervisor")
workflow.add_edge("Coder", "Supervisor")

import asyncio
from langgraph_checkpoint_redis.aio import AsyncRedisSaver

async def initialize_production_swarm():
 """Compiles the StateGraph with distributed Redis state persistence."""
 redis_conn_string = "redis://localhost:6379"

 # Initialize asynchronous Redis connection pool
 checkpointer = AsyncRedisSaver.from_conn_string(redis_conn_string)

 # Compile graph with persistence
 app = workflow.compile(checkpointer=checkpointer)
 return app

async def run_swarm_task(user_request: str, session_id: str):
 """Executes a composite task with thread-isolated state persistence."""
 app = await initialize_production_swarm()

 # Unique configuration namespace for thread isolation
 config = {"configurable": {"thread_id": session_id}}

 initial_state = {
 "messages": [HumanMessage(content=user_request)],
 "next_worker": "Supervisor",
 "task_summary": "",
 "iteration_count": 0
 }

 print(f"\n--- Initiating Session: {session_id} ---")
 async for event in app.astream(initial_state, config=config):
 for node_name, output in event.items():
 print(f"\n[Transition: Node '{node_name}']")
 if "messages" in output and len(output["messages"]) > 0:
 print(output["messages"][-1].content[:300] + "...")

# Example Execution
if __name__ == "__main__":
 test_request = (
 "Research the latest rate limit policies for the GitHub REST API in 2026, "
 "and write an asynchronous Python client implementing exponential backoff with jitter."
 )
 asyncio.run(run_swarm_task(test_request, session_id="thread_gh_sync_001"))