DEV Community: mmalaika.junaid

Build a Zero-Trust Document Pipeline with OpenCV, Presidio, and LangGraph (Beginner Guide)

mmalaika.junaid — Wed, 07 Oct 2026 10:42:57 +0000

When building LLM applications, we often rush to pass user data directly into our prompts or RAG databases. But what happens when that data is a passport, a driver's license, or a medical record? Sending raw identity documents across an external trust boundary to an LLM API is a massive privacy risk.

In this tutorial, we are building SecureKiosk AI, an open-source "edge gatekeeper." We will use local computer vision and NLP to intercept identity documents, extract the text, and deterministically mask Personally Identifiable Information (PII) before it ever reaches an LLM.

Here is the tech stack we will use:

OpenCV & Tesseract (Computer Vision & OCR): To clean image noise and extract raw text locally.
Microsoft Presidio & spaCy (NLP): To detect and redact sensitive PII in memory.
LangGraph (Orchestration): To manage the pipeline as a resilient state machine.
Streamlit: To build a clean, interactive frontend.

Pipeline Architecture
Rather than relying on linear, brittle Python scripts that crash when text extraction yields poor results, the pipeline operates as a state machine:

The Architecture Pipeline

Let’s get started!

Step 1: Setting Up the Environment

Because we are doing native computer vision and OCR, we need OS-level dependencies alongside our Python packages.

1. System Dependencies:
If you are on Ubuntu/Debian, install the C-libraries for vision and OCR:

sudo apt-get update
sudo apt-get install -y tesseract-ocr libgl1 libglib2.0-0

2. Python Environment:
Create a virtual environment (Python 3.11 is highly recommended for compatibility) and install the core packages.

Pro-Tip: We pin numpy<2.0.0 because OpenCV is currently compiled against NumPy 1.x. Upgrading to NumPy 2.0+ will cause your CV pipeline to crash!

# requirements.txt
numpy<2.0.0
opencv-python-headless==4.9.0.80
pytesseract==0.3.10
presidio-analyzer==2.2.355
presidio-anonymizer==2.2.355
langgraph>=0.0.60
streamlit>=1.35.0
https://github.com/explosion/spacy-models/releases/download/en_core_web_sm-3.7.1/en_core_web_sm-3.7.1-py3-none-any.whl

Step 2: The "Eyes" (OpenCV & Tesseract)

Raw photos of IDs are messy, they have glare, watermarks, and bad lighting. If we feed raw images straight to an OCR engine, the text extraction will fail. We need to preprocess the image using OpenCV.

Create src/cv/processor.py and src/ocr/extractor.py. We will convert the image to grayscale, apply Gaussian blur, and use adaptive binarization to make the text pop.

import cv2
import numpy as np
import pytesseract

def preprocess_image(image_bytes):
    # Convert bytes to numpy array
    nparr = np.frombuffer(image_bytes, np.uint8)
    img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)

    # Grayscale and binarize for crisp OCR
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    blur = cv2.GaussianBlur(gray, (5,5), 0)
    binary = cv2.adaptiveThreshold(blur, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, 11, 2)

    return binary

def extract_text(processed_image):
    # Extract text from the cleaned image
    return pytesseract.image_to_string(processed_image)

Step 3: The "Shield" (Zero-Trust PII Redaction)

Now that we have the raw text, we must scrub it. We will use Microsoft Presidio, powered by spaCy's Named Entity Recognition (NER).

Why Presidio Instead of Regular Expressions?
Regex matches fixed patterns like email addresses or phone numbers, but fails on variable context (e.g., extracting person names, physical addresses, or organizational entities). Presidio combines regex pattern recognition with spaCy Named Entity Recognition (NER), context words, and checksum validation.

Example Transformation

INPUT:
Issued to: Alex Vance
Passport No: P9482014
Contact: +971 50 123 4567
Address: Building 4, Al Maryah Island, Abu Dhabi

OUTPUT:
Issued to: <PERSON>
Passport No: <ALPHANUMERIC_ID>
Contact: <PHONE_NUMBER>
Address: <LOCATION>

Architectural Note: We are specifically configuring Presidio to use the lightweight en_core_web_sm model (~12MB). If you default to the large model (~580MB), your app will likely suffer from Out-Of-Memory (OOM) crashes when deployed on free cloud tiers.

Create src/privacy/redactor.py:

from presidio_analyzer import AnalyzerEngine
from presidio_analyzer.nlp_engine import NlpEngineProvider
from presidio_anonymizer import AnonymizerEngine

class PIIRedactor:
    def __init__(self):
        # Force the lightweight spaCy model for edge deployment
        nlp_config = {
            "nlp_engine_name": "spacy",
            "models": [{"lang_code": "en", "model_name": "en_core_web_sm"}],
        }
        provider = NlpEngineProvider(nlp_configuration=nlp_config)

        self.analyzer = AnalyzerEngine(nlp_engine=provider.create_engine())
        self.anonymizer = AnonymizerEngine()

    def redact(self, text: str) -> str:
        # Detect entities like PERSON, LOCATION, DATE_TIME
        results = self.analyzer.analyze(text=text, entities=[], language='en')
        # Mask them deterministically 
        anonymized = self.anonymizer.anonymize(text=text, analyzer_results=results)
        return anonymized.text

✓ No raw PII sent to LLM

This is the core security guarantee of the pipeline:
all document processing and PII redaction occur locally before any downstream AI system receives the content.

Step 4: The Brain (LangGraph Orchestration)

Instead of a brittle, linear script, we will orchestrate these steps using LangGraph. A state machine provides a foundation for handling OCR failures, retries, and alternate processing paths as the pipeline evolves.

Create src/graph.py:

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# Define what data moves between our nodes
class DocumentState(TypedDict):
    raw_image: bytes
    processed_image: object
    extracted_text: str
    redacted_text: str

def process_vision(state: DocumentState):
    state["processed_image"] = preprocess_image(state["raw_image"])
    return state

def process_ocr(state: DocumentState):
    state["extracted_text"] = extract_text(state["processed_image"])
    return state

def process_redaction(state: DocumentState):
    redactor = PIIRedactor()
    state["redacted_text"] = redactor.redact(state["extracted_text"])
    return state

# Build the pipeline
workflow = StateGraph(DocumentState)
workflow.add_node("vision", process_vision)
workflow.add_node("ocr", process_ocr)
workflow.add_node("redact", process_redaction)

workflow.add_edge(START, "vision")
workflow.add_edge("vision", "ocr")
workflow.add_edge("ocr", "redact")
workflow.add_edge("redact", END)

document_pipeline = workflow.compile()

Step 5: The Interface (Streamlit)

Finally, we wrap our LangGraph pipeline in a clean Streamlit interface. Create app.py:

import streamlit as st
from src.graph import document_pipeline

st.title("🛡️ SecureKiosk AI: Zero-Trust Pipeline")
st.write("Upload an ID to extract and redact PII locally.")

uploaded_file = st.file_uploader("Upload Document (PNG/JPG)", type=["png", "jpg", "jpeg"])

if uploaded_file:
    st.image(uploaded_file, caption="Original Upload", use_column_width=True)

    if st.button("Process Document"):
        with st.spinner("Running localized edge pipeline..."):
            # Pass the image into our LangGraph state machine
            initial_state = {"raw_image": uploaded_file.read()}
            final_state = document_pipeline.invoke(initial_state)

            st.subheader("Raw Extracted Text (DANGER)")
            st.warning(final_state["extracted_text"])

            st.subheader("Redacted Text (SAFE FOR LLMs)")
            st.success(final_state["redacted_text"])

Deploying to Streamlit Community Cloud

When you are ready to share this with the world, push your code to GitHub. To deploy on Streamlit Community Cloud, you must include a packages.txt file in your repository root so the server knows to install the C-libraries before running your Python code:

# packages.txt
tesseract-ocr
libgl1
libglib2.0-0

Select Python 3.11 in your deployment settings, hit deploy, and you now have a live, edge-executed zero-trust AI pipeline!

Performance Benchmarks & Edge Footprint
The primary design goal was local in-memory execution on lightweight compute without out-of-memory (OOM) failures:

RAM Footprint: ~145 MB steady state (by avoiding heavier transformer checkpoints and using en_core_web_sm).
Processing Latency: 320 ms to 650 ms per page on standard 2-core CPU environments.
Network Egress: 0 bytes prior to PII stripping.

These benchmarks were collected on a standard 2-core CPU environment without GPU acceleration.

As privacy regulations become stricter and organizations increasingly adopt LLMs, architectures that minimize data exposure before inference will become essential rather than optional.

What features do you think are most important for privacy-first AI pipelines moving forward?
Drop your thoughts below, or check out the full source code on GitHub.

