LangGraph

This integration enables you to apply centralized authorization decisions at each step of a stateful agent graph, ensuring that execution paths, data access, and outputs are governed by Policy.

The integration is built on top of:

• core-plainid for authorization and Policy evaluation.
• langchain-plainid for retrieval and filtering.

See the following articles for more information:

Integration Core
LangChain

Install the SDK

Install the SDK to add the required PlainID and LangGraph integration packages.

To install the SDK:

1. Run the installation command.

pip install langgraph-plainid

This installs:

• core-plainid
• langchain-plainid
• langgraph-plainid

Define Agent State

All nodes operate on a shared AgentState TypedDict that flows through the graph. The state contains the request_context for identity information and optional sub-states for each node type. Alternatively, request_context can be provided at construction time to the underlying components, for example, PlainIDPermissionsProvider and FilterDirectiveProvider. See core-plainid.

To define Agent State:

1. Import the required class.

from langgraph_plainid.models.state.agent_state import AgentState

2. Review the AgentState fields.

The state typically includes:

• request_context for identity information.
• categorization for prompt classification input.
• anonymization for text processing input and output.
• retrieval for the query and retrieved documents.

Initialize Permissions Provider

All PlainID nodes rely on the same permissions provider.

To initialize the permissions provider:

1. Import the required class.

from core_plainid.utils.plainid_permissions_provider import PlainIDPermissionsProvider

2. Configure the PlainIDPermissionsProvider.

permissions_provider = PlainIDPermissionsProvider(
    base_url="https://platform-product.us1.plainid.io",
    client_id="your_client_id",
    client_secret="your_client_secret",
)

Add Categorization Node

The categorization node validates whether a prompt is allowed based on PlainID Policies.

To add a Categorization Node:

1. Import the required classes.

from core_plainid.categorization.categorizer import Categorizer
from langgraph_plainid.nodes.categorization_node import CategorizationNode

2. Create the Categorizer.

categorizer = Categorizer(
    classifier_provider=classifier,
    permissions_provider=permissions_provider,
    all_categories=["contract", "HR", "finance"],
)

3. Create the CategorizationNode.

categorization_node = CategorizationNode(
    categorizer=categorizer,
    next_node="anonymizer",
)

Note that next_node is optional.

Add an Anonymization Node

The anonymization node applies Policy-driven masking or encryption to sensitive data.

To add an Anonymization Node:

1. Import the required classes.

from core_plainid.anonymization.presidio_anonymizer import PresidioAnonymizer
from langgraph_plainid.nodes.anonymizer_node import AnonymizerNode

2. Create the anonymizer.

anonymizer = PresidioAnonymizer(
    permissions_provider=permissions_provider,
    encrypt_key="your_16_char_key!",
)

3. Create the AnonymizerNode.

anonymizer_node = AnonymizerNode(
    anonymizer=anonymizer,
    next_node="retrieval",
)

This node:

• Reads from state["anonymization"]["query"].
• Writes to state["anonymization"]["output_text"].

Add Retrieval Node

The retrieval node enforces PlainID filtering when querying vector stores.

To add a Retrieval Node:

1. Import the required classes.

from langchain_plainid.retrieval.filter_directive_provider import FilterDirectiveProvider
from langchain_plainid.retrieval.multi_store_retriever import MultiStoreRetriever
from langgraph_plainid.nodes.retrieval_node import RetrievalNode

2. Configure the FilterDirectiveProvider.

filter_provider = FilterDirectiveProvider(
    base_url="https://platform-product.us1.plainid.io",
    client_id="your_client_id",
    client_secret="your_client_secret",
)

3. Create the MultiStoreRetriever.

retriever = MultiStoreRetriever(
    filter_provider=filter_provider,
    resource_types=["customer"],
    vector_stores=[customer_store],
    k=4,
)

4. Create the RetrievalNode.

retrieval_node = RetrievalNode(retriever=retriever)

This node:

• Reads from state["retrieval"]["query"].
• Writes to state["retrieval"]["retrieved_documents"].

Build Graph

Compose the nodes into a StateGraph.

To build the graph:

1. Import the required classes.

from langgraph.graph import START, END, StateGraph
from core_plainid.models.context.request_context import RequestContext

2. Create the graph.

graph = StateGraph(AgentState)

3. Add the nodes.

graph.add_node("categorization", categorization_node)
graph.add_node("anonymizer", anonymizer_node)
graph.add_node("retrieval", retrieval_node)

4. Add the edges.

graph.add_edge(START, "categorization")
graph.add_edge("categorization", "anonymizer")
graph.add_edge("anonymizer", "retrieval")
graph.add_edge("retrieval", END)

5. Compile the graph.

app = graph.compile()

Execute Workflow

Invoke the graph with a Request Context and input queries.

To execute the workflow:

1. Create the RequestContext.

request_context = RequestContext(
    entity_id="your_entity_id",
    entity_type_id="your_entity_type",
    additional_identities=[
        AdditionalIdentity(
            entity_id="your_additional_entity_id",
            entity_type_id="your_additional_entity_type",
        ),
    ],
)

2. Invoke the graph.

result = await app.ainvoke({
    "request_context": request_context,
    "categorization": {"query": "What is John Smith's contract status?"},
    "anonymization": {"query": "What is John Smith's contract status?"},
    "retrieval": {"query": "What is John Smith's contract status?"},
})

3. Access the results.

print(result["anonymization"]["output_text"])
print(result["retrieval"]["retrieved_documents"])

Routing Between Nodes

Nodes can define the next execution step using next_node.

To route between nodes:

1. Configure the CategorizationNode with next_node.

categorization_node = CategorizationNode(
    categorizer=categorizer,
    next_node="anonymizer",
)

2. Configure the AnonymizerNode with next_node.

anonymizer_node = AnonymizerNode(
    anonymizer=anonymizer,
    next_node="retrieval",
)

This allows the graph to dynamically control execution flow without explicitly defining all edges.

Architecture Summary

The PlainID and LangGraph integration introduces a Policy enforcement layer embedded into the agent graph execution model. It includes a:

• State Layer (AgentState)
  A shared, structured state object carries identity, inputs, and outputs across all nodes.

• Policy Decision Layer (PlainID)
  Centralized authorization engine evaluates:

  • Prompt intent (categorization).
  • Data sensitivity (anonymization).
  • Data access (retrieval filters).

• Execution Layer (LangGraph Nodes)
  Each node enforces a specific control point:

  • Categorization Node → governs what can be asked.
  • Anonymization Node → governs what data is exposed.
  • Retrieval Node → governs what data can be accessed.

• Flow Control Layer (Graph Orchestration)
  LangGraph manages execution paths, while PlainID Policies influence:

  • Whether execution continues.
  • What transformations occur.
  • What data is returned.

Error Handling

When next_node_on_error is set, errors are caught and the graph routes to the specified error handler node. The error details are available in the sub-state.

To handle errors:

1. Define the error handler.

def error_handler(state: AgentState) -> dict:
    for key in ["categorization", "anonymization", "retrieval"]:
        sub_state = state.get(key)
        
        if sub_state and sub_state.get("error_details"):
            error_details = sub_state["error_details"]
            
            print(f"Error in {key}: {error_details['error_message']}")
            print(f"Exception: {error_details['error']}")

    return {}

2. Add the error handler node to the graph.

graph.add_node("error_handler", error_handler)