LangChain / LangGraph Integration

Installation

pip install comply54 langchain-core langgraph langchain-anthropic

Patterns at a glance

comply54 ships three LangGraph primitives:

Import	Type	When to use
Comply54Guard + comply54_route	Guard node	Recommended — intercepts tool calls in ReAct agents
compliance_node(compliance)	State node	Compliance check as explicit step in a pipeline graph
comply54_tool(compliance)	StructuredTool	The LLM decides when to call the compliance check

For all regulated sectors (fintech, healthcare, insurance), use Comply54Guard — compliance must intercept every tool call automatically.

Pattern 1: Comply54Guard (recommended)

Comply54Guard sits between the agent node and the tools node. When the LLM emits tool_calls, the guard evaluates each against all compliance packs. Blocked calls receive a ToolMessage error; the LLM reads it and explains the refusal to the user.

from langchain_anthropic import ChatAnthropic
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
from comply54.sectors import NigeriaFintechCompliance
from comply54.langchain import Comply54Guard, comply54_route

compliance = NigeriaFintechCompliance()
tools = [transfer_funds_tool, check_balance_tool]

def agent_node(state):
    llm = ChatAnthropic(model="claude-haiku-4-5-20251001").bind_tools(tools)
    return {"messages": [llm.invoke(state["messages"])]}

guard = Comply54Guard(compliance, tools)

def build_graph():
    graph = StateGraph(MessagesState)

    graph.add_node("agent", agent_node)
    graph.add_node("guard", guard)
    graph.add_node("tools", ToolNode(tools))

    graph.set_entry_point("agent")
    graph.add_conditional_edges("agent", comply54_route, {"guard": "guard", "__end__": "__end__"})
    graph.add_edge("guard", "tools")
    graph.add_edge("tools", "agent")

    return graph.compile()


app = build_graph()
result = app.invoke({"messages": [("user", "Transfer ₦15,000,000 to account 0123456789")]})
# Last AIMessage: "I can't process this transfer — it exceeds the ₦10,000,000 CBN NIP cap."

Passing context to the guard: Set state["compliance_context"] before the agent runs:

result = app.invoke({
    "messages": [("user", "Transfer ₦500,000 to 0123456789")],
    "compliance_context": {"kyc_tier": 2},
})

Pattern 2: compliance_node (pipeline graphs)

Use compliance_node when compliance is a named, explicit step in a non-ReAct pipeline:

from comply54.sectors import NigeriaFintechCompliance
from comply54.langchain import compliance_node
from typing import TypedDict, Any
from langgraph.graph import StateGraph


class AgentState(TypedDict):
    action: str
    params: dict
    context: dict
    output: str
    blocked: bool
    compliance_result: Any
    messages: list


compliance = NigeriaFintechCompliance()

def agent_node(state: AgentState) -> AgentState:
    return {**state, "action": "transfer_funds", "params": {"amount": 15_000_000}}

def execute_node(state: AgentState) -> AgentState:
    result = execute_transfer(state["params"])
    return {**state, "output": result}

def rejection_node(state: AgentState) -> AgentState:
    reason = state["compliance_result"].primary_violation.messages[0]
    return {**state, "output": f"Blocked: {reason}"}


def build_graph():
    graph = StateGraph(AgentState)

    graph.add_node("agent", agent_node)
    graph.add_node("check_compliance", compliance_node(compliance))
    graph.add_node("execute", execute_node)
    graph.add_node("reject", rejection_node)

    graph.set_entry_point("agent")
    graph.add_edge("agent", "check_compliance")

    graph.add_conditional_edges(
        "check_compliance",
        lambda state: "reject" if state.get("blocked") else "execute",
        {"execute": "execute", "reject": "reject"},
    )
    graph.add_edge("execute", "__end__")
    graph.add_edge("reject", "__end__")

    return graph.compile()


app = build_graph()
result = app.invoke({
    "action": "transfer_funds",
    "params": {"amount": 15_000_000, "currency": "NGN"},
    "context": {"kyc_tier": 3},
    "messages": [],
})
print(result["output"])
# Blocked: CBN NIP Framework §4.2: Transaction exceeds ₦10,000,000 cap

Pattern 3: comply54_tool

The compliance check becomes a tool the LLM calls before taking actions. Best for exploratory agents where the model needs to reason about whether an action is compliant before deciding to execute it.

from comply54.langchain import comply54_tool
from langchain_anthropic import ChatAnthropic

compliance = NigeriaFintechCompliance()
tools = [comply54_tool(compliance)]

llm = ChatAnthropic(model="claude-sonnet-4-6").bind_tools(tools)

response = llm.invoke(
    "Check if I can transfer ₦12,000,000 to account 0123456789. My KYC tier is 3."
)
# The LLM calls check_compliance tool → receives "deny" with CBN message
# → responds explaining the block

Output guard pattern

Check the agent's output before sending it to the user — catches BVN/NIN leakage regardless of where in the graph it occurs.

def output_guard_node(state: AgentState) -> AgentState:
    result = compliance.check(
        action="respond_to_user",
        output=state["output"],
    )

    if result.blocked:
        # Redact and return a safe message
        safe_output = "[Response contained sensitive data and was redacted]"
        return {**state, "output": safe_output, "pii_blocked": True}

    return state

Add it as the last node before __end__:

graph.add_node("output_guard", output_guard_node)
graph.add_edge("execute", "output_guard")
graph.add_edge("output_guard", "__end__")

Three-stage guard (production pattern)

Production Nigerian fintech agents should gate compliance at three points:

User input → [Input guard] → Agent reasoning → [Tool call guard] → Execute → [Output guard] → User

# Stage 1: Input guard — check what the user is asking to do
def input_guard(state):
    result = compliance.check(
        action=extract_intent(state["messages"]),
        params=extract_params(state["messages"]),
        context=state["context"],
    )
    return {**state, "input_check": result}

# Stage 2: Tool call guard — check before every tool execution
def before_tool(tool_call, context):
    result = compliance.check(
        action=tool_call["name"],
        params=tool_call["args"],
        context=context,
    )
    if result.blocked:
        raise ToolCallBlocked(result.primary_violation.messages[0])

# Stage 3: Output guard — check before response is sent to user
def output_guard(state):
    result = compliance.check(
        action="respond_to_user",
        output=state["output"],
    )
    if result.blocked:
        return {**state, "output": redact(state["output"])}
    return state

Streaming with compliance

comply54 evaluates synchronously in microseconds — it does not interfere with LangChain streaming:

async for chunk in app.astream(input_state):
    # compliance evaluation happens synchronously inside nodes
    # streaming is unaffected
    yield chunk

Related

• Nigeria Fintech sector pack
• Nigeria Health sector pack
• Nigeria Insurance sector pack
• CrewAI integration