LangGraph Cheatsheet: Fundamentals and Implementation

LangGraph is a Python library built on top of LangChain for creating complex conversational AI workflows using a graph-based approach.

I. Core LangGraph Elements and Structure

Element	Description
Graph	The overarching structure that maps how different tasks (nodes) are connected and executed, representing the workflow.
State	A shared data structure that holds the current information or context of the entire application (the application’s memory). Nodes access and modify it. The State is typically defined using a `TypedDict`.
Node	An individual Python function or operation that performs a specific task. It receives the current State as input and returns the updated State as output.
Edge	The connection between nodes that determines the flow of execution, specifying which node executes next.
Conditional Edge	A specialized connection that routes execution based on a specific condition or logic applied to the current State. Used for decision-making.
Start/End Point	Virtual entry and conclusion points for the workflow execution.
Tool	Specialized functions or utilities that nodes can utilize to perform specific tasks, enhancing capabilities (e.g., fetching data from an API).
Tool Node	A special type of node whose main job is to run a Tool and connect the tool’s output back into the State.
State Graph	The framework (`StateGraph` class) used to build and compile the graph structure, managing the nodes, edges, and overall state flow.
Reducer Function	A rule that defines how updates from nodes are combined with the existing state. `add_messages` is a reducer function used to append new data without overwriting the state.

II. Essential Python Type Annotations

These concepts are used extensively for defining the State and handling complex data flow:

Annotation	Purpose	Usage Note
TypedDict	Defines the structure of the State as a class, explicitly setting the expected data type for each key (e.g., `name: str`).	Crucial for type safety, reducing runtime errors.
Union	Specifies that a value can be one of several defined data types (e.g., `int` or `float`).	Used extensively in LangGraph/LangChain.
Optional	Specifies that a parameter can be a defined data type or a `None` value.
Lambda	A shortcut for writing small, anonymous functions.	Often used as a pass-through function (`lambda state: state`) in nodes that only return an edge or perform comparison without state assignment.
Annotated	Provides additional context (metadata) to a type without changing its data type (e.g., specifying a `str` must be a “valid email format”).
Sequence	Helps automatically handle state updates for sequential data structures, like chat history, avoiding manual list manipulation.

III. Message Types (For AI Agents)

These types, often inheriting from BaseMessage, are used to store conversation history within the State:

Message Type	Purpose
Human Message	Represents the input from a user (prompt).
AI Message	Represents responses generated by AI models.
System Message	Used to provide fixed instructions or context to the model (e.g., persona definition).
Tool Message	Contains data passed back to the LLM after a tool call (specific to tool usage).
Base Message	The foundational parent class for all message types in LangGraph.

IV. Basic Graph Implementation Patterns

A. Hello World Graph (Single Node, Sequential Flow)

This template shows how to define the state, create a node, and compile the simplest graph structure:

1. Define the Agent State (Schema)

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class AgentState(TypedDict):
    message: str 
# The state needs to be in the form of a TypedDict.

2. Define the Node Function (Action)

Nodes receive the State and return the updated State. Docstrings are important as they inform LLMs what the function does.

def greeting_node(state: AgentState) -> AgentState:
    """Simple node that adds a greeting message to the state""" #
    
    # Access and update the state key
    state["message"] = "Hey " + state["message"] + " how is your day going"
    
    return state # Must return the updated state

3. Build and Compile the Graph

The graph is initialized using StateGraph and the defined state schema.

graph = StateGraph(AgentState) # Pass the state schema
node_name = "greeter" 

# Add the node (name and function/action)
graph.add_node(node_name, greeting_node) 

# Set the entry and finish points to the single node
graph.set_entry_point(node_name) 
graph.set_finish_point(node_name) 

app = graph.compile() # Compile the graph

4. Invoke the Graph

The compiled graph (app) is run using .invoke().

result = app.invoke({"message": "Bob"})
# Output: {'message': 'Hey Bob how is your day going'}

B. Sequential Graph (Multiple Nodes)

Nodes are connected using graph.add_edge().

# Assuming 'first_node' and 'second_node' are defined
graph.add_node("first_node", first_node)
graph.add_node("second_node", second_node)

graph.set_entry_point("first_node")

# Use add_edge to connect the flow directionally
graph.add_edge("first_node", "second_node")

graph.set_finish_point("second_node") 
app = graph.compile()

C. Conditional Graph (Routing)

This pattern uses a routing function to determine the next path, defining branches with add_conditional_edges.

1. Define the Router Function

The router function inspects the state (e.g., operation attribute) and returns the name of the edge to take.

def decide_next_node(state: AgentState) -> str:
    # This node returns the edge name, not an updated state
    if state["operation"] == "plus":
        return "addition_operation" 
    elif state["operation"] == "minus":
        return "subtraction_operation" 

2. Add the Router Node (Using Lambda)

Since the router function above only compares and returns an edge name without modifying the state, the node action uses a lambda pass-through function.

# Router node uses lambda state: state because it doesn't modify the state
graph.add_node("router", lambda state: state) 
# Also add target operation nodes (add_node, subtract_node)

