Stateful Multi-Agent Application

Single agents can struggle to handle multiple tasks like auditing documentation and planning schedules. In this page, we walk through building a stateful, collaborative multi-agent application utilizing a custom LangGraph configuration and memory checkpointers.

1. Core LangGraph Concepts

LangGraph models agents as state machines. There are four fundamental concepts:

State: The central database or memory structure shared across the graph. Any node can read variables from this state and output key-value updates to write back to it.
Nodes: Python functions representing independent operations or agents. A node receives the current state, processes it (e.g., calls an LLM or runs a tool), and returns a dictionary with state updates.
Edges: Control flow rules. Normal Edges define a fixed sequential route (e.g. from node A to node B). Conditional Edges use a router function to dynamically decide which node to visit next based on state data.
Checkpointers (Memory): Databases that automatically save a snapshot of the graph’s state after every node execution. This allows the graph to resume or remember conversations across separate API requests.

2. Available APIs & Key Classes

To build stateful agents, LangGraph provides the following key classes and decorators:

StateGraph: The primary builder class used to construct graphs. It takes a typed dictionary (schema) defining the structure of the state.
from langgraph.graph import StateGraph builder = StateGraph(StateSchemaClass)
add_messages: An accumulator reducer function. In LangGraph, returning a key updates the state. For lists of messages, we don’t want to overwrite previous history. Using Annotated[list, add_messages] tells the graph to append new messages instead of replacing the list.

MemorySaver: A built-in, in-memory checkpointer class that preserves thread states.

from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

thread_id: A config key passed during graph execution. Any queries sharing the same thread_id share the same conversation history in the checkpointer.

3. Practical Step-by-Step Implementation

Step 1: Define the Graph State Schema

We define a Python class inheriting from TypedDict containing the keys that will be shared between our agents:

from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # Holds message history for conversational query turns
    messages: Annotated[list[BaseMessage], add_messages]
    # Raw resume content passed into the pipeline
    resume_text: str
    # Evaluation findings from the Resume Analyzer Agent
    analysis_notes: str
    # Plan generated by the Interview Planner Agent
    interview_plan: str

Step 2: Create a Verification Tool

We write a tool that the analyzer agent can invoke to cross-check candidate skills:

from langchain_core.tools import tool

@tool
def check_technical_keywords(skills: list[str]) -> str:
    """Verifies candidate skills against standard technical definitions and requirements."""
    verified = []
    for skill in skills:
        cleaned = skill.lower().strip()
        if cleaned in ["python", "langchain", "fastapi", "react", "sql", "docker"]:
            verified.append(f"{skill} (Industry Standard)")
        else:
            verified.append(f"{skill} (Alternative/General)")
    return "Verified Skills: " + ", ".join(verified)

Step 3: Define Agent Node Functions

Nodes receive the shared state, call the LLM, and return updates:

from langchain_core.messages import AIMessage, HumanMessage
from langchain.chat_models import init_chat_model

llm = init_chat_model("gemini-2.5-flash", model_provider="google_genai")

def resume_analyzer_node(state: AgentState):
    """Resume Analyzer Agent: Audits candidate skills and projects, verifying technical terms."""
    resume = state.get("resume_text", "")
    
    prompt = [
        HumanMessage(content=f"Analyze this candidate's resume text. Extract key skills, total experience, and notable projects. Here is the resume:\n\n{resume}")
    ]
    
    # Bind the keyword verification tool
    llm_with_tools = llm.bind_tools([check_technical_keywords])
    response = llm_with_tools.invoke(prompt)
    
    tool_notes = ""
    if response.tool_calls:
        tool_call = response.tool_calls[0]
        tool_result = check_technical_keywords.invoke(tool_call["args"])
        tool_notes = f"\nTool Output: {tool_result}"
    
    analysis = f"--- Profile Analysis ---\n{response.content}\n{tool_notes}"
    return {
        "messages": [AIMessage(content="I have completed analyzing the candidate's profile.")],
        "analysis_notes": analysis
    }

