#28 Build QnA Chatbot with Memory using LangGraph & GPT LLM

A fundamental challenge when moving from basic Large Language Model (LLM) scripts to production conversational agents is maintaining contextual state. In a standard stateless request, an LLM treats every interaction as an isolated event. This lecture demonstrates how to build a stateful Question-and-Answer (Q&A) chatbot from the ground up using LangGraph, LangChain, and checkpointer memory mechanisms, avoiding high-level abstractions like pre-built agent factories to understand the underlying mechanics of graph state persistence.

1. From Stateless Pipelines to Stateful Graphs

A naive LLM implementation follows a purely request-response pattern: the user submits a prompt, the model responds, and all internal memory of that exchange is lost once the execution frame exits.

To conduct coherent multi-turn conversations, the system must retain history across calls. Previous patterns relied on manually appending strings to a list or utilizing monolithic wrapper classes. LangGraph models the entire interaction as a State Machine (directed graph), where:

• State flows through predefined computational nodes.
• Execution transitions along directed edges.
• State is automatically persisted at checkpoints across discrete conversation sessions (threads).

2. Core Mental Model: State Reducers & Graph Checkpointing

LangGraph structures conversation flow around three foundational concepts:

The Shared State

The graph relies on a centralized data structure (schema) that is passed to every node. Nodes read from this state, execute their logic, and return updates.

The Reducer Pattern (add_messages)

In standard Python dictionaries or Pydantic models, updating a field overwrites its existing value:

To preserve historical context, LangGraph introduces reducers. A reducer defines how new data combines with existing data. The built-in add_messages function acts as an append-and-merge reducer:

• If a message has a new unique identifier, it is appended to the message list.
• If a message shares an existing identifier, it updates that message in place.

The Checkpointer

Nodes execute logic, but they do not manage persistence. The Checkpointer (InMemorySaver) acts as the graph's storage engine. It captures a snapshot of the updated state at the boundary of every node execution and indexes it by a unique session key known as a thread_id.

3. Environment Setup & Dependency Imports

Building this stateful graph requires integrating LangChain LLM wrappers, LangGraph graph-building blocks, Pydantic for validation, and typing utilities.

import os
from typing import Annotated
from dotenv import load_dotenv
from pydantic import BaseModel

# LangChain Groq LLM integration
from langchain_groq import ChatGroq

# LangGraph core primitives
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver

# Load environment variables (GROQ_API_KEY, OPENAI_API_KEY, etc.)
load_dotenv()

Note: The lecture utilizes ChatGroq serving open models (e.g., openai/gpt-oss-20b or Llama variants hosted via Groq) for low-latency inferences. The identical graph architecture applies if swapping ChatGroq for ChatOpenAI or ChatAnthropic.

4. Defining State Schema with Reducer Annotations

To ensure strict data contracts, LangGraph accepts Pydantic models or TypedDict structures. Using Pydantic's BaseModel provides runtime validation and clean type hinting.

class ChatState(BaseModel):
    # Annotated binds the 'add_messages' reducer function to the list
    messages: Annotated[list, add_messages]

How Annotated[list, add_messages] Operates

1. messages: list declares that this state property holds a list of message objects (HumanMessage, AIMessage, or role-content dictionaries).
2. Annotated[..., add_messages] attaches metadata that LangGraph's engine detects.
3. When any node returns {"messages": [new_message]}, the engine passes the existing list and the new list to add_messages(current_list, update). Instead of overwriting, it yields an expanded list preserving conversation history:

5. Implementing the Chatbot Execution Node

A node in LangGraph is a Python function that takes the current state as input and returns a state update dictionary or a modified state object.

# Initialize LLM backend
llm = ChatGroq(model="openai/gpt-oss-20b")

def chatbot_node(state: ChatState) -> ChatState:
    """
    Executes the conversational step:
    1. Extracts current message history from state.
    2. Invokes LLM with full context.
    3. Updates state messages with the newly generated AI response.
    """
    # Invoke LLM with complete contextual history stored in state
    response = llm.invoke(state.messages)
    
    # Assign the new message wrapped in a list
    # The 'add_messages' reducer will append this response to the history
    state.messages = [response]
    
    return state

