Agent Orchestration with LangGraph in Python

By Khaled Md Tuhidul Hossain · 2024-10-21 · AI Engineering

As AI agents grow more complex, structured orchestration becomes necessary. Simple linear chains break down when agents need to make decisions, loop back, or coordinate with other agents. This is the problem LangGraph solves. Built by the LangChain team, LangGraph provides a framework for constructing stateful, multi-step agent workflows as directed graphs.

Unlike traditional chain-based approaches where execution flows in a straight line, LangGraph lets you define nodes (units of work), edges (transitions between them), and conditional routing (decision points that alter the flow). The result is an agent architecture that can handle complex, branching logic while maintaining state across every step.

Why LangGraph?

If you have used LangChain's AgentExecutor, you have probably hit its limitations. It works well for simple tool-calling loops but gets unwieldy when you need custom control flow, human-in-the-loop interactions, or multi-agent coordination. LangGraph addresses this by giving you explicit control over every transition in your agent's execution graph.

Key advantages:

Stateful execution: The graph maintains a typed state object that persists and evolves across nodes.
Conditional branching: Route execution based on the current state, LLM output, or any custom logic.
Cycles: Unlike DAGs, LangGraph supports cycles — essential for agent loops that retry or iterate.
Checkpointing: Built-in support for persisting state, enabling pause/resume and debugging.

Installation

pip install langgraph langchain langchain-openai

import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

Core concepts

State

Every LangGraph workflow starts with a state definition. The state is a TypedDict that flows through the graph, accumulating information as each node processes it:

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    next_action: str
    iteration_count: int

The Annotated type with operator.add tells LangGraph how to merge state updates. For the messages field, new messages are appended to the existing list rather than replacing it.

Nodes

Nodes are Python functions that receive the current state, perform some work, and return state updates:

from langchain_core.messages import HumanMessage, AIMessage

def research_node(state: AgentState) -> dict:
    """Simulates a research step."""
    messages = state["messages"]
    last_message = messages[-1].content

    # In a real application, this would call an LLM or search tool
    research_result = f"Research findings for: {last_message}"

    return {
        "messages": [AIMessage(content=research_result)],
        "next_action": "analyze",
    }

def analysis_node(state: AgentState) -> dict:
    """Analyzes research findings."""
    messages = state["messages"]
    research = messages[-1].content

    analysis = f"Analysis of: {research}"

    return {
        "messages": [AIMessage(content=analysis)],
        "next_action": "respond",
    }

Edges and conditional routing

Edges define how nodes connect. Conditional edges use a function to determine the next node based on the current state:

def route_decision(state: AgentState) -> str:
    """Determine the next node based on state."""
    return state.get("next_action", "end")

Building a complete agent with tool calling

Let us build a practical agent that uses an LLM with tools and conditional routing:

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool
import ast

@tool
def search_web(query: str) -> str:
    """Search the web for information on a given query."""
    # Simulated search results
    results = {
        "python trends 2024": "Python remains the top language for AI/ML...",
        "latest ai research": "Transformer architectures continue to dominate...",
    }
    for key, value in results.items():
        if key in query.lower():
            return value
    return f"Search results for: {query}"

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression safely."""
    try:
        # Use ast.literal_eval for safe evaluation of simple expressions
        tree = ast.parse(expression, mode="eval")
        result = compile(tree, "<string>", "eval")
        value = __builtins__["eval"](result)  # noqa: S307
        return f"{expression} = {value}"
    except Exception as e:
        return f"Error evaluating expression: {e}"

tools = [search_web, calculate]
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)

Now define the graph nodes:

from langchain_core.messages import ToolMessage
import json

def agent_node(state: AgentState) -> dict:
    """The main agent node that decides what to do."""
    messages = state["messages"]
    system_message = SystemMessage(
        content="You are a helpful assistant. Use your tools when needed."
    )
    response = llm_with_tools.invoke([system_message] + list(messages))
    return {"messages": [response], "iteration_count": state.get("iteration_count", 0)}

def tool_executor_node(state: AgentState) -> dict:
    """Executes tool calls from the agent's response."""
    messages = state["messages"]
    last_message = messages[-1]

    tool_results = []
    tool_map = {t.name: t for t in tools}

    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]

        if tool_name in tool_map:
            result = tool_map[tool_name].invoke(tool_args)
            tool_results.append(
                ToolMessage(
                    content=str(result),
                    tool_call_id=tool_call["id"],
                )
            )

    return {
        "messages": tool_results,
        "iteration_count": state.get("iteration_count", 0) + 1,
    }

The conditional routing function checks whether the agent wants to call tools or is ready to respond:

def should_continue(state: AgentState) -> str:
    """Determine whether to continue with tools or finish."""
    messages = state["messages"]
    last_message = messages[-1]

    # Safety check: prevent infinite loops
    if state.get("iteration_count", 0) >= 5:
        return "end"

    # If the LLM made tool calls, execute them
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"

    return "end"

Assembling the graph

Now wire everything together:

workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_executor_node)

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

# Add conditional edges from the agent node
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        "end": END,
    },
)

# After tool execution, always go back to the agent
workflow.add_edge("tools", "agent")

# Compile the graph
app = workflow.compile()

The graph structure is: Agent -> (tools needed?) -> Tool Executor -> Agent -> (tools needed?) -> ... -> END. This loop continues until the agent produces a response without tool calls or hits the iteration limit.

Running the agent

from langchain_core.messages import HumanMessage

initial_state = {
    "messages": [HumanMessage(content="What are the latest Python trends in 2024?")],
    "next_action": "",
    "iteration_count": 0,
}

result = app.invoke(initial_state)

for message in result["messages"]:
    role = message.__class__.__name__.replace("Message", "")
    print(f"{role}: {message.content[:200]}")
    print()

Adding checkpointing for persistence

LangGraph supports checkpointing, which saves the state at each step. This is invaluable for debugging and for building workflows that can be paused and resumed:

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "conversation-1"}}

result = app.invoke(
    {
        "messages": [HumanMessage(content="Search for latest AI research")],
        "next_action": "",
        "iteration_count": 0,
    },
    config=config,
)

With a thread ID, you can resume conversations or inspect the state at any checkpoint. In production, you would swap MemorySaver for a persistent backend like SQLite or PostgreSQL.

Visualizing the graph

LangGraph can generate a visual representation of your workflow, which is extremely helpful for understanding and documenting complex agent architectures:

from IPython.display import Image, display

display(Image(app.get_graph().draw_mermaid_png()))

This produces a flowchart showing all nodes, edges, and conditional branches in your graph.

Practical design patterns

A few patterns I have found effective when building with LangGraph:

Guard nodes: Add validation nodes before expensive operations to catch bad inputs early.
Iteration limits: Always include a maximum iteration count in your conditional routing to prevent runaway loops.
State accumulation: Use the operator.add annotation for list fields so that information accumulates rather than getting overwritten.
Modular subgraphs: For complex workflows, build and test subgraphs independently before composing them into a larger system.

Conclusion

LangGraph brings graph-based orchestration to AI agent development. By making state, transitions, and decision points explicit, it gives you control and visibility that chain-based approaches cannot match. The learning curve is steeper than a simple agent executor, but the payoff in reliability and debuggability is real. For any agent workflow that goes beyond a single tool-calling loop, LangGraph is the framework I reach for first.