RAG security SDK

The RAG surface submits a query and retrieved chunks to POST /api/rag-security/evaluate. It does not retrieve or embed documents for you; it evaluates data produced by your existing RAG stack.

Chunk contract

from deepintshield import RetrievedChunk, build_chunk

chunk = build_chunk(
    content="Refunds take five business days.",
    chunk_id="chunk-17",
    document_id="refund-policy",
    document_version="v3",
    source_id="support-kb",
    source_name="Support knowledge base",
    trust_score=95,
    acl_tags=["support"],
    labels=["policy"],
)

RetrievedChunk also supports offsets, source health, injection score, PII flags, quarantine state, and arbitrary metadata. to_payload() fills a missing or non-positive offset_end with len(content) and omits empty metadata.

Chunk IDs must be stable and unique within the evaluated result set. Filtering joins the response to input chunks by chunk_id; duplicate IDs make that join ambiguous.

Evaluate and filter

allowed, raw = shield.rag.filter(
    query="How long do refunds take?",
    chunks=[chunk],
    source_id="support-kb",
    requester="alice@example.com",
    requester_role="support",
    metadata={"request_id": "req-42"},
)

evaluate() accepts RetrievedChunk objects or mappings and returns the raw gateway dictionary. filter() accepts RetrievedChunk objects and returns (allowed_chunks, raw_response) while preserving input order.

The helper considers a chunk allowed only when its ID occurs in result.trace.retrieved_chunks. If that trace or its IDs are absent, the filtered list is empty. This is intentionally conservative, but you should alert on an unexpected empty trace rather than treating it as an ordinary “nothing matched” result.

filter() returns the original chunk objects. It does not replace their content with a redacted copy from the response. If policy requires content redaction, consume the reviewed sanitized response field or apply a separate redaction step before assembling the prompt.

The standalone helpers are also public:

from deepintshield import allowed_chunk_ids, filter_chunks

ids = allowed_chunk_ids(raw)
survivors = filter_chunks([chunk], raw)

Wrap a retriever

retriever = shield.rag.guard_retriever(
    vectorstore.as_retriever(),
    source_id="support-kb",
)

documents = retriever.invoke("How long do refunds take?")

The wrapper mutates the object in place and wraps the first available method in this priority order:

1. invoke
2. retrieve
3. _get_relevant_documents
4. get_relevant_documents

It supports list/tuple results from synchronous methods. Empty or non-sequence results pass through. It reads document content from page_content by default, then text, then str(document). Metadata keys default to chunk_id and document_id; missing chunk IDs fall back to the document index.

Use chunk_mapper(index, document) when your document type or identifiers do not match those defaults:

def map_document(index, doc):
    return build_chunk(
        content=doc.body,
        chunk_id=doc.id,
        document_id=doc.parent_id,
        acl_tags=doc.permissions,
    )

shield.rag.guard_retriever(retriever, chunk_mapper=map_document)

Caution

SDK 2.5.1 does not await async retrievers. Do not wrap ainvoke or another coroutine-returning retrieval method with this helper; evaluate its results explicitly after awaiting, or use a synchronous adapter.

Wrap an embedder

shield.rag.guard_embedder(
    embedder,
    stage="input",
    raise_on_block=True,
)

vectors = embedder.embed_documents(texts)

The SDK wraps these synchronous LangChain/LlamaIndex method names when present:

• embed_documents
• embed_query
• get_text_embedding
• get_text_embedding_batch
• get_query_embedding

Each string is checked before the original method runs. A batch performs one guardrail network request per string, so latency and request volume scale with batch size. If you need bulk evaluation, rate limiting, or async ingestion, design that orchestration explicitly rather than assuming the wrapper batches checks.

Entitlement and operational behavior

The Python types and wrappers ship in the core package. The gateway’s RAG evaluation endpoint is a Team-or-higher runtime feature and can return a feature-locked error when the authenticated workspace is not entitled. See the error catalog and RAG Security console guide.

Test filtering against poisoned, PII-bearing, unauthorized, stale, missing-ID, duplicate-ID, and empty-result inputs. Detector scores and policy decisions are security signals, not proof of factual correctness; keep source authorization, index hygiene, citations, and answer-grounding evaluation in the pipeline.