Build a Multi-Agent RAG Legal Assistant with LangGraph, FastAPI, and Streamlit (Beginner Guide)

mmalaika.junaid — Tue, 22 Sep 2026 10:40:02 +0000

Retrieval-Augmented Generation (RAG) sounds complex, but the core concept is straightforward: instead of asking an AI model to answer purely from memory, you hand it specific reference documents and tell it to answer using only that text.

In this guide, you will build an end-to-end legal assistant tailored for UAE Federal Law. Although we use UAE legal documents in this tutorial, the same architecture can be applied to company policies, research papers, medical guidelines, knowledge bases, or any custom document collection. We will walk through every layer, from turning raw PDFs into searchable vectors to forcing an AI agent to fact-check its own answers.

(Above: The final Streamlit UI showing a verified answer and expandable sources)

What is Retrieval-Augmented Generation (RAG)?

What this step does: Explains the foundational concept behind our application.
Why we need it: To understand why we aren't just using ChatGPT out of the box.

Traditional LLMs answer using their training data. Because that data is frozen in time, they often hallucinate or invent fake legal clauses.

RAG allows an LLM to retrieve information from external documents before generating a response.

In this project, the assistant will:

Retrieve UAE legal text from Pinecone.
Send relevant passages to the LLM.
Generate an answer using those passages.
Verify the response using a fact-checking agent.

What Are We Building?

Most beginner tutorials teach "naive RAG," which follows a single straight line: Question ➔ Search ➔ Answer. If the model hallucinates a fake legal clause, the user receives false information.

We are building a cyclic multi-agent system that verifies its own output before sending it back:

User Question
      │
      ▼
Vector Search (Pinecone) ──────────► Retrieves statutory text chunks
      │
      ▼
Synthesizer Node ──────────────────► Drafts an answer using ONLY retrieved text
      │
      ▼
Fact-Checker Node (Gatekeeper) ────► Compares draft against raw legal text
      │
      ├── [FALSE: Unsupported claims] ──► Loops back to Synthesizer to rewrite
      │
      └── [TRUE: 100% Supported] ───────► Sends final response to user

Prerequisites & Project Structure

Before starting, you should be familiar with basic Python syntax, virtual environments, and basic HTTP requests. No prior experience with LangGraph or Docker is required.

Create a root directory named uae-legal-rag and organize your files like this:

uae-legal-rag/
├── data/
│   └── uae_labor_law.pdf # Place your legal document here
├── backend/
│   ├── __init__.py
│   ├── schemas.py        # Request/Response data models
│   ├── agent.py          # Multi-agent LangGraph state machine
│   └── server.py         # FastAPI application
├── frontend/
│   └── app.py            # Streamlit user interface
├── ingest.py             # Script to chunk & upload PDFs to Pinecone
├── .env                  # Secret API keys
├── requirements.txt      # Pinned dependencies
└── Dockerfile            # Container definition

Setting Up Pinecone & Free-Tier Dependencies

What this step does: Configures our zero-cost infrastructure and installs libraries.

Pinecone: Create a free starter account at pinecone.io. Create an index named uae-law with 384 dimensions (matching our open-source embedding model) and the cosine similarity metric.
OpenRouter: Sign up at openrouter.ai and generate an API key. We will use their openrouter/free endpoint.

Create a .env file in the project root:

PINECONE_API_KEY="your_pinecone_api_key"
OPENROUTER_API_KEY="your_openrouter_api_key"

Create requirements.txt and lock these exact versions to avoid breaking changes:

fastapi==0.110.0
uvicorn==0.29.0
pydantic==2.6.4
langgraph==0.0.30
langchain==0.1.13
langchain-community==0.0.29
langchain-pinecone==0.0.3
langchain-huggingface==0.0.2
langchain-openai==0.1.1
pinecone-client==3.2.2
streamlit==1.32.2
python-dotenv==1.0.1
pypdf==4.1.0
requests==2.31.0

Install them:

python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install -r requirements.txt

Document Ingestion: PDF to Pinecone (`ingest.py`)

What this step does: Converts a human-readable PDF into machine-searchable numbers (vectors).
Why we need it: A vector database starts empty. Without this, the AI has no law to search.

The flow is: PDF ➔ Chunking ➔ Embeddings ➔ Pinecone.

Place a PDF in the data/ directory, then create ingest.py:

