graph.add_node("research", research_node)
# or
graph.add_node("agent", agent_llm_with_tools)

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode

# Adding different types of nodes
graph = StateGraph(MessagesState)

graph.add_node("agent", agent_node)                    # LLM / Agent node
graph.add_node("tools", ToolNode(tools))               # Tool node
graph.add_node("retriever", retrieval_node)            # Custom node

# With metadata and retry (advanced)
graph.add_node(
    "web_search",
    web_search_node,
    metadata={"description": "Search the internet", "type": "tool"},
    retry=RetryPolicy(max_attempts=3)                  # Auto-retry on failure
)

• Use clear, descriptive names that explain what the node does
• Use snake_case (lowercase with underscores)
• Keep names concise but meaningful
• Use verbs when possible (fetch, validate, generate, summarize, etc.)

# Bad Examples
builder.add_node("node1", func1)           # Too vague
builder.add_node("step", func)             # Not descriptive
builder.add_node("my_node", process)       # Unclear purpose
builder.add_node("llm", llm_call)          # Too generic (what does it do?)

from langgraph.graph import StateGraph

def my_node(state: dict):
    # Perform any logic
    result = some_calculation(state["input"])
    return {"output": result, "messages": [...]}

# Add to graph
graph.add_node("my_node", my_node)

class MyNode:
    def __init__(self, model_name: str):
        self.llm = ChatOpenAI(model=model_name)
    
    def __call__(self, state: dict):
        response = self.llm.invoke(state["messages"])
        return {"messages": [response]}

# Usage
node = MyNode("gpt-4o")
graph.add_node("agent", node)

def llm_node(state: MessagesState):
    response = llm.invoke(state["messages"])   # or .ainvoke()
    return {"messages": [response]}

# Common pattern with tool calling
tools = [search_web, get_weather, run_sql]
llm_with_tools = llm.bind_tools(tools)

def agent_node(state: MessagesState):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

• Web search
• Database queries
• Calculator tools
• Weather APIs
• Custom business logic tools

from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
from langchain_community.tools import TavilySearchResults

# 1. Define your tools using LangChain
@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

search_tool = TavilySearchResults(max_results=2)

tools = [multiply, search_tool]

# 2. Create ToolNode
tool_node = ToolNode(tools=tools)

# 3. Add to your graph
graph.add_node("tools", tool_node)

• Classifying user requests
• Choosing different workflow paths
• Deciding whether to call tools or end the process
• Routing based on LLM output or conditions

def route_tools(state):
    """Router that decides the next node based on the last message."""
    messages = state["messages"]
    last_message = messages[-1]
    
    # If the LLM called a tool, go to tools node
    if last_message.tool_calls:
        return "tools"
    
    # Otherwise, end the graph
    return "END"


# Add to graph
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

# Conditional routing
graph.add_conditional_edges(
    "agent",          # Start from agent node
    route_tools,      # Router function
    {
        "tools": "tools",
        "END": END
    }
)

• JSON validation
• Schema checking
• Response quality checks
• Safety / moderation filters
• Data consistency verification

from pydantic import BaseModel, ValidationError

class ResponseSchema(BaseModel):
    answer: str
    confidence: float

def validation_node(state):
    try:
        validated = ResponseSchema.model_validate(state["output"])
        return {"validated_output": validated.model_dump()}
    except ValidationError as e:
        return {"error": str(e), "next": "fix_output"}  # Send back for retry

• Storing memory
• Retrieving memory
• Updating conversation history

• Vector database retrieval (Chroma, Pinecone, etc.)
• Persistent memory updates
• Message summarization
• Conversation history management

def memory_node(state):
    # Retrieve relevant past memories
    query = state["messages"][-1].content
    relevant_memory = vector_retriever.invoke(query)
    
    # Update state with memory
    return {
        "messages": state["messages"] + relevant_memory,
        "memory": "updated"
    }

• Retrieving relevant documents
• Searching embeddings (vector similarity search)
• Fetching additional context
• Knowledge base lookups
• Semantic search

from langchain_core.vectorstores import VectorStoreRetriever

# Usually created from a vector store
retriever = vector_store.as_retriever(search_kwargs={"k": 4})

def retrieval_node(state):
    """Retrieve relevant documents based on the user's query."""
    query = state["messages"][-1].content
    
    retrieved_docs = retriever.invoke(query)
    
    return {
        "documents": retrieved_docs,           # Add documents to state
        "context": "\n\n".join([doc.page_content for doc in retrieved_docs])
    }

# Add to graph
graph.add_node("retrieve", retrieval_node)

• Approval workflows
• Manual review and editing
• Feedback collection
• Content moderation
• Decision confirmation

• interrupt_before / interrupt_after
• Checkpoints & persistence (SQLite, PostgreSQL, etc.)

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END

# Enable checkpointing
checkpointer = MemorySaver()

# Build graph with interrupt
graph = StateGraph(MessagesState)

# ... add other nodes ...

