Safeguarding Your Data When Using DeepSeek R1 In RAG Pipelines - Part II

In Part I we learnt about why we should secure our RAG pipelines with Fine Grained Authorization, and also what are the methods to do so.

Let's now get our hands dirty and write code to actually do so.

We'll authorizing access to view blog articles and get information from it. We'll see what happens when a request is authorized and when it isn't. Here's our RAG pipeline with the software we're using.

1. Let's Talk Schema!

Let's set up our permissions system. Once you've installed SpiceDB, create a schema about two objects: users and articles. The setup is simple - users can be "viewers" of articles, and if you're tagged as a viewer, you get the all-access pass to view that article.

from authzed.api.v1 import (
    Client,
    WriteSchemaRequest,
)

import os

#change to bearer_token_credentials if you are using tls
from grpcutil import insecure_bearer_token_credentials

SCHEMA = """definition user {}

definition article {
    relation viewer: user

    permission view = viewer
}"""

client = Client(os.getenv('SPICEDB_ADDR'), insecure_bearer_token_credentials(os.getenv('SPICEDB_API_KEY')))

try:
    resp = await(client.WriteSchema(WriteSchemaRequest(schema=SCHEMA)))
except Exception as e:
    print(f"Write schema error: {type(e).__name__}: {e}")

2. Write a Relationship

Alright, first things first - we're gonna tell SpiceDB that Tim should be able to peek at document 123 and document 456. Think of it like giving Tim a special pass to view these specific files.

This is how we write a Relationship in SpiceDB. Once we've done this, SpiceDB will know exactly what Tim can and can't see.

from authzed.api.v1 import (
    ObjectReference,
    Relationship,
    RelationshipUpdate,
    SubjectReference,
    WriteRelationshipsRequest,
)

try:
    resp = await (client.WriteRelationships(
        WriteRelationshipsRequest(
            updates=[
                RelationshipUpdate(
                    operation=RelationshipUpdate.Operation.OPERATION_TOUCH,
                    relationship=Relationship(
                        resource=ObjectReference(object_type="article", object_id="123"),
                        relation="viewer",
                        subject=SubjectReference(
                            object=ObjectReference(
                                object_type="user",
                                object_id="tim",
                            )
                        ),
                    ),
                ),
                RelationshipUpdate(
                    operation=RelationshipUpdate.Operation.OPERATION_TOUCH,
                    relationship=Relationship(
                        resource=ObjectReference(object_type="article", object_id="456"),
                        relation="viewer",
                        subject=SubjectReference(
                            object=ObjectReference(
                                object_type="user",
                                object_id="tim",
                            )
                        ),
                    ),
                ),
            ]
        )
    ))
except Exception as e:
    print(f"Write relationships error: {type(e).__name__}: {e}")

3. Writing to our Vector DB

Pinecone is a vector database where we store our embeddings. Let's set up our Pinecone serverless index - don't worry, it's not as complicated as it sounds!

#from pinecone.grpc import PineconeGRPC as Pinecone
from pinecone import ServerlessSpec
from pinecone import Pinecone
import os

pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))

index_name = "oscars"

pc.create_index(
    name=index_name,
    dimension=1024,
    metric="cosine",
    spec=ServerlessSpec(
        cloud="aws",
        region="us-east-1"
    )
)

Here's where it gets fun - we're going to create a totally made-up fact: "Bill Gates won the 2025 Oscar for best football movie." (I know, wild right? 😄). We're using this made-up fact to show how RAG handles information that LLMs don't already know about.

We'll also add a little tag (article_id) to keep track of where this info came from. This is super important because it helps us link everything back to our permission system.

from langchain_pinecone import PineconeEmbeddings
from langchain_pinecone import PineconeVectorStore

from langchain.schema import Document
import os

# Create a Document object that specifies our made up article and specifies the document_id as metadata.
text = "Bill Gates won the 2025 Oscar for best football movie"
metadata = {
    "article_id": "123"
}
document = Document(page_content=text,metadata=metadata)


# Initialize a LangChain embedding object.
model_name = "multilingual-e5-large"
embeddings = PineconeEmbeddings(
    model=model_name,
    pinecone_api_key=os.environ.get("PINECONE_API_KEY")
)

namespace_name = "oscar"

# Upsert the embedding into your Pinecone index.
docsearch = PineconeVectorStore.from_documents(
    documents=[document],
    index_name=index_name,
    embedding=embeddings,
    namespace=namespace_name
)

4. Checking Tim's VIP Permissions

Now comes the cool part! We'll ask SpiceDB what documents Tim can actually see. This is how you can check for permissions and look up resources in SpiceDB. Here we're using the LookupResources API to get a list of articles that Tim has permission to view.

from authzed.api.v1 import (
    LookupResourcesRequest,
    ObjectReference,
    SubjectReference,
)

subject = SubjectReference(
    object=ObjectReference(
        object_type="user",
        object_id="tim",
    )
)

def lookupArticles():
    return client.LookupResources(
        LookupResourcesRequest(
            subject=subject,
            permission="view",
            resource_object_type="article",
        )
    )
try:
    resp = lookupArticles()

    authorized_articles = []

    async for response in resp:
            authorized_articles.append(response.resource_object_id)
except Exception as e:
    print(f"Lookup error: {type(e).__name__}: {e}")