import os
from dotenv import load_dotenv
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_pinecone import PineconeVectorStore

load_dotenv()

def ingest_documents():
    pdf_path = "data/uae_labor_law.pdf"
    if not os.path.exists(pdf_path):
        raise FileNotFoundError(f"Missing PDF at {pdf_path}. Please place a document there.")

    print("1. Loading PDF...")
    loader = PyPDFLoader(pdf_path)
    raw_documents = loader.load()

    print("2. Chunking text...")
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=150
    )
    docs = text_splitter.split_documents(raw_documents)
    print(f"Created {len(docs)} text chunks.")

    print("3. Generating embeddings & uploading to Pinecone...")
    # Runs locally on CPU at zero cost (384 dimensions)
    embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")

    PineconeVectorStore.from_documents(
        documents=docs,
        embedding=embeddings,
        index_name="uae-law"
    )
    print("Ingestion complete. Documents are now indexed in Pinecone.")

if __name__ == "__main__":
    ingest_documents()

How to test: Run python ingest.py in your terminal.
Expected outcome: You will see "Ingestion complete." Check your Pinecone dashboard to verify the vectors are there.
Common error: IndexNotFoundError means you forgot to create the uae-law index in the Pinecone console first.

Defining Data Schemas (`backend/schemas.py`)

What this step does: Sets up strict rules for what data can enter and leave our API.
Why we need it: To keep the application decoupled, the API validates user input before it ever touches the agent workflow.

Create backend/schemas.py:

from pydantic import BaseModel, Field
from typing import List

class ChatRequest(BaseModel):
    query: str = Field(..., min_length=5, max_length=500, description="Legal question")

class ChatResponse(BaseModel):
    verified_answer: str
    sources: List[str]

LangGraph Agents (`backend/agent.py`)

What this step does: Creates the "brain" of our application using three specific agents.

The Retriever Agent performs semantic search. Its job is simple: Receive a user question, search Pinecone, and return relevant legal passages. These passages are then passed to the Synthesizer Agent.

The Synthesizer drafts an initial answer grounded strictly in those excerpts.

The Fact-Checker compares the draft against the raw legal text. If it detects assumptions, it loops back to the Synthesizer to rewrite. (We loop back to the Synthesizer rather than the Retriever because retrieval is usually correct; the LLM simply needs to be forced to write a more conservative answer).

Create backend/agent.py:

import os
from typing import TypedDict, List
from dotenv import load_dotenv
from langgraph.graph import StateGraph, END
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_openai import ChatOpenAI

load_dotenv()

embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
vectorstore = PineconeVectorStore(index_name="uae-law", embedding=embeddings)

llm = ChatOpenAI(
    base_url="https://openrouter.ai/api/v1",
    api_key=os.getenv("OPENROUTER_API_KEY"),
    model="openrouter/free"
)

class GraphState(TypedDict):
    query: str
    context: List[str]
    draft: str
    verified_answer: str
    cycle_count: int

def retriever_node(state: GraphState):
    docs = vectorstore.similarity_search(state["query"], k=3)
    return {"context": [doc.page_content for doc in docs]}

def synthesizer_node(state: GraphState):
    context_block = "\n\n".join(state["context"])
    prompt = (
        f"You are a strict UAE legal assistant. Answer the question using ONLY the provided text.\n"
        f"Context:\n{context_block}\n\n"
        f"Question: {state['query']}\n"
        f"Answer:"
    )
    response = llm.invoke(prompt)
    return {"draft": response.content}

def fact_checker_node(state: GraphState):
    context_block = "\n\n".join(state["context"])
    prompt = (
        f"Evaluate if the following Answer is 100% supported by the Context.\n"
        f"Context:\n{context_block}\n\n"
        f"Answer:\n{state['draft']}\n\n"
        f"If completely supported without assumptions, reply ONLY with 'TRUE'.\n"
        f"If unsupported, reply ONLY with 'FALSE'."
    )
    result = llm.invoke(prompt).content.strip().upper()
    if "TRUE" in result:
        return {"verified_answer": state["draft"]}
    return {"cycle_count": state.get("cycle_count", 0) + 1}

def routing_gate(state: GraphState):
    if state.get("verified_answer"):
        return "approved"
    if state.get("cycle_count", 0) >= 5:
        return "limit_reached"
    return "rejected"