# Interrupt before a sensitive node
graph.add_node("human_approval", human_approval_node)

# Compile with interrupt
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_approval"]   # Pause before this node
)

# Run until interrupt
config = {"configurable": {"thread_id": "123"}}
result = app.invoke(inputs, config)

# Later, after human input:
app.invoke(None, config)   # Resume with human feedback

• Parallel searches (e.g., web + vector search)
• Concurrent tool calls
• Batch processing
• Multi-document analysis
• Gathering data from multiple sources

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# Example: Run multiple retrievers in parallel
def web_search(state):
    return {"documents": web_retriever.invoke(state["query"])}

def vector_search(state):
    return {"documents": vector_retriever.invoke(state["query"])}

def combine_results(state):
    # Merge documents from all parallel branches
    return {"documents": state["documents"]}

graph = StateGraph(MessagesState)

graph.add_node("web_search", web_search)
graph.add_node("vector_search", vector_search)
graph.add_node("combine", combine_results)

# Fan-out: Start both searches in parallel
graph.add_edge(START, "web_search")
graph.add_edge(START, "vector_search")

# Fan-in: Wait for both to finish, then combine
graph.add_edge("web_search", "combine")
graph.add_edge("vector_search", "combine")

graph.add_edge("combine", END)

from langgraph.types import Send

def route_to_parallel(state):
    return [
        Send("web_search", state),
        Send("vector_search", state)
    ]

graph.add_conditional_edges(START, route_to_parallel)

• Merging search results from multiple sources
• Summarizing outputs from parallel operations
• Combining tool responses
• Reducer-based state updates
• Final result synthesis

from langgraph.graph.message import add_messages
from typing import Annotated

# Using built-in reducers (Recommended)
class State(TypedDict):
    messages: Annotated[list, add_messages]
    documents: Annotated[list, lambda x, y: x + y]   # Custom reducer

def aggregate_results(state):
    """Merge and summarize results from parallel branches."""
    all_docs = state.get("documents", [])
    
    # Optional: Summarize using LLM
    summary = llm.invoke(f"Summarize these documents:\n{all_docs}")
    
    return {
        "documents": all_docs,           # Keep all raw docs
        "summary": summary.content       # Add final combined result
    }

# Add to graph
graph.add_node("aggregate", aggregate_results)

# Connect parallel branches to aggregator
graph.add_edge("web_search", "aggregate")
graph.add_edge("vector_search", "aggregate")
graph.add_edge("tool_search", "aggregate")

• Confidence scoring
• Retry decisions
• Fallback selection
• Threshold-based routing
• Quality checks and branching logic

def decision_node(state):
    """Evaluate confidence and decide next action."""
    last_message = state["messages"][-1]
    confidence = getattr(last_message, "confidence", 0.0)
    
    if confidence >= 0.85:
        return "end"                    # High confidence → finish
    elif confidence >= 0.6:
        return "retry"                  # Medium → try again
    else:
        return "fallback"               # Low → use fallback path


# Add to graph
graph.add_node("decide", decision_node)

# Conditional routing based on decision
graph.add_conditional_edges(
    "decide",
    lambda state: decision_node(state),   # or pass the function directly
    {
        "end": END,
        "retry": "agent",
        "fallback": "fallback_node"
    }
)

• Modular workflows
• Reusable components
• Hierarchical agents (agent teams)
• Multi-stage pipelines
• Easier testing and maintenance

from langgraph.graph import StateGraph, START, END

# 1. Create a reusable subgraph
def create_research_subgraph():
    subgraph = StateGraph(MessagesState)
    
    subgraph.add_node("retrieve", retrieval_node)
    subgraph.add_node("analyze", analysis_node)
    subgraph.add_node("summarize", summarizer_node)
    
    subgraph.add_edge(START, "retrieve")
    subgraph.add_edge("retrieve", "analyze")
    subgraph.add_edge("analyze", "summarize")
    subgraph.add_edge("summarize", END)
    
    return subgraph.compile()

# 2. Add the subgraph as a normal node in the main graph
research_subgraph = create_research_subgraph()

main_graph = StateGraph(MessagesState)
main_graph.add_node("research", research_subgraph)   # ← Subgraph as node

main_graph.add_node("final_answer", final_answer_node)

main_graph.add_edge(START, "research")
main_graph.add_edge("research", "final_answer")
main_graph.add_edge("final_answer", END)

from langchain_core.runnables import RunnableConfig
from langgraph.graph import MessagesState

def my_node(state: MessagesState, config: RunnableConfig):
    """Node that can access thread metadata and runtime config."""
    
    # Access configurable fields (very common)
    thread_id = config["configurable"].get("thread_id")
    user_id = config["configurable"].get("user_id")
    
    # Access metadata and tags
    metadata = config.get("metadata", {})
    tags = config.get("tags", [])
    
    print(f"Running on thread: {thread_id} | User: {user_id}")
    
    # Your logic here...
    return {"messages": ["Node executed successfully"]}