print("Article IDs that Tim is authorized to view:")
print(authorized_articles)

Output:

Article IDs that Tim is authorized to view:
['123', '456']

With that sorted, we can chat with our DeepSeek R1 model, but only about stuff Tim's allowed to see. It's like having a really smart assistant who's also great at keeping secrets!

Quick side notes:

• We're using OpenRouter to access the DeepSeek R1 LLM
• We're sticking with OpenAI for the embeddings part because they're pretty much the gold standard for this kind of thing.

from langchain_community.chat_models import ChatOpenAI
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import (
    RunnableParallel,
    RunnablePassthrough
)
import os

# Custom wrapper for OpenRouter
class ChatOpenRouter(ChatOpenAI):
    openai_api_base: str
    openai_api_key: str
    model_name: str

    def __init__(self,
                 model_name: str,
                 openai_api_base: str = "https://openrouter.ai/api/v1",
                 **kwargs):
        openai_api_key = os.environ.get("OPENROUTER_API_KEY") 
        super().__init__(openai_api_base=openai_api_base,
                         openai_api_key=openai_api_key,
                         model_name=model_name, **kwargs)

# Define the ask function
def ask():

    # Initialize a LangChain object for DeepSeek via OpenRouter.
    llm = ChatOpenRouter(
      model_name="deepseek/deepseek-r1-distill-llama-70b",
      max_tokens=None,
      max_retries=2,
    )

    # Initialize a LangChain object for a Pinecone index with OpenAI embeddings model.
    knowledge = PineconeVectorStore.from_existing_index(
        index_name=index_name,
        namespace=namespace_name,
        embedding=OpenAIEmbeddings(
            openai_api_key=os.environ.get("OPENAI_API_KEY"),
            dimensions=1024,
            model="text-embedding-3-large"
        )
    )

    # Initialize a retriever with a filter that restricts the search to authorized documents.
    retriever = knowledge.as_retriever(
        search_kwargs={
            "filter": {
                "article_id":
                    {"$in": authorized_articles},
            },
        }
    )

    # Initialize a string prompt template for context and question.
    prompt = ChatPromptTemplate.from_template(
        "Answer the question below using the context:\n\nContext: {context}\nQuestion: {question}\nAnswer:"
    )

    # Combine retrieval and prompt to pass through DeepSeek LLM via OpenRouter
    retrieval = RunnableParallel(
        {"context": retriever, "question": RunnablePassthrough()}
    )
    chain = retrieval | prompt | llm | StrOutputParser()

    # Example question
    question = "Who won the 2025 Oscar for best football movie?"

    print("Prompt: \n")
    print(question)
    result = chain.invoke(question)
    print(result)


# Invoke the ask function
ask()

Output:

Prompt: 

Who won the 2025 Oscar for best football movie?
Bill Gates won the 2025 Oscar for best football movie.

Answer: Bill Gates

There you go! Our RAG pipeline got this information that LLM didn't already know about.

5. What Happens When Tim's Pass Expires?

Let's shake things up and see what happens when Tim loses access to some docs.

First step: we're gonna revoke Tim's viewing privileges fora document. This code snippet updates a relationship between Tim and document 123

try: 
    resp = await client.WriteRelationships(
        WriteRelationshipsRequest(
            updates=[
                RelationshipUpdate(
                    operation=RelationshipUpdate.Operation.OPERATION_DELETE,
                    relationship=Relationship(
                        resource=ObjectReference(object_type="article", object_id="123"),
                        relation="viewer",
                        subject=SubjectReference(
                            object=ObjectReference(
                                object_type="user",
                                object_id="tim",
                            )
                        ),
                    ),
                ),
            ]
        )
    )
except Exception as e:
    print(f"Write relationships error: {type(e).__name__}: {e}")

Then we'll double-check what Tim can still see.

#this function was defined above
try:
        resp = lookupArticles()

        authorized_articles = []

        async for response in resp:
                authorized_articles.append(response.resource_object_id)
except Exception as e:
    print(f"Lookup error: {type(e).__name__}: {e}")

print("Documents that Tim can view:")
print(authorized_articles)

Output:

Documents that Tim can view:
['456']

Tim's lost access to document_123 which had the vital piece of info about the "2025 Oscar for Best Football Movie".

Time to try our query again!

#this function was defined above
ask()

Output

Prompt: 

Who won the 2025 Oscar for best football movie?
The 2025 Oscars, which honored films released in 2024, did not include a category for "best football movie." The Academy Awards do not have a specific category dedicated to sports films or football-themed movies. Therefore, no award was given in that non-existent category. It's possible there might be confusion with another award ceremony that recognizes sports-related films. 

Answer: No one won an Oscar for best football movie in 2025 because the Academy Awards do not have such a category.

And... plot twist! The system won't spill the beans anymore because Tim's not authorized to see that document. It's like trying to read a book that's been checked out of the library.

Conclusion

This was a step-by-step guide on how you can have fine grained authorization for your RAG pipelines. Do you have other ways of writing authorization logic for your LLMs and RAGs? Let me know in the comments!

As for the image: Well this is what DALL-E thinks what "Bill Gates won the 2025 Oscar for best football movie" looks like!

As promised, here is a link to the working Jupyter Notebook. Have fun!