RAG System

I built a RAG system for financial documents and put the whole thing on GitHub: fintech-rag-system. OpenAI for embeddings and generation, Pinecone for the vectors, Flask in front. What follows is how the pieces fit together and the parts that turned out to matter more than the tutorials suggest.

The hard part of RAG isn’t the retrieval. It’s the chunking, and financial documents are a bad case for it.

System architecture

Two flows, one for getting documents in and one for answering questions:

Steps 1 to 3 are the ingest side and run offline. Steps 4 to 9 are the query side and run on every request, which is where your latency budget goes.

Current State of Technology

Key Components:

  1. Large Language Models (LLMs): OpenAI’s GPT-4, Anthropic’s Claude, or open-source alternatives like Llama 2.
  2. Vector Databases: Pinecone, Weaviate, or Milvus for efficient similarity search.
  3. Embedding Models: SentenceTransformers, OpenAI’s text-embedding-ada-002, or domain-specific models.
  4. Cloud Platforms: AWS, Google Cloud, or Azure for scalable infrastructure.

Go-To Tools:

  • LLM: OpenAI’s GPT-4 (for its superior performance in financial contexts)
  • Vector Database: Pinecone (for its ease of use, scalability, and free tier!)
  • Embedding Model: text-embedding-ada-002 (for its performance and compatibility with GPT-4)
  • Cloud Platform: AWS (for its comprehensive services and wide adoption in finance)

Building a RAG System: Step-by-Step Guide

Step 1: Data Preparation

  1. Collect and clean financial documents (reports, news articles, regulatory filings).
  2. Chunk documents into smaller segments (e.g., paragraphs or sentences).
import nltk
from nltk.tokenize import sent_tokenize

nltk.download('punkt')

def chunk_document(doc, max_chunk_size=1000):
    sentences = sent_tokenize(doc)
    chunks = []
    current_chunk = ""
    
    for sentence in sentences:
        if len(current_chunk) + len(sentence) <= max_chunk_size:
            current_chunk += sentence + " "
        else:
            chunks.append(current_chunk.strip())
            current_chunk = sentence + " "
    
    if current_chunk:
        chunks.append(current_chunk.strip())
    
    return chunks

# Example usage
document = "Your long financial document text here..."
chunks = chunk_document(document)

Step 2: Embedding Generation

Use OpenAI’s API to generate embeddings for each chunk.

from openai import OpenAI
import os

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def generate_embedding(text):
    text = text.replace("\n", " ")
    response = client.embeddings.create(input=[text], model="text-embedding-3-small")
    return response.data[0].embedding

# Generate embeddings for chunks
chunk_embeddings = [generate_embedding(chunk) for chunk in chunks]

Step 3: Vector Database Setup

Set up Pinecone and insert the embeddings.

from pinecone import Pinecone, ServerlessSpec

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

# Create or connect to the Pinecone index
index_name = "fintech-documents"
if index_name not in pc.list_indexes().names():
    pc.create_index(
        name=index_name,
        dimension=1536,
        metric='cosine',
        spec=ServerlessSpec(
            cloud='aws',
            region='us-east-1'
        )
    )
index = pc.Index(index_name)

Step 4: Query Processing

Implement the RAG system to process user queries.

def process_query(query):
    query_embedding = generate_embedding(query)
    search_results = index.query(vector=query_embedding, top_k=3, include_metadata=True)
    context = " ".join([result.metadata['text'] for result in search_results.matches])
    
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "You are a financial expert assistant. Use the following context to answer the user's question."},
            {"role": "user", "content": f"Context: {context}\n\nQuestion: {query}"}
        ]
    )
    
    return response.choices[0].message.content

# Example usage
user_query = "What are the recent trends in cryptocurrency regulations?"
answer = process_query(user_query)
print(answer)

Step 5: Cloud Deployment

To containerize your application, create a Dockerfile in your project root:

FROM python:3.9-slim

WORKDIR /app

COPY . /app

RUN pip install --no-cache-dir -r requirements.txt

EXPOSE 8000

# Run app.py when the container launches
CMD ["python", "app.py"]

Make sure to create a requirements.txt file with all the necessary dependencies:

nltk==3.8.1
openai==1.35.14
pinecone==4.0.0
Flask==3.0.3

Where this one falls down

The naive chunker above splits on sentences at a fixed character count, and on financial documents that’s the weakest part of the system. A 10-K has tables that lose all meaning when you cut them in half, footnotes that carry the qualification for a number three pages earlier, and multi-column layouts that a text extractor happily interleaves into nonsense. Retrieval quality tracks chunking quality almost exactly, so this is where the effort goes if you’re building something real.

Two other things worth knowing before you get attached to a result:

Similarity is not relevance. Cosine similarity over embeddings retrieves passages that read like the question. That isn’t the same as passages that answer it, and on a corpus where many documents use near-identical regulatory boilerplate the top three matches can be three copies of the same disclaimer.

A confident wrong answer is the failure mode you’ll ship. The model gets context and a question and writes fluent prose either way. When retrieval misses, nothing in the output looks different. If the answers matter, the citations back to source chunks aren’t a nice extra, they’re the only thing making the output checkable.

The repo is a working skeleton rather than something I’d put in front of a regulator. It’s the right size to read in one sitting and find out whether the approach fits your problem.