Design Pattern Note: Returning state.messages = [response] leverages the reducer to append response to the pre-existing history list automatically. Alternatively, a node can simply return a dictionary: return {"messages": [response]}.

6. Constructing and Compiling the StateGraph

With the state schema and node logic defined, assemble the state graph, register nodes, wire directional edges, and attach the checkpointer.

# 1. Initialize Graph with state schema
graph_builder = StateGraph(ChatState)

# 2. Add computation nodes
graph_builder.add_node("chatbot", chatbot_node)

# 3. Define control flow via directed edges
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

# 4. Instantiate the persistence engine
memory = InMemorySaver()

# 5. Compile graph with checkpointer attached
chatbot_graph = graph_builder.compile(checkpointer=memory)

7. Graph Visualization

LangGraph supports graph introspection, allowing developers to visually audit execution flows and confirm that control edges match architectural designs:

from IPython.display import Image, display

try:
    # Generates a PNG representation of the compiled execution pipeline
    display(Image(chatbot_graph.get_graph().draw_png()))
except Exception:
    # Fallback to ASCII or Mermaid representation if pygraphviz/dependencies are missing
    print(chatbot_graph.get_graph().draw_ascii())

8. The Role of Checkpointers & Resolving the thread_id Error

When invoking a graph compiled with a checkpointer (checkpointer=memory), running an unconfigured call will trigger a runtime failure:

# INCORRECT: Will raise a ValueError
response = chatbot_graph.invoke({
    "messages": [{"role": "user", "content": "Hello!"}]
})

ValueError: Checkpointer requires one or more of the following 'configurable' keys: ['thread_id']

Why Does This Error Occur?

The InMemorySaver checkpointer manages state across concurrent sessions. Without a session identifier, it cannot determine which history to read or update.

To resolve this, every invocation must supply a configuration dictionary containing a unique thread_id:

# CORRECT: Supplying thread configuration
config = {"configurable": {"thread_id": "session_prateek_001"}}

input_data = {
    "messages": [
        {"role": "user", "content": "Hello! I am Prateek Mishra."}
    ]
}

response = chatbot_graph.invoke(input_data, config=config)

Thread Isolation Model

Multiple users can interact with the same compiled graph concurrently. LangGraph maintains strict state boundaries by scoping checkpointer operations to the provided thread_id:

9. Validating Conversational Memory (Multi-Turn Execution)

To verify that state persists across separate execution runs, execute a two-step conversational sequence within the same thread_id.

Turn 1: Providing Personal Context

config = {"configurable": {"thread_id": "session_prateek_001"}}

turn_1_input = {
    "messages": [
        {"role": "user", "content": "Hello, my name is Prateek Mishra."}
    ]
}

turn_1_output = chatbot_graph.invoke(turn_1_input, config=config)
print("AI Response:", turn_1_output["messages"][-1].content)

Output:

AI Response: Hello Prateek Mishra! Nice to meet you. How can I help you today?

Turn 2: Testing Contextual Recall

turn_2_input = {
    "messages": [
        {"role": "user", "content": "What is my name?"}
    ]
}

turn_2_output = chatbot_graph.invoke(turn_2_input, config=config)
print("AI Response:", turn_2_output["messages"][-1].content)

Output:

AI Response: Your name is Prateek Mishra.

Internal State Verification

Inspecting turn_2_output["messages"] reveals the full accumulated message sequence maintained by the add_messages reducer:

1. HumanMessage: "Hello, my name is Prateek Mishra."
2. AIMessage: "Hello Prateek Mishra! Nice to meet you..."
3. HumanMessage: "What is my name?"
4. AIMessage: "Your name is Prateek Mishra."

10. Refactoring into an Interactive Command-Line Interface (CLI)

To turn this prototype into an interactive, production-style terminal application, wrap the compiled graph inside an execution loop with exit conditions and input formatting.

Save the following implementation as langgraph_qna_bot.py:

