Guardian-Graph

Guardian-Graph is a reliability layer for LangGraph applications. It validates state transitions against strict Pydantic contracts, detects runaway loops, and emits structured events you can route to your observability stack.

Quickstart

from guardian_graph import wrap_with_guardian
from pydantic import BaseModel

class MyState(BaseModel):
    query: str
    answer: str | None = None

class DemoGraph:
    def invoke(self, state: dict) -> dict:
        return {**state, "answer": "hello"}

guarded_graph = wrap_with_guardian(DemoGraph(), schema=MyState)
result = guarded_graph.invoke({"query": "hello"})

Docs

• docs/index.md — documentation map and quickstart
• docs/getting-started.md — install and first guarded graph
• docs/concepts.md — validation, loop guard, judge, auto-heal
• docs/api-reference.md — public API reference
• docs/events.md — event schema reference
• docs/integrations.md — LangSmith, Postgres, FastAPI ingestion

Wrap a LangGraph Graph

from guardian_graph import wrap_with_guardian

# `base_graph` is your compiled LangGraph graph with an `.invoke(...)` method.
guarded_graph = wrap_with_guardian(
    base_graph,
    schema=MyState,
    max_steps=20,
    stagnation_limit=5,
)
result = guarded_graph.invoke({"query": "hello"})

max_steps is enforced across the full .invoke(...) execution (counting steps across nodes) to prevent runaway tool loops.

Auto-Heal (Optional)

def repair(prompt: str, bad_state: dict) -> dict:
    # Replace with a model call; keep return value JSON-serializable.
    return {**bad_state, "answer": "repaired"}

guarded_graph = wrap_with_guardian(
    base_graph,
    schema=MyState,
    repair=repair,
    repair_retries=2,
)

Observability

from guardian_graph import CompositeEmitter, JsonlFileEmitter

file_emitter = JsonlFileEmitter("logs/guardian.jsonl")
base_graph = DemoGraph()
guarded_graph = wrap_with_guardian(
    base_graph,
    schema=MyState,
    emitter=CompositeEmitter([file_emitter]),
)

Hallucination Judge (Optional)

from guardian_graph.core.judge import JudgeDecision, JudgeInput, JudgeResult
from guardian_graph import wrap_with_guardian

def judge(input: JudgeInput) -> JudgeResult:
    if "risky" in input.state.get("answer", ""):
        return JudgeResult(decision=JudgeDecision.block, rationale="unsafe output")
    return JudgeResult(decision=JudgeDecision.allow)

base_graph = DemoGraph()
guarded_graph = wrap_with_guardian(base_graph, schema=MyState, judge=judge)

HITL Gateway (Optional)

from guardian_graph import interrupt_human_approval

def dangerous_tool(state: dict) -> dict:
    interrupt_human_approval(action="delete_users", payload={"ids": [1, 2, 3]})
    return {}

Postgres Checkpointer Integration

Guardian-Graph can return a LangGraph Postgres checkpointer when LangGraph is installed:

from guardian_graph import get_postgres_checkpointer

checkpointer = get_postgres_checkpointer("postgresql://user:pass@localhost:5432/db")

If langgraph.checkpoint.postgres is not installed, the helper raises a clear ImportError with guidance.

Integrations

See docs/integrations.md for LangSmith, Postgres checkpointers, and FastAPI event ingestion.

Tests

python -m pytest
ruff check .