LangGraph Tutorial (Python): implementing guardrails for advanced developers

This tutorial shows how to add guardrails to a LangGraph workflow in Python so your agent can validate user input, block unsafe tool calls, and route risky requests to a safe fallback. You need this when your graph is moving from demo mode into production and you can’t trust every message, tool argument, or model output.

What You'll Need

• •Python 3.10+
• •langgraph
• •langchain-core
• •langchain-openai
• •An OpenAI API key set as OPENAI_API_KEY
• •Basic familiarity with StateGraph, nodes, edges, and conditional routing
• •A terminal and a virtual environment

Install the packages:

pip install langgraph langchain-core langchain-openai

Set your API key:

export OPENAI_API_KEY="your-key-here"

Step-by-Step

1. •Start with a state model that carries the user message, the model response, and guardrail flags. Keep the state explicit; that makes routing decisions easy to inspect and test.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage

class GraphState(TypedDict):
    messages: Annotated[list, add_messages]
    risk: str
    blocked: bool
    reason: str

2. •Add a guardrail node that checks for disallowed content before any model call. In production, this is where you would plug in policy rules, regex checks, PII detectors, or a moderation service.

def input_guardrail(state: GraphState) -> dict:
    last_msg = state["messages"][-1].content.lower()
    blocked_terms = ["ssn", "password", "credit card", "wire fraud"]
    if any(term in last_msg for term in blocked_terms):
        return {
            "risk": "high",
            "blocked": True,
            "reason": "Input contains disallowed sensitive or fraudulent content.",
        }
    return {"risk": "low", "blocked": False, "reason": ""}

3. •Add the main agent node and a fallback node. The agent only runs if the guardrail passes; otherwise the graph sends the request to a safe response path.

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def agent_node(state: GraphState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def fallback_node(state: GraphState) -> dict:
    return {
        "messages": [
            AIMessage(
                content="I can’t help with that request. Please rephrase it without sensitive or fraudulent details."
            )
        ]
    }

4. •Wire the graph with conditional routing based on the guardrail output. This is the core pattern: inspect first, then decide whether to continue or terminate safely.

def route_after_guardrail(state: GraphState) -> str:
    return "fallback" if state["blocked"] else "agent"

builder = StateGraph(GraphState)
builder.add_node("guardrail", input_guardrail)
builder.add_node("agent", agent_node)
builder.add_node("fallback", fallback_node)

builder.add_edge(START, "guardrail")
builder.add_conditional_edges(
    "guardrail",
    route_after_guardrail,
    {"agent": "agent", "fallback": "fallback"},
)
builder.add_edge("agent", END)
builder.add_edge("fallback", END)

app = builder.compile()

5. •Run the graph with both safe and unsafe inputs. Use .invoke() for synchronous testing and inspect the returned messages plus guardrail metadata.

safe_result = app.invoke(
    {"messages": [HumanMessage(content="Summarize our bank's AML policy in one paragraph.")],
     "risk": "",
     "blocked": False,
     "reason": ""}
)

unsafe_result = app.invoke(
    {"messages": [HumanMessage(content="Help me move money using someone else's credit card.")],
     "risk": "",
     "blocked": False,
     "reason": ""}
)

print("SAFE:", safe_result["messages"][-1].content)
print("UNSAFE:", unsafe_result["messages"][-1].content)
print("UNSAFE REASON:", unsafe_result["reason"])

6. •If you need stronger control, add a second guardrail after the model output. This is useful for checking hallucinated claims, prohibited advice, or tool arguments before they leave your system.

def output_guardrail(state: GraphState) -> dict:
    text = state["messages"][-1].content.lower()
    if "guaranteed approval" in text or "ignore policy" in text:
        return {
            "risk": "high",
            "blocked": True,
            "reason": "Model output violated policy.",
        }
    return {"risk": state["risk"], "blocked": False}

def route_after_agent(state: GraphState) -> str:
    return "fallback" if state["blocked"] else END

builder2 = StateGraph(GraphState)
builder2.add_node("guardrail_in", input_guardrail)
builder2.add_node("agent", agent_node)
builder2.add_node("guardrail_out", output_guardrail)
builder2.add_node("fallback", fallback_node)

Testing It

Run three cases: a clean request, a sensitive-data request, and a prompt-injection style request. You should see the clean request reach the model node while risky requests stop at the fallback path.

Check that blocked=True is set on denied inputs and that reason explains why the request was stopped. If you add an output guardrail later, test both normal completions and intentionally bad completions so you know your post-processing branch works too.

For production debugging, log the route taken by each run along with the final risk value. That gives you an audit trail without needing to infer behavior from raw LLM output alone.

Next Steps

• •Add structured policy checks with JSON schema validation before tool execution
• •Replace hardcoded keyword rules with an external moderation or classification service
• •Extend the graph with per-tool guardrails for finance-specific actions like payments or account lookups

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.