> ## Documentation Index
> Fetch the complete documentation index at: https://genai.codewithsiva.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# Stateful Multi-Agent Application

> Build a stateful Resume Analyzer & Interview Planner agent pipeline with memory checkpointers using LangGraph

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.
  ```python theme={null}
  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.
  ```python theme={null}
  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:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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:**

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

<Accordion title="Solution">
  ```python theme={null}
  # 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())
  ```
</Accordion>