langchain-plainid 0.3.1

PlainID LangChain library

langchain_plainid

PlainID for LangChain. Library which helps you to integrate PlainID with LangChain.

Installation

Based on your environment, you can install the library using pip:

pip install langchain_plainid

Setup with PlainID

Once you have installed the library, you can set up PlainID access.

1. Retrieve your PlainID credentials to access the platform - client ID and client secret.
2. Find you PlainID base URL. For productiщn platform you can use https://platform-product.us1.plainid.io.

Note URL starts from platform-product.

These are 3 parameters you need to use with the library.
Note Please don't share your credentials with anyone, don't store them in your code. Use environment variables or secret management tools to store them.

Exceptions

The library provides specific PlainID exceptions to help identify which module caused an error during chain execution. These exceptions contain the exact PlainID error message and the component that raised it.

• PlainIDException: Base exception for all PlainID errors.
• PlainIDClientException: Errors in the PlainID client.
• PlainIDPermissionsException: Errors in permissions processing.
• PlainIDFilterException: Errors in filter processing.
• PlainIDRetrieverException: Errors in the retriever components.
• PlainIDCategorizerException: Errors in the categorizer component.
• PlainIDAnonymizerException: Errors in the anonymizer component.
• PlainIDSQLAuthorizerException: Errors in the SQL Authorizer component.

Category filtering

To use category filtering with this library, you need to setup related ruleset in PlainID.
e.g if we are using categories template name, we need to add the following ruleset:

# METADATA
# custom:
#   plainid:
#     kind: Ruleset
#     name: All
ruleset(asset, identity, requestParams, action) if {
	asset.template == "categories"
}

and setup what categories are available in PlainID through asset types. e.g. add the following assets: contract, HR.

Now it's time to use category filtering in your LangChain application.

	from langchain_plainid import PlainIDCategorizer, PlainIDPermissionsProvider


	permissions_provider = PlainIDPermissionsProvider(
	    base_url="https://platform-product.us1.plainid.io",
	    client_id="your_client_id",
	    client_secret="your_client_secret",
	    entity_id="your_entity_id",
	    entity_type_id="your_entity_type_id",
		plainid_categories_resource_type="categories")

    plainid_categorizer = PlainIDCategorizer(classifier_provider=<classifier>, permissions_provider=permissions_provider, all_categories=["contract", "HR", "finance"])
    chain = plainid_categorizer
    query = "I'd like to know the weather forecast for today"
	result = chain.invoke(f"{query}") # push your prompt to the chain

Categorizer will connect to PlainID and retrieve the list of categories available in your PlainID account. Then it will classify your prompt with provided classifier and pass your query to the next chain element or break execution with PlainIDCategorizerException exception.

The all_categories parameter (optional) allows you to specify the full list of possible categories for classification. If not provided, the classifier will use the allowed categories from permissions.

Category classifiers

We provide 2 classifiers out of the box:

LLMCategoryClassifierProvider

This classifier uses LLM to classify your prompt. It uses langchain LLMs to classify your prompt. You can configure it with any LLM you want.

	from langchain_plainid import LLMCategoryClassifierProvider

	llm_classifier = LLMCategoryClassifierProvider(llm=OllamaLLM(model="llama2"))

Use it with caution, quality of classification depends on the LLM you are using. Some base models could return bad or even wrong results, so use it with big models (OpenAI, Anthropic, etc.) or with models which are trained for classification tasks.

ZeroShotCategoryClassifierProvider

This used LLM model which suits best for classification tasks. During the work it will download the model from HuggingFace and use it to classify your prompt.

	from langchain_plainid import ZeroShotCategoryClassifierProvider

	zeroshot_classifier = ZeroShotCategoryClassifierProvider()

Use it if you want better classification results, but also have free space on your disk, and can wait for the model to be downloaded.

Anonymizer

To use anonymizer with this library, you need to setup related ruleset in PlainID. e.g if we are using entities template name, we need to add the following ruleset:

# METADATA
# custom:
#   plainid:
#     kind: Ruleset
#     name: PERSON
ruleset(asset, identity, requestParams, action) if {
	asset.template == "entities"
	asset["path"] == "PERSON"
	action.id in ["MASK"]
}

