Automating Unstructured Documents: RAG & n8n for High-Efficiency Business Workflows

Introduction & The Problem

The Solution Concept & Architecture

1. Document Ingestion: New documents (PDFs, images, Word docs) are uploaded to a cloud storage (S3, Google Drive) or received via email, triggering the workflow.
2. Preprocessing: Documents undergo Optical Character Recognition (OCR) if image-based, then are split into manageable chunks. Each chunk is embedded into a numerical vector.
3. Vector Database: These embeddings are stored in a specialized vector database (e.g., Pinecone, Qdrant, Weaviate) for efficient semantic search.
4. n8n Workflow Orchestration: n8n acts as the central orchestrator, triggering actions, making API calls to the RAG service, transforming data, and updating target systems.
5. RAG Service: When n8n requests data extraction, the RAG service queries the vector database to retrieve semantically similar document chunks. These chunks, along with the specific extraction prompt, are fed to an LLM.
6. LLM Extraction: The LLM processes the retrieved context and prompt to accurately extract the desired structured data (e.g., invoice number, total amount, contract terms).
7. Business System Integration: The extracted data is then pushed by n8n into relevant business systems (ERP, CRM, accounting software, databases) or used for further automation.

Step-by-Step Implementation

• n8n installed (local or cloud).
• Python environment with openai, langchain, pypdf, pinecone-client (or similar vector DB client).
• OpenAI API key (or another LLM provider).

Part 1: Building the RAG Backend (Python)

# rag_service.py
from flask import Flask, request, jsonify
from pypdf import PdfReader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_pinecone import PineconeVectorStore
from pinecone import Pinecone, ServerlessSpec
import os

# Initialize Flask App
app = Flask(__name__)

# Environment Variables
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
PINE_API_KEY = os.getenv("PINE_API_KEY")
PINE_ENV = os.getenv("PINE_ENV") # e.g., "us-east-1"
INDEX_NAME = "document-processor"

# Initialize Pinecone
pinecone_client = Pinecone(api_key=PINE_API_KEY, environment=PINE_ENV)

# Check if index exists, if not create it
if INDEX_NAME not in pinecone_client.list_indexes():
    pinecone_client.create_index(
        name=INDEX_NAME,
        dimension=1536, # OpenAI 'text-embedding-ada-002' dimension
        metric='cosine',
        spec=ServerlessSpec(cloud='aws', region='us-east-1')
    )

embeddings = OpenAIEmbeddings(openai_api_key=OPENAI_API_KEY)
vectorstore = PineconeVectorStore(index_name=INDEX_NAME, embedding=embeddings)
llm = ChatOpenAI(model_name="gpt-4o", temperature=0, openai_api_key=OPENAI_API_KEY)

# Text splitter for processing documents
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
    add_start_index=True,
)

@app.route('/ingest_document', methods=['POST'])
def ingest_document():
    if 'file' not in request.files:
        return jsonify({"error": "No file part"}), 400
    file = request.files['file']
    if file.filename == '':
        return jsonify({"error": "No selected file"}), 400

    try:
        # Read PDF content
        reader = PdfReader(file.stream)
        raw_text = ''
        for i, page in enumerate(reader.pages):
            text = page.extract_text()
            if text:
                raw_text += text

        # Split text into chunks
        docs = text_splitter.create_documents([raw_text])

        # Add unique metadata for later retrieval specific to this document
        # For simplicity, we'll use a dummy doc_id, in production use actual UUIDs
        doc_id = "invoice_12345" # Replace with actual unique document ID
        for doc in docs:
            doc.metadata = {"source": doc_id}

        # Store embeddings in Pinecone
        vectorstore.add_documents(documents=docs)

        return jsonify({"message": f"Document {doc_id} ingested successfully"}), 200
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route('/extract_data', methods=['POST'])
def extract_data():
    data = request.get_json()
    question = data.get('question')
    doc_id = data.get('doc_id') # To filter retrieval for a specific document

    if not question or not doc_id:
        return jsonify({"error": "'question' and 'doc_id' are required"}), 400

    try:
        # Retrieve relevant chunks from Pinecone for the specific document
        retriever = vectorstore.as_retriever(search_kwargs={"filter": {"source": doc_id}})
        docs = retriever.get_relevant_documents(query=question)

        # Construct prompt for LLM with context
        context = "

".join([doc.page_content for doc in docs])
        prompt_template = f"""
        Given the following context from a document:
        ---
        {context}
        ---
        Please answer the following question. If the answer is not in the context, state 'Not found'.
        Question: {question}
        """

        # Get answer from LLM
        response = llm.invoke(prompt_template)
        extracted_text = response.content

        return jsonify({"question": question, "answer": extracted_text}), 200
    except Exception as e:
        return jsonify({"error": str(e)}), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

Part 2: Designing the n8n Workflow

1. Be triggered by a new file upload (e.g., to an S3 bucket or via a webhook).
2. Call the ingest_document endpoint to process the PDF.
3. Call the extract_data endpoint multiple times for specific fields.
4. Structure the extracted data.
5. Update a hypothetical accounting system (e.g., via a REST API).

1. Start Node: Webhook (or S3 Trigger)

• Description: Listens for an incoming document upload event. For demonstration, we'll use a simple Webhook that receives a file.
• Configuration: Mode: POST, Path: /new-document.

1. HTTP Request Node (Ingest Document):

