GLLM Privacy

Description

A library to protect Personal Identifiable Information (PII) in a Generative AI project.

Installation

Prerequisites

Mandatory:

1. Python 3.11+ — Install here
2. pip — Install here
3. uv — Install here

Extras (required only for Artifact Registry installations):

1. gcloud CLI (for authentication) — Install here, then log in using:

   gcloud auth login
   

Option 1: Install from Artifact Registry

This option requires authentication via the gcloud CLI.

uv pip install \
  --extra-index-url "https://oauth2accesstoken:$(gcloud auth print-access-token)@glsdk.gdplabs.id/gen-ai-internal/simple/" \
  gllm-privacy

Option 2: Install from PyPI

This option requires no authentication. However, it installs the binary wheel version of the package, which is fully usable but does not include source code.

uv pip install gllm-privacy-binary

Local Development Setup

Prerequisites

1. Python 3.11+ — Install here

2. pip — Install here

3. uv — Install here

4. gcloud CLI — Install here, then log in using:

   gcloud auth login
   

5. Git — Install here

6. Access to the GDP Labs SDK GitHub repository

1. Clone Repository

git clone git@github.com:GDP-ADMIN/gl-sdk.git
cd gl-sdk/libs/gllm-privacy

2. Setup Authentication

Set the following environment variables to authenticate with internal package indexes:

export UV_INDEX_GEN_AI_INTERNAL_USERNAME=oauth2accesstoken
export UV_INDEX_GEN_AI_INTERNAL_PASSWORD="$(gcloud auth print-access-token)"
export UV_INDEX_GEN_AI_USERNAME=oauth2accesstoken
export UV_INDEX_GEN_AI_PASSWORD="$(gcloud auth print-access-token)"

3. Quick Setup

Run:

make setup

4. Activate Virtual Environment

source .venv/bin/activate

Local Development Utilities

The following Makefile commands are available for quick operations:

Install uv

make install-uv

Install Pre-Commit

make install-pre-commit

Install Dependencies

make install

Update Dependencies

make update

Run Tests

make test

Run unit tests and model-free integration tests without downloading neural checkpoints:

uv run pytest

The download-backed Indonesian model tests are opt-in. To include them (requires network access and time to download the checkpoint and initialize its inference backend):

RUN_REAL_MODEL_TESTS=1 uv run pytest

Use uv run pytest -rs to display skipped-test requirements.

Usage

from gllm_privacy.pii_detector import TextAnalyzer, TextAnonymizer
from gllm_privacy.pii_detector.constants import Entities
from gllm_privacy.pii_detector.anonymizer import Operation
from asyncio import run

text = """
    contoh nomor ktp 3525011212941001
    repeat nomor ktp 3525011212941001
    contoh email john.doe@example.com
    contoh nomor telepon +628121729819 dan 0812898029384.
    contoh npwp 01.123.456.7-891.234
"""
text_analyzer = TextAnalyzer()
entities = [Entities.EMAIL_ADDRESS, Entities.KTP, Entities.NPWP, Entities.PHONE_NUMBER]

text_anonymizer = TextAnonymizer(text_analyzer)
anonymized_text = run(text_anonymizer.run(text=text, entities=entities))
print(anonymized_text)

deanonymized_text = run(text_anonymizer.run(text=text, entities=entities, operation=Operation.DEANONYMIZE))
print(deanonymized_text)

If you need to detect person, organization, or location entities in text written in Bahasa Indonesia, you can use either TransformersRecognizer or ProsaRemoteRecognizer. To use the TransformersRecognizer, you can use it like this:

from gllm_privacy.pii_detector.recognizer.config import CAHYA_BERT_CONFIGURATION
from gllm_privacy.pii_detector.recognizer.transformers_recognizer import TransformersRecognizer
from gllm_privacy.pii_detector import TextAnalyzer, TextAnonymizer
from gllm_privacy.pii_detector.constants import Entities

# Load the model, if you run it for the first time, it will download the model from the Hugging Face model hub
transformers_recognizer = TransformersRecognizer(
  model_path=CAHYA_BERT_CONFIGURATION.get("DEFAULT_MODEL_PATH"),
  supported_entities=CAHYA_BERT_CONFIGURATION.get("PRESIDIO_SUPPORTED_ENTITIES"),
)
transformers_recognizer.load_transformer(**CAHYA_BERT_CONFIGURATION)
analyzer = TextAnalyzer(additional_recognizers=[transformers_recognizer])