We support 2 actions: MASK and ENCRYPT. You can use them to mask or encrypt your data. Data is always masked with *** symbols.

The list of possible anonymization sources are based on PII entities. We are using presidio library from Microsoft to detect PII entities in your text. You can find the list of supported entities here

	from langchain_plainid import PlainIDPermissionsProvider,PlainIDAnonymizer


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

    plainid_anonymizer = PlainIDAnonymizer(permissions_provider=permissions_provider, encrypt_key="your_encryption_key")
    chain = plainid_anonymizer
    query = "What's the name of the person who is responsible for the contract?"
	result = chain.invoke(f"{query}") # push your prompt to the chain

Anonymizer will connect to PlainID and retrieve the list of categories available in your PlainID account. Then it will classify your text and anonymize it. Processed text will be passed the next chain element or break execution with PlainIDAnonymizerException exception. Exception will be raised if there are some problems, or misalignment in your PlainID ruleset.

Tools authorization

To use tools authorization with this library, you need to setup related ruleset in PlainID.
e.g if we are using tools_v template name, we need to add the following ruleset:

# METADATA
# custom:
#   plainid:
#     kind: Ruleset
#     name: All
ruleset(asset, identity, requestParams, action) if {
	asset.template == "tools_v"
}

and setup what tools are available in PlainID through asset types. e.g. add the following assets: search_tool, calculator, email_sender.

Now it's time to use tools authorization in your LangChain application.

	from langchain_plainid import PlainIDPermissionsProvider

	permissions_provider = PlainIDPermissionsProvider(
	    base_url="https://platform-product.us1.plainid.io",
	    client_id="your_client_id",
	    client_secret="your_client_secret",
	    entity_id="your_entity_id",
	    entity_type_id="your_entity_type_id",
	    plainid_tools_resource_type="tools_v")

	allowed_tools = permissions_provider.get_allowed_tools()
	# Use allowed_tools to filter available tools in your LangChain agent

Power of creating your chains with PlainID's internal category filtering and anonymization

As a result you can add something like this to your processing chain:

	chain = plainid_categorizer | llm | vector_store | anonymizer | output_parser

This will allow you to filter your data based on categories and anonymize it before passing to the next chain element. You can use any chain element you want, and it will work with PlainID's internal category filtering and anonymization.

PlainID retriever

To use category filtering with this library, you need to setup related policies in PlainID.
e.g if we are using customer template name, we need to add the following ruleset to filter data based on country metadata and some test_num field:

# METADATA
# custom:
#   plainid:
#     kind: Ruleset
#     name: rs1
ruleset(asset, identity, requestParams, action) if {
	asset.template == "customer"
	asset["country"] == "Sweden"
	asset["country"] != "Russia"
	contains(asset["country"], "we")
	startswith(asset["country"], "Sw")
}

# METADATA
# custom:
#   plainid:
#     kind: Ruleset
#     name: rs1
ruleset(asset, identity, requestParams, action) if {
	asset.template == "customer"
	asset["country"] in ["aaa", "bbb"]
	asset["age"] <= 11111
	endswith(asset["country"], "wwww")
}

Note that you need to add country and age parameters to your vector store as metadata. This is what PlainID will use to filter your data.

	from langchain_community.vectorstores import Chroma
	from langchain_core.documents import Document
	from langchain_plainid import PlainIDRetriever

	 docs = [
            Document(
                page_content="Stockholm is the capital of Sweden.",
                metadata={"country": "Sweden", "age": 5},
            ),
            Document(
                page_content="Oslo is the capital of Norway.",
                metadata={"country": "Norway", "age": 5},
            ),
            Document(
                page_content="Copenhagen is the capital of Denmark.",
                metadata={"country": "Denmark", "age": 5},
            ),
            Document(
                page_content="Helsinki is the capital of Finland.",
                metadata={"country": "Finland", "age": 5},
            ),
            Document(
                page_content="Malmö is a city in Sweden.",
                metadata={"country": "Sweden", "age": 5},
            ),
        ]

	vector_store = Chroma.from_documents(documents, embeddings)
    plainid_retriever = PlainIDRetriever(vectorstore=vector_store, filter_provider=filter_provider, k=4)
    docs = plainid_retriever.invoke("What is the capital of Sweden?")

The `k` parameter limits the number of documents returned (default 4).

