0Pricing
AI Engineering Academy · Lesson

Building Multi-Agent Pipelines with LangGraph

Use LangGraph to define agent nodes, conditional edges, and shared state in a directed graph, enabling complex multi-agent workflows with loops, branching, and human-in-the-loop checkpoints.

Building Multi-Agent Pipelines with LangGraph is a free AI Engineering Academy lesson on CoddyKit — lesson 3 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Engineering Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

What Is LangGraph?

LangGraph is a library built on top of LangChain that lets you define multi-agent workflows as directed graphs. Nodes in the graph represent agents or processing steps, edges represent the flow of state between them, and conditional edges allow dynamic routing based on the current state. LangGraph handles the execution engine, state persistence, and human-in-the-loop checkpoints.

# pip install langgraph
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# Define your shared state schema
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # append-only list
    research: str
    draft: str
    status: str

Defining State in LangGraph

Every LangGraph workflow operates on a shared state object that all nodes can read and write. The state is a typed dictionary (TypedDict) that carries data through the graph. When a node runs, it receives the current state, performs its work, and returns a dictionary of updates to merge into the state. This shared-state model is what enables agents to communicate without direct coupling.

from typing import TypedDict

class ResearchState(TypedDict):
    query: str           # input from user
    research_notes: str  # filled by researcher node
    written_draft: str   # filled by writer node
    review_feedback: str # filled by reviewer node
    final_output: str    # filled by synthesizer node
    iteration_count: int # tracks how many revision loops occurred

# Each node returns a PARTIAL update - only the keys it modifies
def researcher_node(state: ResearchState) -> dict:
    notes = do_research(state['query'])
    return {'research_notes': notes}  # only update this key

Creating Agent Nodes

In LangGraph, each agent is a node function that takes the current state, performs its LLM call and tool executions, and returns a state update. Nodes are pure functions — they do not store internal state. All state lives in the shared graph state object, making the workflow easy to inspect, resume, and debug.

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

llm = ChatOpenAI(model='gpt-4o')

def researcher_node(state: ResearchState) -> dict:
    messages = [
        SystemMessage(content='You are a research specialist. Find accurate information.'),
        HumanMessage(content=f'Research this topic: {state["query"]}')
    ]
    response = llm.invoke(messages)
    return {'research_notes': response.content}

def writer_node(state: ResearchState) -> dict:
    messages = [
        SystemMessage(content='You are a technical writer. Write clear, engaging prose.'),
        HumanMessage(content=f'Write a draft using these notes:\n{state["research_notes"]}')
    ]
    response = llm.invoke(messages)
    return {'written_draft': response.content}

Building the Graph with StateGraph

After defining node functions, you wire them together with StateGraph. You add nodes with graph.add_node(), add edges between them with graph.add_edge(), set the entry point with graph.set_entry_point(), and compile the graph into an executable with graph.compile(). The compiled graph is a runnable that accepts an initial state and returns the final state.

from langgraph.graph import StateGraph, END

# Build the graph
workflow = StateGraph(ResearchState)

# Add nodes
workflow.add_node('researcher', researcher_node)
workflow.add_node('writer', writer_node)
workflow.add_node('reviewer', reviewer_node)

# Add edges (sequential pipeline)
workflow.set_entry_point('researcher')
workflow.add_edge('researcher', 'writer')
workflow.add_edge('writer', 'reviewer')
workflow.add_edge('reviewer', END)

# Compile into an executable
app = workflow.compile()

# Run it
result = app.invoke({'query': 'What is RAG?', 'iteration_count': 0})
print(result['written_draft'])

Conditional Edges for Dynamic Routing

Conditional edges allow the graph to route to different nodes based on the current state. Instead of a fixed edge, you provide a routing function that inspects the state and returns the name of the next node. This enables revision loops, quality gates, human approval steps, and branching based on the content of agent outputs.

def should_revise(state: ResearchState) -> str:
    '''Router function: returns the name of the next node.'''
    if state['iteration_count'] >= 3:
        return 'finalize'  # Too many revisions - accept as is
    if 'insufficient' in state.get('review_feedback', '').lower():
        return 'researcher'  # Need more research
    if 'rewrite' in state.get('review_feedback', '').lower():
        return 'writer'  # Needs rewriting
    return 'finalize'  # Looks good

# Add conditional edge from reviewer
workflow.add_conditional_edges(
    'reviewer',                    # from node
    should_revise,                 # routing function
    {
        'researcher': 'researcher', # route name -> node name
        'writer': 'writer',
        'finalize': 'finalizer'
    }
)

Loops and Iteration in LangGraph

LangGraph natively supports loops — an agent can be revisited multiple times. This is essential for revision cycles, retry-on-failure patterns, and iterative refinement. Always include a loop termination condition in your state (such as an iteration counter or a quality score threshold) and enforce it in your conditional edge router to prevent infinite loops.

def researcher_node(state: ResearchState) -> dict:
    notes = do_research(state['query'])
    return {
        'research_notes': notes,
        'iteration_count': state['iteration_count'] + 1  # always increment
    }

def should_continue_research(state: ResearchState) -> str:
    # Terminate loop after 3 iterations regardless of quality
    if state['iteration_count'] >= 3:
        return END
    # Continue if research is incomplete
    if len(state.get('research_notes', '')) < 500:
        return 'researcher'  # loop back
    return 'writer'  # proceed to next stage