3. Define Conditional Edges

The path_map links the returned edge name (e.g., “addition_operation”) to the target node name (e.g., “add_node”).

graph.add_edge(START, "router") # Connect start to router

graph.add_conditional_edges(
    source="router",
    path=decide_next_node, # The routing function
    path_map={
        "addition_operation": "add_node", 
        "subtraction_operation": "subtract_node" 
    }
)
# Edges from operations back to the END point
graph.add_edge("add_node", END) 
graph.add_edge("subtract_node", END)
app = graph.compile()

D. Looping Graph

Looping logic is handled by a conditional edge routing back to a prior node, typically controlled by a counter variable in the state.

1. Define the Continuation Logic

The function checks a state attribute (counter) to decide whether to continue the loop or exit.

def should_continue(state: AgentState):
    # Assuming AgentState has a 'counter: int' attribute
    if state["counter"] < 5:
        print(f"Counter: {state['counter']}") #
        return "loop"
    else:
        return "exit"

2. Implement the Conditional Loop Edge

If the result is “loop”, execution routes back to the source node (random_node). If “exit”, it routes to END.

# Define the conditional edge starting from the random node
graph.add_conditional_edges(
    source="random_node",
    path=should_continue, 
    path_map={
        "loop": "random_node", # Loop back to itself
        "exit": END # End the graph
    }
)

V. AI Agent Implementation (React/Chatbots)

A. Defining State for LLM Conversation History

For agents, the state must store a history of various message types (BaseMessage) and use the add_messages reducer function to handle appending automatically.

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

class AgentState(TypedDict):
    # Ensures new messages are appended instead of overwriting
    messages: Annotated[Sequence[BaseMessage], add_messages] 

B. Tool Definition and Binding (React Agent)

Tools are defined as standard Python functions using the @tool decorator. The docstring is mandatory as it tells the LLM the purpose of the tool.

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

@tool 
def add(a: int, b: int) -> int:
    """This is an addition function that adds two numbers together.""" # Necessary docstring
    return a + b

tools = [add] 
model = ChatOpenAI(model="gpt-4o") 

# Bind tools to the LLM so the model knows they exist
llm_with_tools = model.bind_tools(tools) 

C. LLM Node Call and System Prompt

The LLM node receives the entire conversation history (state["messages"]) and optionally binds a system prompt for instructions.

def model_call(state: AgentState) -> AgentState:
    # Define system prompt using SystemMessage for readability
    system_prompt = SystemMessage(content="You are my AI assistant.") 
    
    # Combine system prompt with existing conversation history (messages)
    messages_to_send = [system_prompt] + state["messages"] 
    
    # Invoke the bound model with the messages
    response = llm_with_tools.invoke(messages_to_send) 
    
    # Return the updated state. add_messages handles appending the response
    return {"messages": response}

D. React Agent Conditional Edge (Loop Logic)

The conditional edge checks the latest message from the LLM to see if it requested a tool call. If so, it continues the loop to the Tool Node.

def should_continue(state: AgentState) -> str:
    # Get the most recent message
    last_message = state["messages"][-1]
    
    # Check if the LLM response requested tool calls 
    if not last_message.tool_calls:
        return "end" # No more tool calls needed
    else:
        return "continue" # Proceed to tool node

Graph Flow for React: The LLM node routes execution:

Agent Node (LLM decides)
Conditional Edge (should_continue)
- If Tool Call: Go to “continue” edge ➡️ Tool Node (runs tool)
- If Final Answer: Go to “end” edge ➡️ END

A directed edge must also be added from the Tool Node back to the Agent Node to allow the LLM to process the tool output and potentially loop again.

# Agent to Tool Node (via conditional edge setup)
# Tool Node back to Agent Node (to process tool result)
graph.add_edge("tool_node", "agent") 

LangChain MCP Adapters

A useful repository for Multi-Channel Protocol (MCP) adapters that integrate with LangChain and related workflows:

Brief: adapters for connecting LangChain-based agents and pipelines to external messaging channels and MCP-compatible systems (useful for routing messages, webhooks, and channel-specific glue logic).

LangSmith Server A2A (Agent-to-Agent)

LangSmith Server A2A enables agent-to-agent communication patterns (A2A) using the LangSmith server. This is useful when you want agents to call other agents or coordinate workflows through a central server.

Docs: https://docs.langchain.com/langsmith/server-a2a

Summary:

Purpose: Route messages and tool calls between agents via the LangSmith server, enabling orchestrated multi-agent flows and persistent observability.
Typical steps:
1. Configure the LangSmith server and obtain credentials/API endpoint.
2. Register agents (or agent endpoints) with LangSmith.
3. Use the provided client bindings or HTTP endpoints to send A2A requests and receive responses/events.
Common use cases: tool orchestration across agents, agent escalation/fallback, distributed tool execution, and audit/logging of inter-agent interactions.

Example (conceptual):

# Pseudocode: use the LangSmith A2A client to send a request to another agent
from langsmith_client import LangSmithClient

