LangGraph Integration

Wire Nomotic governance into a LangGraph state graph using GovernedAgentBase as governance nodes.

Installation

pip install nomotic langgraph

Pattern

Each governed operation becomes a graph node that calls governed_run(). Governance verdicts drive conditional edges — allowed actions proceed, vetoed actions route to error handling.

Example

from nomotic import GovernanceRuntime
from nomotic.governed_agent import GovernedAgentBase, GovernanceVetoError

def build_governed_graph(governed_agent: GovernedAgentBase):
    """Build a LangGraph StateGraph with governed nodes."""

    def governed_read_node(state: dict) -> dict:
        """Read node with governance gate."""
        try:
            result = governed_agent.governed_run(
                action_type="read",
                target=state.get("target", "data"),
                execute_fn=lambda: f"Data from {state.get('target', 'data')}",
            )
            return {**state, "data": result, "status": "read_complete"}
        except GovernanceVetoError as e:
            return {**state, "error": str(e), "status": "read_denied"}

    def governed_write_node(state: dict) -> dict:
        """Write node with governance gate."""
        try:
            result = governed_agent.governed_run(
                action_type="write",
                target=state.get("target", "output"),
                execute_fn=lambda: f"Wrote: {state.get('data', '')}",
            )
            return {**state, "result": result, "status": "write_complete"}
        except GovernanceVetoError as e:
            return {**state, "error": str(e), "status": "write_denied"}

    def should_continue(state: dict) -> str:
        """Route based on governance verdict."""
        if state.get("status", "").endswith("_denied"):
            return "end"
        return "write"

    # Build the graph:
    # read → (if allowed) → write → end
    # read → (if denied) → end
    from langgraph.graph import StateGraph, END
    graph = StateGraph(dict)
    graph.add_node("read", governed_read_node)
    graph.add_node("write", governed_write_node)
    graph.set_entry_point("read")
    graph.add_conditional_edges("read", should_continue, {"write": "write", "end": END})
    graph.add_edge("write", END)
    return graph.compile()

Governance evaluations happen at each node. If a read is denied, the graph short-circuits to END without attempting the write. Every evaluation is recorded in the audit trail.