Creating a Multi-Agent System with LangGraph

LangGraph extends LangChain with powerful state management and cyclic graph capabilities, enabling the creation of sophisticated multi-agent systems. In this tutorial, we'll build a collaborative multi-agent system where specialized agents work together to solve complex tasks.

Understanding Multi-Agent Systems

Multi-agent systems distribute cognitive labor across specialized agents, each with unique capabilities. This approach has several advantages:

• Specialized expertise: Agents can focus on specific tasks
• Workflow control: Complex decision logic determines agent interactions
• Emergent capabilities: The system can solve problems no single agent could handle

LangGraph provides the infrastructure to coordinate these agents through state management and cyclic execution flows.

Building a Research Team of Agents

Let's build a research team consisting of three specialized agents:

1. Researcher: Gathers information and generates research questions
2. Planner: Coordinates the research process and delegates tasks
3. Writer: Synthesizes findings into coherent documentation

First, let's set up our imports and define the state interface:

from typing import TypedDict, List, Union, Literal, Annotated
from typing_extensions import TypedDict

from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, AIMessage, SystemMessage
from langgraph.graph import StateGraph, END

# Define our state structure
class AgentState(TypedDict):
    messages: List[Union[HumanMessage, AIMessage, SystemMessage]]
    next: str
    context: dict
    research_summary: str

Next, let's define our specialized agents:

# Create the LLM
llm = ChatOpenAI(temperature=0, model="gpt-4")

# Researcher Agent
def researcher_agent(state):
    messages = state["messages"]
    context = state.get("context", {})
    
    # Add system message for researcher role
    system_message = SystemMessage(content="""
    You are a research agent. Your job is to gather information on the given topic.
    Generate insightful research questions and provide detailed findings.
    Focus on accuracy and depth of information.
    """)
    
    # Process the request
    response = llm.invoke([system_message] + messages)
    
    # Update the state
    return {
        "messages": messages + [response],
        "next": "planner",  # Always go to planner next
        "context": context,
        "research_summary": state.get("research_summary", "")
    }

# Planner Agent
def planner_agent(state):
    messages = state["messages"]
    context = state.get("context", {})
    research_summary = state.get("research_summary", "")
    
    # Add system message for planner role
    system_message = SystemMessage(content="""
    You are a planning agent. Your job is to coordinate the research process.
    Review the current research, identify gaps, and decide the next steps.
    You can:
    1. Send the task back to the researcher for more information (next: "researcher")
    2. Forward to the writer to document findings (next: "writer")
    3. End the process if the research is complete (next: "END")
    
    Always specify your decision in the format: NEXT: researcher/writer/END
    """)
    
    # Get the planner's decision
    response = llm.invoke([system_message] + messages)
    
    # Extract the next step from the response
    response_text = response.content
    if "NEXT: researcher" in response_text:
        next_step = "researcher"
    elif "NEXT: writer" in response_text:
        next_step = "writer"
    elif "NEXT: END" in response_text:
        next_step = "END"
    else:
        # Default to researcher if no clear directive
        next_step = "researcher"
    
    return {
        "messages": messages + [response],
        "next": next_step,
        "context": context,
        "research_summary": research_summary
    }

# Writer Agent
def writer_agent(state):
    messages = state["messages"]
    context = state.get("context", {})
    
    # Add system message for writer role
    system_message = SystemMessage(content="""
    You are a writing agent. Your job is to synthesize research findings into clear, 
    well-organized documentation.
    Create a comprehensive summary that covers all key points.
    Be concise but thorough.
    """)
    
    # Generate the documentation
    response = llm.invoke([system_message] + messages)
    
    # Create a research summary
    summary = response.content
    
    return {
        "messages": messages + [response],
        "next": "planner",  # Return to planner for next decision
        "context": context,
        "research_summary": summary
    }

Now, let's create the graph structure to coordinate these agents:

# Create the graph
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("researcher", researcher_agent)
workflow.add_node("planner", planner_agent)
workflow.add_node("writer", writer_agent)

# Add edges
workflow.add_edge("researcher", "planner")
workflow.add_edge("writer", "planner")