text = "John Doe adalah seorang karyawan PT ABCD yang berlokasi di Jakarta."
text_analyzer = TextAnalyzer(additional_recognizers=[transformers_recognizer])
entities = [Entities.PERSON, Entities.LOCATION]

text_anonymizer = TextAnonymizer(text_analyzer)
anonymized_text = text_anonymizer.anonymize(text=text, entities=entities)
print(anonymized_text)

deanonymized_text = text_anonymizer.deanonymize(text=text)
print(deanonymized_text)

Detecting Brand Names

BRAND_NAME is a supported entity. Like PERSON, it is detected by a model-backed recognizer instead of a built-in regex pattern. Point the model's brand label at BRAND_NAME in MODEL_TO_PRESIDIO_MAPPING, add BRAND_NAME to supported_entities, then request it during analysis or anonymization:

from gllm_privacy.pii_detector import TextAnalyzer, TextAnonymizer
from gllm_privacy.pii_detector.constants import Entities
from gllm_privacy.pii_detector.recognizer.config import CAHYA_BERT_CONFIGURATION
from gllm_privacy.pii_detector.recognizer.transformers_recognizer import TransformersRecognizer

configuration = {
    **CAHYA_BERT_CONFIGURATION,
    "PRESIDIO_SUPPORTED_ENTITIES": [*CAHYA_BERT_CONFIGURATION["PRESIDIO_SUPPORTED_ENTITIES"], "BRAND_NAME"],
    "MODEL_TO_PRESIDIO_MAPPING": {**CAHYA_BERT_CONFIGURATION["MODEL_TO_PRESIDIO_MAPPING"], "BRD": "BRAND_NAME"},
}

brand_recognizer = TransformersRecognizer(
    model_path="<YOUR_BRAND_NER_MODEL>",
    supported_entities=configuration["PRESIDIO_SUPPORTED_ENTITIES"],
)
brand_recognizer.load_transformer(**configuration)

text = "Saya memakai sepatu Nike."
text_analyzer = TextAnalyzer(additional_recognizers=[brand_recognizer])
text_anonymizer = TextAnonymizer(text_analyzer, add_default_faker_operators=True)
entities = [Entities.BRAND_NAME]

anonymized_text = text_anonymizer.anonymize(text=text, entities=entities)
print(anonymized_text)

print(text_anonymizer.deanonymize(text=anonymized_text))

With add_default_faker_operators=True, BRAND_NAME values are pseudo-anonymized with Faker company names (fake.company()). Without it, they are replaced by the default <BRAND_NAME_n> placeholder.

The brand recognizer also accepts LINE_BREAK_TOKEN and CHUNK_SIZE:

• LINE_BREAK_TOKEN (str | None, default None): token the model was trained on to represent a line break (for example "→"). When set, every "\n" in the input is replaced by " <token> " before inference and detected offsets are mapped back to the original text. Newlines remain intact when the model predicts separate entities per line; this is not enforced by the mapper. Leave it unset for models trained without a break token — their input is passed through unchanged. A literal token already present in the text is preserved and may split a model span. Conversely, a model span crossing an inserted token includes the original newline, so masking that span removes the line break. Such spans are not split into per-line results.
• CHUNK_SIZE (int): maximum tokens per chunk, including special tokens. Set it to the model's real limit (for example 512 for XLM-R based checkpoints) so long multi-line documents are chunked instead of truncated.

Enhanced TransformersRecognizer with Optimum

The TransformersRecognizer supports Hugging Face Optimum for faster inference. Choose the backend with OPTIMUM_BACKEND in the configuration passed to load_transformer.

Available Backends:

When the requested backend is not available, the recognizer logs a warning and falls back: onnx and cuda fall back to transformers, cuda without a GPU falls back to onnx, openvino falls back to onnx, then to transformers, and mps without Apple Silicon falls back to the standard CPU pipeline.

Configuration Options:

config = {
    **CAHYA_BERT_CONFIGURATION,
    "OPTIMUM_BACKEND": "auto",              # "auto", "onnx", "cuda", "openvino", "mps", "transformers"
    "OPTIMUM_DEVICE": "auto",               # "auto", "cuda", "mps", "cpu"
    "ONNX_INTRA_OP_THREADS": 0,             # 0 lets ONNX Runtime decide
    "ONNX_INTER_OP_THREADS": 0,             # 0 lets ONNX Runtime decide
    "ONNX_ENABLE_CPU_MEM_ARENA": True,
    "OPENVINO_PERFORMANCE_HINT": "THROUGHPUT",  # "THROUGHPUT" or "LATENCY"
    "OPENVINO_NUM_STREAMS": "AUTO",
    "OPENVINO_INFERENCE_NUM_THREADS": 0,    # 0 lets OpenVINO decide
}