PlainID filter provider

Filter provider is used to connect to PlainID and retrieve the list of categories available in your PlainID account. The following parameters are required:

base_url (str): Base URL for PlainID service
client_id (str): Client ID for authentication
client_secret (str): Client secret for authentication
entity_id (str): Entity ID for the request
entity_type_id (str): Entity type ID for the request
resource_type (str): Resource type for the request

Supported vector stores and limitations

We support different vector stores, but some of them have limitations in filtering or querying data. Below is the list of tested vector stores and their limitations (list of not supported PlainID operators).

FAISS

It doesn't support STARTSWITH, ENDSWITH, CONTAINS operators.

Chroma

It doesn't support IN, NOT_IN, STARTSWITH, ENDSWITH, CONTAINS operators.

SQL Database Authorizer

The SQL Database Authorizer dynamically modifies SQL queries based on PlainID authorization policies, enforcing Row-Level Security (RLS) and Column-Level Security (CLS) at query time.

Setup

You need the following from your PlainID SQL Authorizer deployment:

• Base URL of the SQL Authorizer service (e.g. https://your-sql-authz.plainid.cloud)
• Client ID for authentication
• Client Secret or a JWT token for authentication (at least one is required)

Authentication

The client supports two authentication modes:

1. Client credentials — provide client_id and client_secret at construction time.
2. JWT token — provide an auth_token per request. The Bearer prefix is added automatically if missing.

Basic usage

from langchain_plainid import (
    PlainIDSQLAuthorizer,
    SQLAuthorizerFlags,
    SQLAuthorizerRequest,
)
from langchain_plainid.models.sql_authorizer_models import PoliciesJoinOperation

sql_authorizer = PlainIDSQLAuthorizer(
    base_url="https://your-sql-authz.plainid.cloud",
    client_id="your_client_id",
    client_secret="your_client_secret",
)

request = SQLAuthorizerRequest(
    sql="SELECT * FROM accounts WHERE country = 'US'",
    entity_id="your_entity_id",
    entity_type_id="User",
    flags=SQLAuthorizerFlags(
        empty_rls_treat_as_denied=True,
        empty_cls_treat_as_permitted=True,
        expand_star_column=True,
        policies_join_operation=PoliciesJoinOperation.OR,
    ),
)

response = sql_authorizer.authorize_sql(request)
print(response.sql)           # The modified SQL query
print(response.was_modified)  # True if policies were applied
print(response.error)         # Empty string if no errors

Using a JWT token instead of client secret

sql_authorizer = PlainIDSQLAuthorizer(
    base_url="https://your-sql-authz.plainid.cloud",
    client_id="your_client_id",
)

response = sql_authorizer.authorize_sql(request, auth_token="your_jwt_token")

Request parameters

Parameter	Type	Required	Description
sql	str	Yes	The SQL query to enforce
entity_id	str	No	The ID of the entity
entity_type_id	str	No	The entity type ID (Identity Template)
user	str	No	User identification string
flags	SQLAuthorizerFlags	No	Modification flags (RLS/CLS behavior)
entity_attributes	dict	No	Additional entity attributes
context_data	dict	No	Context data for the request
environment	dict	No	Environment data for the request
runtime_fine_tune	SQLAuthorizerRuntimeFineTune	No	Optional PDP resolution parameters

Flags

Flag	Type	Description
empty_rls_treat_as_denied	bool	Treat empty row-level security as denied
empty_cls_treat_as_permitted	bool	Treat empty column-level security as permitted
ignore_runtime_cls_response	bool	Ignore runtime CLS response
expand_star_column	bool	Expand * columns into specific fields
opposite_column_filtering_behavior	bool	Use opposite column filtering behavior
policies_join_operation	PoliciesJoinOperation	Join operation for policies (OR or AND)
runtime_cls_as_masked	bool	Treat runtime CLS as masked
columns_resource_type	str	Resource type for columns

Error handling

All errors are wrapped in PlainIDSQLAuthorizerException, including HTTP errors, validation errors, and connection failures.

from langchain_plainid import PlainIDSQLAuthorizerException

try:
    response = sql_authorizer.authorize_sql(request)
except PlainIDSQLAuthorizerException as e:
    print(f"SQL Authorizer error: {e.message}")