# Add conditional edges from planner to other nodes
workflow.add_conditional_edges(
    "planner",
    lambda x: x["next"],
    {
        "researcher": "researcher",
        "writer": "writer",
        "END": END
    }
)

# Set the entry point
workflow.set_entry_point("researcher")

# Compile the graph
graph = workflow.compile()

Running the Multi-Agent System

With our graph set up, we can now run a research task through our multi-agent system:

# Initialize the state
initial_state = {
    "messages": [HumanMessage(content=
        "Research the impact of large language models on education. "
        "Focus on both benefits and potential concerns.")],
    "next": "",
    "context": {},
    "research_summary": ""
}

# Run the workflow
result = graph.invoke(initial_state)

# Extract the final research summary
final_summary = result["research_summary"]
print(f"Research Summary:\n{final_summary}")

Visualizing the Agent Workflow

One of the benefits of LangGraph is the ability to visualize the agent workflow. You can generate a visual representation of your graph with:

from IPython.display import display, Image
from langgraph.graph import get_graph_bytes

# Get the visualization
visualization = get_graph_bytes(workflow)

# Display the image in a notebook
display(Image(visualization))

Tracking State Transitions

LangGraph allows you to trace the execution of your multi-agent system, showing how the state evolves through each step:

# Create a trace for debugging
result_with_trace = graph.invoke(
    initial_state,
    {"recursion_limit": 25},  # Prevent infinite loops
    return_trace=True
)

# View the execution path
for step in result_with_trace.keys():
    print(f"Step: {step}")

# Examine a specific step
step_data = result_with_trace[list(result_with_trace.keys())[2]]
print(f"Agent: {step_data['node']}")
print(f"Input: {step_data['input']}")
print(f"Output: {step_data['output']}")

Advanced Patterns

Adding Memory to Agents

You can enhance the system by adding memory to agents:

from langchain.memory import ConversationBufferMemory

# Update the state interface
class AgentStateWithMemory(TypedDict):
    messages: List[Union[HumanMessage, AIMessage, SystemMessage]]
    next: str
    context: dict
    research_summary: str
    memory: dict

# Initialize memory in the agent functions
def researcher_with_memory(state):
    memory = state.get("memory", {}).get("researcher", ConversationBufferMemory())
    # Use memory to provide context
    memory_context = memory.load_memory_variables({})
    # ... rest of the agent logic
    
    # Update memory
    memory.save_context({"input": input_text}, {"output": response.content})
    
    # Return updated state with memory
    return {
        # ... other state updates
        "memory": {**state.get("memory", {}), "researcher": memory}
    }

Parallel Agent Execution

For more complex workflows, you can implement parallel agent execution:

# Define parallel execution branches
def branch_selector(state):
    task = state["messages"][-1].content.lower()
    if "technical" in task:
        return ["technical_researcher", "technical_writer"]
    else:
        return ["general_researcher", "general_writer"]

# Add branching logic
workflow.add_conditional_edges(
    "planner",
    branch_selector,
    {
        "technical_researcher, technical_writer": ["technical_researcher", "technical_writer"],
        "general_researcher, general_writer": ["general_researcher", "general_writer"]
    }
)

# Add aggregation node to combine parallel work
def results_aggregator(state):
    # Combine results from parallel branches
    # ... aggregation logic
    return {
        # ... updated state
    }

workflow.add_node("aggregator", results_aggregator)
workflow.add_edge("technical_writer", "aggregator") 
workflow.add_edge("general_writer", "aggregator")

Best Practices for Multi-Agent Systems

1. Clear agent roles: Define specific responsibilities for each agent
2. Robust state management: Ensure state is properly passed between agents
3. Error handling: Implement fallback strategies for agent failures
4. Cycle detection: Prevent infinite loops with recursion limits
5. Observability: Use LangSmith to monitor agent performance

Conclusion

Multi-agent systems powered by LangGraph enable sophisticated workflows where specialized agents collaborate to solve complex problems. By defining clear agent roles, managing state efficiently, and structuring agent interactions, you can create AI systems that achieve outcomes beyond what any single agent could accomplish.

In our next article, we'll explore how to build evaluation systems with LangSmith to assess and improve the performance of your multi-agent systems.