def interview_planner_node(state: AgentState):
    """Interview Planner Agent: Takes candidate analysis notes and creates a custom plan."""
    analysis = state.get("analysis_notes", "")
    
    prompt = [
        HumanMessage(content=f"You are a technical interviewer. Design a customized interview plan including 2 technical questions and 1 behavioral question tailored specifically to the candidate's profile below:\n\n{analysis}")
    ]
    
    response = llm.invoke(prompt)
    return {
        "messages": [response],
        "interview_plan": response.content
    }

Step 4: Assemble Graph Nodes & Edges

We add nodes and map the route sequentially:

from langgraph.graph import StateGraph, START, END

builder = StateGraph(AgentState)

# Register nodes
builder.add_node("analyzer", resume_analyzer_node)
builder.add_node("planner", interview_planner_node)

# Set up edges
builder.add_edge(START, "analyzer")
builder.add_edge("analyzer", "planner")
builder.add_edge("planner", END)

Step 5: Compile with Checkpointer Memory

We attach MemorySaver to compile our final application graph:

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
hr_agent_app = builder.compile(checkpointer=memory)

4. Combined Running Code Project

Here is the complete, self-contained Python script ready to copy and run:

import os
from typing import Annotated, TypedDict
from dotenv import load_dotenv
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.tools import tool
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

# 1. Load keys
load_dotenv()

# 2. Define State Schema
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    resume_text: str
    analysis_notes: str
    interview_plan: str

# 3. Define Verification Tool
@tool
def check_technical_keywords(skills: list[str]) -> str:
    """Verifies candidate skills against standard technical definitions and requirements."""
    verified = []
    for skill in skills:
        cleaned = skill.lower().strip()
        if cleaned in ["python", "langchain", "fastapi", "react", "sql", "docker"]:
            verified.append(f"{skill} (Industry Standard)")
        else:
            verified.append(f"{skill} (Alternative/General)")
    return "Verified Skills: " + ", ".join(verified)

# 4. Initialize LLM
llm = init_chat_model("gemini-2.5-flash", model_provider="google_genai")

# 5. Define Nodes
def resume_analyzer_node(state: AgentState):
    print("\n[Node: Resume Analyzer] Evaluating resume profile...")
    resume = state.get("resume_text", "")
    prompt = [
        HumanMessage(content=f"Analyze this candidate's resume text. Extract key skills, total experience, and notable projects. Here is the resume:\n\n{resume}")
    ]
    llm_with_tools = llm.bind_tools([check_technical_keywords])
    response = llm_with_tools.invoke(prompt)
    
    tool_notes = ""
    if response.tool_calls:
        tool_call = response.tool_calls[0]
        tool_result = check_technical_keywords.invoke(tool_call["args"])
        tool_notes = f"\nTool Output: {tool_result}"
    
    analysis = f"--- Profile Analysis ---\n{response.content}\n{tool_notes}"
    print(" -> Analysis Complete.")
    return {
        "messages": [AIMessage(content="I have completed analyzing the candidate's profile.")],
        "analysis_notes": analysis
    }

def interview_planner_node(state: AgentState):
    print("\n[Node: Interview Planner] Generating custom technical interview questions...")
    analysis = state.get("analysis_notes", "")
    prompt = [
        HumanMessage(content=f"You are a technical interviewer. Design a customized interview plan including 2 technical questions and 1 behavioral question tailored specifically to the candidate's profile below:\n\n{analysis}")
    ]
    response = llm.invoke(prompt)
    print(" -> Interview Plan Generated.")
    return {
        "messages": [response],
        "interview_plan": response.content
    }

# 6. Assemble Graph
builder = StateGraph(AgentState)
builder.add_node("analyzer", resume_analyzer_node)
builder.add_node("planner", interview_planner_node)
builder.add_edge(START, "analyzer")
builder.add_edge("analyzer", "planner")
builder.add_edge("planner", END)

# 7. Compile with Memory
memory = MemorySaver()
hr_agent_app = builder.compile(checkpointer=memory)