Usage Example:

from gllm_privacy.pii_detector import TextAnalyzer
from gllm_privacy.pii_detector.recognizer.config import CAHYA_BERT_CONFIGURATION
from gllm_privacy.pii_detector.recognizer.transformers_recognizer import TransformersRecognizer

config = {**CAHYA_BERT_CONFIGURATION, "OPTIMUM_BACKEND": "onnx"}

transformers_recognizer = TransformersRecognizer(
    model_path=config["DEFAULT_MODEL_PATH"],
    supported_entities=config["PRESIDIO_SUPPORTED_ENTITIES"],
)
transformers_recognizer.load_transformer(**config)

pipeline_info = transformers_recognizer.get_pipeline_info()
print(f"Backend: {pipeline_info['backend']}")
print(f"Device: {pipeline_info['device']}")
print(f"Optimizations: {pipeline_info['optimizations']}")

# Use as before
analyzer = TextAnalyzer(additional_recognizers=[transformers_recognizer])

To use the ProsaRemoteRecognizer, you can use it like the following example. Please replace <PROSA_API_URL> and <PROSA_API_KEY> with the valid values.

from gllm_privacy.pii_detector.recognizer.prosa_remote_recognizer import ProsaRemoteRecognizer
from gllm_privacy.pii_detector import TextAnalyzer, TextAnonymizer
from gllm_privacy.pii_detector.constants import Entities

text = "John Doe adalah seorang karyawan PT ABCD yang berlokasi di Jakarta."
prosa_recognizer = ProsaRemoteRecognizer('<PROSA_API_URL>', '<PROSA_API_KEY>')
text_analyzer = TextAnalyzer(additional_recognizers=[prosa_recognizer])
entities = [Entities.PERSON, Entities.LOCATION]

text_anonymizer = TextAnonymizer(text_analyzer)
anonymized_text = text_anonymizer.anonymize(text=text, entities=entities)
print(anonymized_text)

deanonymized_text = text_anonymizer.deanonymize(text=text)
print(deanonymized_text)

Error Handling

Errors raised by gllm-privacy subclass BasePrivacyError, which is built on BaseError from gllm-core. Each error has a stable code, a message that you can show to users, a debug_info dict for logs, and a recovery dict that says whether retrying can succeed. str(error) returns only the message. debug_info never holds the analyzed text, request payloads, response bodies, headers, or URL query strings.

The default messages are written for end users. They do not name the library, the model, or a setting. The developer fix for each code is in the table below. debug_info["reason"] holds the internal cause when no original error was caught.

Branch on code, not on the message. Each error also keeps its previous builtin base, such as ValueError, so existing except clauses keep working.

from gllm_privacy.exceptions import BasePrivacyError, BaseRemoteRecognizerError

try:
    results = text_analyzer.analyze(text=text, language="id", entities=entities)
except BaseRemoteRecognizerError as error:
    # A remote NER service failed. The text was not checked, so do not treat it as PII-free.
    logger.warning("PII detection failed: %s", error.code, extra={"debug_info": error.debug_info})
    if error.recovery["retryable"]:
        # Wait for error.recovery.get("retry_after_seconds", <your backoff>), then retry with your own limits.
        ...
    raise
except BasePrivacyError as error:
    logger.error("PII detection failed: %s", error.code, extra={"debug_info": error.debug_info})
    raise

Remote recognizers (ProsaRemoteRecognizer, GDPLabsNerApiRemoteRecognizer, and GDPLabsNerApiRemoteRecognizerAsync) raise a BaseRemoteRecognizerError subclass when the call fails or the response has an unexpected format, including an entity without offsets. FlairRecognizer raises ModelInferenceError when its model fails. None of them return an empty result in place of a failure.

TextAnalyzer.analyze raises UnsupportedLanguageError when no recognizer supports the language, and InvalidInputError when no recognizer can detect any of the requested entities.

Only the remote timeout, connection, rate limit and server errors are retryable. A rate limit or server error also carries recovery["retry_after_seconds"] when the service sends a Retry-After header. How many times to retry, and the backoff between attempts, is up to the caller.