client = LangSmithClient(api_key="YOUR_KEY", base_url="https://your-langsmith-server")
response = client.send_to_agent(agent_id="research_agent", payload={"query": "Summarize Q4 report"})
print(response)

Refer to the official Server A2A docs for concrete setup, API details, and security best practices: https://docs.langchain.com/langsmith/server-a2a

RAG Example

from dotenv import load_dotenv
import os
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, ToolMessage
from operator import add as add_messages
from langchain_openai import ChatOpenAI
from langchain_openai import OpenAIEmbeddings
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma
from langchain_core.tools import tool

load_dotenv()

llm = ChatOpenAI(
    model="gpt-4o", temperature = 0) # I want to minimize hallucination - temperature = 0 makes the model output more deterministic 

# Our Embedding Model - has to also be compatible with the LLM
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
)


pdf_path = "Stock_Market_Performance_2024.pdf"


# Safety measure I have put for debugging purposes :)
if not os.path.exists(pdf_path):
    raise FileNotFoundError(f"PDF file not found: {pdf_path}")

pdf_loader = PyPDFLoader(pdf_path) # This loads the PDF

# Checks if the PDF is there
try:
    pages = pdf_loader.load()
    print(f"PDF has been loaded and has {len(pages)} pages")
except Exception as e:
    print(f"Error loading PDF: {e}")
    raise

# Chunking Process
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)


pages_split = text_splitter.split_documents(pages) # We now apply this to our pages

persist_directory = r"C:\Vaibhav\LangGraph_Book\LangGraphCourse\Agents"
collection_name = "stock_market"

# If our collection does not exist in the directory, we create using the os command
if not os.path.exists(persist_directory):
    os.makedirs(persist_directory)


try:
    # Here, we actually create the chroma database using our embeddigns model
    vectorstore = Chroma.from_documents(
        documents=pages_split,
        embedding=embeddings,
        persist_directory=persist_directory,
        collection_name=collection_name
    )
    print(f"Created ChromaDB vector store!")
    
except Exception as e:
    print(f"Error setting up ChromaDB: {str(e)}")
    raise


# Now we create our retriever 
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5} # K is the amount of chunks to return
)

@tool
def retriever_tool(query: str) -> str:
    """
    This tool searches and returns the information from the Stock Market Performance 2024 document.
    """

    docs = retriever.invoke(query)

    if not docs:
        return "I found no relevant information in the Stock Market Performance 2024 document."
    
    results = []
    for i, doc in enumerate(docs):
        results.append(f"Document {i+1}:\n{doc.page_content}")
    
    return "\n\n".join(results)


tools = [retriever_tool]

llm = llm.bind_tools(tools)

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]


def should_continue(state: AgentState):
    """Check if the last message contains tool calls."""
    result = state['messages'][-1]
    return hasattr(result, 'tool_calls') and len(result.tool_calls) > 0


system_prompt = """
You are an intelligent AI assistant who answers questions about Stock Market Performance in 2024 based on the PDF document loaded into your knowledge base.
Use the retriever tool available to answer questions about the stock market performance data. You can make multiple calls if needed.
If you need to look up some information before asking a follow up question, you are allowed to do that!
Please always cite the specific parts of the documents you use in your answers.
"""


tools_dict = {our_tool.name: our_tool for our_tool in tools} # Creating a dictionary of our tools

# LLM Agent
def call_llm(state: AgentState) -> AgentState:
    """Function to call the LLM with the current state."""
    messages = list(state['messages'])
    messages = [SystemMessage(content=system_prompt)] + messages
    message = llm.invoke(messages)
    return {'messages': [message]}


# Retriever Agent
def take_action(state: AgentState) -> AgentState:
    """Execute tool calls from the LLM's response."""

    tool_calls = state['messages'][-1].tool_calls
    results = []
    for t in tool_calls:
        print(f"Calling Tool: {t['name']} with query: {t['args'].get('query', 'No query provided')}")
        
        if not t['name'] in tools_dict: # Checks if a valid tool is present
            print(f"\nTool: {t['name']} does not exist.")
            result = "Incorrect Tool Name, Please Retry and Select tool from List of Available tools."
        
        else:
            result = tools_dict[t['name']].invoke(t['args'].get('query', ''))
            print(f"Result length: {len(str(result))}")
            

        # Appends the Tool Message
        results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result)))

    print("Tools Execution Complete. Back to the model!")
    return {'messages': results}


graph = StateGraph(AgentState)
graph.add_node("llm", call_llm)
graph.add_node("retriever_agent", take_action)

graph.add_conditional_edges(
    "llm",
    should_continue,
    {True: "retriever_agent", False: END}
)
graph.add_edge("retriever_agent", "llm")
graph.set_entry_point("llm")

rag_agent = graph.compile()


def running_agent():
    print("\n=== RAG AGENT===")
    
    while True:
        user_input = input("\nWhat is your question: ")
        if user_input.lower() in ['exit', 'quit']:
            break
            
        messages = [HumanMessage(content=user_input)] # converts back to a HumanMessage type

        result = rag_agent.invoke({"messages": messages})
        
        print("\n=== ANSWER ===")
        print(result['messages'][-1].content)


running_agent()