import os
from typing import Annotated
from dotenv import load_dotenv
from pydantic import BaseModel

from langchain_groq import ChatGroq
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver

# 1. Environment initialization
load_dotenv()

# 2. State schema definition
class ChatState(BaseModel):
    messages: Annotated[list, add_messages]

# 3. Model setup
llm = ChatGroq(model="openai/gpt-oss-20b")

# 4. Node logic
def chatbot_node(state: ChatState) -> ChatState:
    response = llm.invoke(state.messages)
    state.messages = [response]
    return state

# 5. Graph assembly
graph_builder = StateGraph(ChatState)
graph_builder.add_node("chatbot", chatbot_node)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

# 6. Memory compilation
memory = InMemorySaver()
app = graph_builder.compile(checkpointer=memory)

# 7. Interactive execution loop
def run_cli_chatbot():
    print("=" * 60)
    print("LangGraph Memory Q&A Bot Initialized. (Type 'exit' to quit)")
    print("=" * 60)
    
    # Session configuration identifier
    session_config = {"configurable": {"thread_id": "cli_active_session"}}
    
    while True:
        try:
            user_input = input("\nUser: ").strip()
            
            # Termination check
            if user_input.lower() in ["quit", "exit", "bye"]:
                print("\nAI: Thanks for using the assistant. Goodbye!")
                break
            
            if not user_input:
                continue

            # Construct graph input payload
            payload = {
                "messages": [
                    {"role": "user", "content": user_input}
                ]
            }
            
            # Execute graph traversal
            result = app.invoke(payload, config=session_config)
            
            # Retrieve latest AI message from history
            ai_message = result["messages"][-1].content
            print(f"\nAI: {ai_message}")

        except KeyboardInterrupt:
            print("\n\nSession terminated by user.")
            break
        except Exception as e:
            print(f"\n[Error]: {e}")

if __name__ == "__main__":
    run_cli_chatbot()

11. Real-World Practical Pitfalls & Debugging Strategies

When developing graph-based conversational workflows, subtle design and runtime mistakes can lead to lost context or unexpected crashes.

1. Missing Reducer Metadata

# BUGGY DEFINITION:
class BrokenState(BaseModel):
    messages: list  # Missing Annotated[list, add_messages]

• Symptom: The chatbot remembers only the single most recent message.
• Root Cause: Without add_messages, standard Python assignment overwrites the messages list rather than appending incoming messages.
• Fix: Always wrap message sequences with Annotated[list, add_messages].

2. Omitted configurable Parameter

• Symptom: ValueError: Checkpointer requires 'thread_id'.
• Root Cause: A checkpointer requires an active session key to locate and partition stored snapshots.
• Fix: Always supply {"configurable": {"thread_id": "<id>"}} whenever a checkpointer is passed to compile().

3. State Schema Key Mismatch

# State Schema:
class ChatState(BaseModel):
    messages: Annotated[list, add_messages]

# BUGGY INVOCATION:
app.invoke({"query": "Hello!"}, config=config)

• Symptom: KeyError or input ignored during node execution.
• Root Cause: Keys in the input dictionary must match the attribute names declared in the StateGraph schema definition (messages vs query).

12. Summary and Key Takeaways

• Explicit Architecture Over Black Boxes: Instead of relying on monolithic wrapper utilities like create_agent, building state graphs directly exposes how states, reducers, and nodes interact.
• State Reducers Enable History Preservation: The Annotated[list, add_messages] pattern defines how message lists update, merging new interactions into existing history rather than overwriting state.
• Checkpointers Handle Graph Persistence: Attaching InMemorySaver allows graphs to preserve snapshots across individual turns without manual serialization.
• Thread IDs Isolate User Sessions: Checkpointers index state snapshots by thread_id, preventing cross-talk between multiple users while retaining session continuity.
• Nodes Act as State Transformers: Graph nodes receive the current state, run side effects or model inference, and return state updates.
• Extensible Foundation: While this implementation uses a single chatbot node, the same graph structure readily scales to multi-node systems with conditional branching, tool-calling nodes, and human-in-the-loop validation checkpoints.