# 8. Run
if __name__ == "__main__":
    config = {"configurable": {"thread_id": "candidate-assessment-john-doe"}}
    
    sample_resume = """
    John Doe
    Python Developer - 3 Years Experience
    Skills: Python, LangChain, FastAPI, TensorFlow
    Projects:
    - Built a RAG chatbot using LangChain and FastAPI to query customer support documents.
    """
    
    print("=== Start Candidate Assessment ===")
    
    # Turn 1
    final_state = hr_agent_app.invoke(
        {
            "messages": [HumanMessage(content="Evaluate John Doe's resume and generate an interview plan.")],
            "resume_text": sample_resume
        },
        config
    )
    print("\n=== Tailored Interview Plan ===")
    print(final_state["interview_plan"])
    
    # Turn 2: Follow-up conversational check
    follow_up = "What technical questions did you generate? Remind me."
    print(f"\nUser: {follow_up}")
    
    follow_up_state = hr_agent_app.invoke(
        {"messages": [HumanMessage(content=follow_up)]},
        config
    )
    print("\n=== Agent Follow-Up Response ===")
    print(follow_up_state["messages"][-1].content)

5. Practice Exercises

Practice 1: Adding a Feedback Evaluation Node

Extend the stateful graph by adding a third node called feedback_evaluator that runs after the planner node. The node should format a short summary checklist of topics to check off during the interview. Instructions:

Update AgentState to include an eval_checklist key.
Define a function feedback_evaluator_node(state: AgentState) that takes the interview_plan and creates a checklist.
Add the node to the graph and update the edges: planner $\rightarrow$ evaluator $\rightarrow$ END.

Solution

# 1. Update state schema
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    resume_text: str
    analysis_notes: str
    interview_plan: str
    eval_checklist: str # New Key

# 2. Define Evaluator Node
def feedback_evaluator_node(state: AgentState):
    print("\n[Node: Evaluator] Generating evaluation checklist...")
    plan = state.get("interview_plan", "")
    prompt = [
        HumanMessage(content=f"Create a short checklist of skills to grade based on this plan:\n\n{plan}")
    ]
    response = llm.invoke(prompt)
    return {
        "messages": [response],
        "eval_checklist": response.content
    }

# 3. Add to Graph Builder
builder = StateGraph(AgentState)
builder.add_node("analyzer", resume_analyzer_node)
builder.add_node("planner", interview_planner_node)
builder.add_node("evaluator", feedback_evaluator_node) # New Node

builder.add_edge(START, "analyzer")
builder.add_edge("analyzer", "planner")
builder.add_edge("planner", "evaluator") # Updated Edge
builder.add_edge("evaluator", END) # New Edge

# Compile
hr_agent_app = builder.compile(checkpointer=MemorySaver())

Foundations

Agentic Implementation

Stateful Multi-Agent Application

1. Core LangGraph Concepts

2. Available APIs & Key Classes

3. Practical Step-by-Step Implementation

Step 1: Define the Graph State Schema

Step 2: Create a Verification Tool

Step 3: Define Agent Node Functions

Step 4: Assemble Graph Nodes & Edges

Step 5: Compile with Checkpointer Memory

4. Combined Running Code Project

5. Practice Exercises

Practice 1: Adding a Feedback Evaluation Node

​1. Core LangGraph Concepts

​2. Available APIs & Key Classes

​3. Practical Step-by-Step Implementation

​Step 1: Define the Graph State Schema

​Step 2: Create a Verification Tool

​Step 3: Define Agent Node Functions

​Step 4: Assemble Graph Nodes & Edges

​Step 5: Compile with Checkpointer Memory

​4. Combined Running Code Project

​5. Practice Exercises

​Practice 1: Adding a Feedback Evaluation Node

1. Core LangGraph Concepts

2. Available APIs & Key Classes

3. Practical Step-by-Step Implementation

Step 1: Define the Graph State Schema

Step 2: Create a Verification Tool

Step 3: Define Agent Node Functions

Step 4: Assemble Graph Nodes & Edges

Step 5: Compile with Checkpointer Memory

4. Combined Running Code Project

5. Practice Exercises

Practice 1: Adding a Feedback Evaluation Node