Parallel Node Execution

LangGraph supports parallel branches using RunnableParallel within a node or by fanning out to multiple nodes. When independent subtasks need to run simultaneously, you can create a fan-out from one node to multiple parallel nodes, then a fan-in node that waits for all of them and merges their results into the shared state.

# Fan-out: one node triggers multiple parallel ones
workflow.add_edge('planner', 'researcher_a')
workflow.add_edge('planner', 'researcher_b')
workflow.add_edge('planner', 'researcher_c')

# Fan-in: aggregator waits for all three
workflow.add_edge('researcher_a', 'aggregator')
workflow.add_edge('researcher_b', 'aggregator')
workflow.add_edge('researcher_c', 'aggregator')
workflow.add_edge('aggregator', 'writer')

# Aggregator merges parallel results
def aggregator_node(state: ResearchState) -> dict:
    combined = state.get('notes_a', '') + '\n' + state.get('notes_b', '') + '\n' + state.get('notes_c', '')
    return {'research_notes': combined}

Checkpointing and Human-in-the-Loop

LangGraph supports checkpointing by integrating with a checkpointer backend (SQLite, Redis, or PostgreSQL). With checkpointing enabled, the graph state is persisted after each node execution. This allows long-running workflows to be paused, inspected, and resumed. It also enables human-in-the-loop patterns where the graph pauses at a specific node and waits for a human to approve or provide input before continuing.

from langgraph.checkpoint.sqlite import SqliteSaver

# Use SQLite for persistent checkpoints
checkpointer = SqliteSaver.from_conn_string(':memory:')
app = workflow.compile(checkpointer=checkpointer, interrupt_before=['human_review'])

# First run - pauses at human_review node
thread = {'configurable': {'thread_id': 'my-workflow-1'}}
result = app.invoke({'query': 'Analyze competitors'}, config=thread)
# result.next == 'human_review' -- waiting for human input

# Human provides feedback and resumes
app.update_state(thread, {'review_feedback': 'Good research, proceed with writing'})
final = app.invoke(None, config=thread)  # resume from checkpoint

Streaming LangGraph Outputs

LangGraph supports streaming mode that emits intermediate state updates as each node completes, rather than waiting for the entire workflow to finish. This is valuable for long pipelines where you want to display partial progress to the user. Use app.stream() to get an iterator of state snapshots from each step.

# Stream intermediate results as each node completes
for event in app.stream({'query': 'What is RAG?'}):
    for node_name, node_output in event.items():
        print(f'Node completed: {node_name}')
        if 'research_notes' in node_output:
            print('Research done:', node_output['research_notes'][:100])
        if 'written_draft' in node_output:
            print('Draft done:', node_output['written_draft'][:100])

Visualizing the Graph

LangGraph can draw the graph as a Mermaid diagram, which is invaluable for understanding complex workflows with many nodes and conditional edges. Call app.get_graph().draw_mermaid_png() to get a PNG image, or app.get_graph().draw_mermaid() for the Mermaid syntax you can paste into any Mermaid renderer.

# Visualize the workflow graph
graph_image = app.get_graph().draw_mermaid_png()
with open('workflow.png', 'wb') as f:
    f.write(graph_image)

# Or print Mermaid syntax
print(app.get_graph().draw_mermaid())
# Outputs:
# graph TD
#    __start__ --> researcher
#    researcher --> writer
#    writer --> reviewer
#    reviewer -->|Good| finalize
#    reviewer -->|Needs work| researcher

End-to-End LangGraph Agent Example

Putting it all together: a complete LangGraph multi-agent pipeline defines a TypedDict state, creates node functions for each agent, wires them with StateGraph, adds conditional edges for revision loops, compiles with a checkpointer, and invokes with an initial state. This gives you a robust, observable, resumable multi-agent workflow with minimal boilerplate.

Quick Check

Test your understanding of LangGraph multi-agent pipelines from this lesson.

Lesson Recap

In this lesson you learned: LangGraph represents multi-agent workflows as directed graphs where nodes are agents and edges are flow, conditional edges enable dynamic routing, loops, and revision cycles based on the current state, and checkpointing allows long-running workflows to pause and resume with human-in-the-loop approval. Next up we explore shared memory and inter-agent communication.

Frequently asked questions

Is the “Building Multi-Agent Pipelines with LangGraph” lesson free?

Yes — the full text of “Building Multi-Agent Pipelines with LangGraph” is free to read here on the web, and the AI Engineering Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Engineering Academy course, upgrade to CoddyKit PRO.

What will I learn in “Building Multi-Agent Pipelines with LangGraph”?

Use LangGraph to define agent nodes, conditional edges, and shared state in a directed graph, enabling complex multi-agent workflows with loops, branching, and human-in-the-loop checkpoints. You practise AI Engineering Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start AI Engineering Academy?

No prior experience is required. AI Engineering Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 3 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Building Multi-Agent Pipelines with LangGraph” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this AI Engineering Academy lesson?

Yes. Every AI Engineering Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

  1. Why Single Agents Hit a Wall
  2. Orchestrator-Subagent Pattern
  3. Building Multi-Agent Pipelines with LangGraph
  4. Shared Memory and Inter-Agent Communication
← Back to AI Engineering Academy