workflow = StateGraph(GraphState)
workflow.add_node("retriever", retriever_node)
workflow.add_node("synthesizer", synthesizer_node)
workflow.add_node("fact_checker", fact_checker_node)

workflow.set_entry_point("retriever")
workflow.add_edge("retriever", "synthesizer")
workflow.add_edge("synthesizer", "fact_checker")

workflow.add_conditional_edges(
    "fact_checker",
    routing_gate,
    {
        "approved": END,
        "limit_reached": END,
        "rejected": "synthesizer"
    }
)

legal_graph = workflow.compile()

def run_agent(query: str) -> dict:
    return legal_graph.invoke({"query": query, "cycle_count": 0})

FastAPI Backend (`backend/server.py`)

What this step does: Wraps our LangGraph brain in a web server.
Why we need it: So our frontend UI (or any other app) can communicate with the AI securely over HTTP.

Create backend/server.py:

from fastapi import FastAPI, HTTPException
from backend.schemas import ChatRequest, ChatResponse
from backend.agent import run_agent

app = FastAPI(title="UAE Legal RAG API")

@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
    try:
        result = run_agent(request.query)
        if not result.get("verified_answer"):
            raise HTTPException(
                status_code=500,
                detail="Safety check: Agent could not reach a verified answer within 5 retries."
            )
        return ChatResponse(
            verified_answer=result["verified_answer"],
            sources=result.get("context", [])
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

How to test: Run uvicorn backend.server:app --reload in your terminal. Navigate to http://127.0.0.1:8000/docs in your browser. You will see the Swagger UI where you can test the /chat endpoint directly.

(Above: FastAPI Swagger UI running locally)

Streamlit Frontend (`frontend/app.py`)

What this step does: Creates a visual chat window for the user.
Why we need it: To provide an interactive web UI with expandable source citations so users can verify the AI's claims.

Create frontend/app.py:

import streamlit as st
import requests

st.set_page_config(page_title="UAE Legal Assistant", page_icon="⚖️")
st.title("⚖️ UAE Legal Assistant")
st.caption("Multi-Agent Fact-Checked Legal Q&A (LangGraph + FastAPI)")

query = st.text_input("Enter your statutory inquiry:", placeholder="e.g., What is the probation period limit under UAE Labor Law?")

if st.button("Submit Query", type="primary"):
    if not query.strip():
        st.warning("Please provide a question.")
    else:
        with st.spinner("Retrieving clauses, drafting answer, and running fact-checker..."):
            try:
                response = requests.post(
                    "http://localhost:8000/chat",
                    json={"query": query},
                    timeout=60
                )
                if response.status_code == 200:
                    data = response.json()
                    st.success("Verification Passed")
                    st.markdown(f"**Answer:**\n{data['verified_answer']}")

                    with st.expander("Inspect Referenced Statutory Clauses"):
                        for idx, source in enumerate(data["sources"], start=1):
                            st.info(f"**Clause Chunk {idx}:**\n{source}")
                else:
                    st.error(f"Error {response.status_code}: {response.text}")
            except requests.exceptions.ConnectionError:
                st.error("Cannot connect to backend. Ensure FastAPI is running on port 8000.")

How to test: Open a new terminal window (keep FastAPI running in the first one) and run streamlit run frontend/app.py. Your browser will open the app automatically.

(Above: The Streamlit interface querying the backend)

Deployment (`Dockerfile`)

What this step does: Packages the entire backend into a standardized container.
Why we need it: So the application runs exactly the same way on any machine or cloud server, without dependency errors.

Create a Dockerfile in the root directory:

FROM python:3.10-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
EXPOSE 8000

CMD ["uvicorn", "backend.server:app", "--host", "0.0.0.0", "--port", "8000"]

Build and run your container:

docker build -t uae-legal-api .
docker run -p 8000:8000 --env-file .env uae-legal-api

You now have a fully decoupled, multi-agent RAG application. By separating the retrieval, synthesis, and fact-checking steps, you drastically reduce hallucinations.

At the time of writing, all services used in this guide have free tiers that are sufficient for learning and experimentation.

You can view the complete source code and run the project yourself here: MalaikaJunaid/multi-agent-rag-assistant

What You Learned

In this tutorial you:

Converted PDFs into embeddings
Stored vectors in Pinecone
Built a LangGraph workflow
Added fact-checking loops
Exposed the system through FastAPI
Created a Streamlit UI
Containerized the backend with Docker

These are the same building blocks used in production RAG systems.

Happy coding!