• Description: Sends the uploaded PDF to our RAG service's /ingest_document endpoint.
• Method: POST
• URL: http://your-rag-service:5000/ingest_document (replace with your actual URL)
• Headers: Content-Type: multipart/form-data
• Body: Binary Data, Reference: {{$binary.data}} (assuming the webhook receives binary data).
• Outputs: Will return {"message": "Document ingested successfully"}.

1. Set Node (Prepare Extraction Questions):

• Description: Define the questions for data extraction. The doc_id should be dynamically passed from the ingestion step (or a unique ID generated for each document).
• Configuration (Example):

[
          {
            "question": "What is the invoice number?",
            "field": "invoiceNumber",
            "doc_id": "invoice_12345" // Dynamic doc_id from previous step
          },
          {
            "question": "What is the total amount due? Only provide the numerical value.",
            "field": "totalAmount",
            "doc_id": "invoice_12345"
          },
          {
            "question": "Who is the vendor/supplier?",
            "field": "vendorName",
            "doc_id": "invoice_12345"
          }
        ]

1. Looping Node (Split in Batches or Item Lists + Merge):

• Description: Iterate through each question defined in the 'Set Node'.

1. HTTP Request Node (Extract Data):

• Description: For each question, call the RAG service's /extract_data endpoint.
• Method: POST
• URL: http://your-rag-service:5000/extract_data
• Headers: Content-Type: application/json
• Body (JSON):

{
          "question": "{{$json.question}}",
          "doc_id": "{{$json.doc_id}}"
        }

1. Merge Node (Merge by key):

• Description: Combine the results from the looping extraction into a single JSON object.
• Key: Could be question or field.
• Input 1: Results from 'Set Node' (original questions).
• Input 2: Results from 'HTTP Request (Extract Data)'.

1. Set Node (Final Data Structure):

• Description: Structure the extracted data into a clean JSON object ready for your business system.
• Configuration (Example Expression):

{
          "documentId": "{{$json.doc_id}}",
          "invoiceNumber": "{{$('HTTP Request (Extract Data)').item.json.answer[0].answer}}",
          "totalAmount": "{{$('HTTP Request (Extract Data)').item.json.answer[1].answer}}",
          "vendorName": "{{$('HTTP Request (Extract Data)').item.json.answer[2].answer}}",
          "extractedAt": "{{new Date().toISOString()}}"
        }

1. HTTP Request Node (Update Accounting System):

• Description: Send the structured data to your accounting or ERP system's API.
• Method: POST
• URL: https://your-accounting-system.com/api/invoices
• Headers: Content-Type: application/json, Authorization: Bearer YOUR_API_KEY
• Body: JSON, Reference: {{$json}} (from previous 'Set Node').

Optimization & Best Practices

• Prompt Engineering: Fine-tune extraction prompts for the LLM. Use few-shot examples or provide specific output formats (e.g., JSON schema) to guide the LLM's response for better parsing.
• Chunking Strategy: Experiment with chunk_size and chunk_overlap for optimal RAG performance. Smaller chunks might be too fragmented; larger chunks might introduce irrelevant context.
• Error Handling & Retries: Implement robust error handling in n8n (e.g., Continue On Fail, Retry on Error) and within the RAG service. For critical data, incorporate a human-in-the-loop review step if confidence scores are low or extraction fails.
• Document Quality: The accuracy of OCR directly impacts the RAG system. Use high-quality scanning and consider advanced OCR services for complex or poor-quality documents.
• Security & Access Control: Secure your RAG API endpoints with proper authentication (API keys, JWTs). Ensure sensitive data handled by n8n or the RAG service is encrypted both in transit and at rest. Manage API keys securely using environment variables or a secret management service.
• Scalability: For high-volume processing, consider deploying your RAG service on serverless platforms (AWS Lambda, Google Cloud Functions) and using managed vector databases that scale automatically. n8n can also be deployed in a scalable configuration.
• Version Control: Treat your n8n workflows and RAG code as production code, using version control systems (Git) and CI/CD pipelines for deployment.
• LLM Choice: Evaluate different LLMs (GPT-4o, Claude 3, Llama 3) based on performance, cost, and specific task requirements. GPT-4o often excels at complex extraction tasks.

Business Impact & ROI

• Reduced Operational Costs: Automating data extraction from invoices, contracts, and other documents can cut labor costs associated with manual data entry by up to 70%. This frees up valuable human resources for higher-value, strategic activities.
• Accelerated Processing Cycles: Tasks that once took days or even weeks (e.g., invoice approval, contract review) can now be completed in minutes. This leads to faster cash flow, quicker supplier payments, and improved customer satisfaction.
• Improved Data Accuracy: AI-driven extraction significantly reduces human error rates. Cleaner, more accurate data leads to better financial reporting, fewer compliance issues, and more reliable business intelligence.
• Enhanced Compliance & Auditability: Automated workflows provide clear audit trails for every processed document, ensuring adherence to regulatory requirements and simplifying compliance efforts.
• Strategic Resource Reallocation: By eliminating monotonous tasks, employees can be re-skilled or re-deployed to roles that require human creativity, critical thinking, and interpersonal skills, boosting overall organizational productivity and morale.
• Scalability & Agility: The system can easily scale to handle increasing document volumes without proportional increases in staffing, providing a flexible foundation for business growth and adapting to changing demands.

Conclusion