Complete RAG System Implementation Guide: Building an Enterprise-Grade Intelligent Q&A System

RAG (Retrieval-Augmented Generation) is currently the most popular LLM application architecture, enabling large language models to answer domain-specific questions while significantly reducing hallucination issues. This article provides a detailed walkthrough on building a production-grade RAG system from scratch.

RAG System Architecture Overview

Core Components

┌─────────────────────────────────────────────────────────┐
│                  RAG System Architecture                  │
├─────────────────────────────────────────────────────────┤
│                                                          │
│  ┌──────────┐      ┌──────────┐      ┌──────────┐     │
│  │ Document  │ ───→ │Vectorize │ ───→ │  Vector  │     │
│  │Processing │      │          │      │ Database │     │
│  └──────────┘      └──────────┘      └──────────┘     │
│       │                                     ↑           │
│       │                                     │           │
│       ↓                                     │           │
│  ┌──────────┐                         ┌─────────┐     │
│  │  User    │ ─── Query ───→ Vector ──→│  LLM    │     │
│  │  Query   │          Search ↓        │Generate │     │
│  └──────────┘      ┌─────────┐        └─────────┘     │
│                     │ Re-rank │               │          │
│                     └─────────┘               │          │
│                           │                   │          │
│                           └────→ Merge ←──────┘          │
│                                  ↓                       │
│                            ┌──────────┐                 │
│                            │  Answer   │                 │
│                            └──────────┘                 │
└─────────────────────────────────────────────────────────┘

Technology Selection Recommendations

Step 1: Document Processing and Chunking Strategy

Intelligent Document Chunking

Document chunking is the foundation of a RAG system and directly impacts retrieval quality.

Basic Chunking Strategy:

from langchain.text_splitter import RecursiveCharacterTextSplitter

def smart_text_splitter(documents, chunk_size=1000, chunk_overlap=200):
    """
    Intelligent text chunker

    Parameters:
    - chunk_size: Size of each text chunk (token count)
    - chunk_overlap: Overlap portion to maintain contextual coherence
    """
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=[
            "\n\n",  # Paragraph separator
            "\n",    # Line separator
            "。",    # Chinese period
            ".",     # English period
            "!",
            "?",
            ";",
            ":",
            " ",     # Space
            "",      # Character
        ]
    )

    chunks = text_splitter.split_documents(documents)
    return chunks

# Usage example
from langchain.document_loaders import PyPDFLoader, TextLoader

# Load document
pdf_loader = PyPDFLoader("company_handbook.pdf")
documents = pdf_loader.load()

# Chunk processing
chunks = smart_text_splitter(documents, chunk_size=800, chunk_overlap=150)

print(f"Total documents: {len(documents)}")
print(f"Total chunks: {len(chunks)}")

Advanced: Semantic Chunking

Traditional character-based chunking may break semantic integrity. Semantic chunking can improve this:

from langchain.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

def semantic_chunking(documents, breakpoint_threshold_type="percentile"):
    """
    Intelligent chunking based on semantic similarity

    breakpoint_threshold_type:
    - percentile: Percentile threshold
    - standard_deviation: Standard deviation threshold
    - interquartile: Interquartile range threshold
    """
    embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

    text_splitter = SemanticChunker(
        embeddings=embeddings,
        breakpoint_threshold_type=breakpoint_threshold_type,
        breakpoint_threshold_amount=75  # 75th percentile
    )

    chunks = text_splitter.create_documents([doc.page_content for doc in documents])
    return chunks

# Compare both chunking approaches
traditional_chunks = smart_text_splitter(documents)
semantic_chunks = semantic_chunking(documents)

print(f"Traditional chunks: {len(traditional_chunks)}")
print(f"Semantic chunks: {len(semantic_chunks)}")

Document Metadata Enhancement

Adding rich metadata can improve retrieval precision:

def enhance_chunks_metadata(chunks, source_info):
    """
    Add structured metadata to document chunks
    """
    enhanced_chunks = []

    for i, chunk in enumerate(chunks):
        # Basic metadata
        chunk.metadata.update({
            'chunk_id': i,
            'source': source_info['filename'],
            'document_type': source_info['type'],
            'created_date': source_info['created_date'],
            'department': source_info.get('department', 'general'),
            'security_level': source_info.get('security_level', 'public'),
        })

        # Content analysis metadata
        chunk.metadata.update({
            'word_count': len(chunk.page_content.split()),
            'has_code': '```' in chunk.page_content,
            'has_table': '|' in chunk.page_content,
            'language': detect_language(chunk.page_content),
        })

        # Semantic tags (optional)
        chunk.metadata['tags'] = extract_key_topics(chunk.page_content)

        enhanced_chunks.append(chunk)

    return enhanced_chunks

def extract_key_topics(text, max_topics=5):
    """
    Extract key topics using LLM
    """
    from openai import OpenAI
    client = OpenAI()

    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{
            "role": "user",
            "content": f"Extract {max_topics} key topics from the following text, separated by commas:\n\n{text}"
        }],
        temperature=0.3,
    )

    topics = response.choices[0].message.content.strip().split(',')
    return [topic.strip() for topic in topics]

Step 2: Vector Embedding and Index Building

Choosing the Right Embedding Model

Chinese Embedding Model Comparison:

from sentence_transformers import SentenceTransformer
from openai import OpenAI
import numpy as np

class EmbeddingComparison:
    def __init__(self):
        # Local models
        self.bge_model = SentenceTransformer('BAAI/bge-large-zh-v1.5')
        self.m3e_model = SentenceTransformer('moka-ai/m3e-base')

        # API model
        self.openai_client = OpenAI()

    def get_embeddings(self, text, model='bge'):
        """Get text embedding vectors"""
        if model == 'bge':
            return self.bge_model.encode(text, normalize_embeddings=True)

        elif model == 'm3e':
            return self.m3e_model.encode(text, normalize_embeddings=True)

        elif model == 'openai':
            response = self.openai_client.embeddings.create(
                model="text-embedding-3-large",
                input=text
            )
            return np.array(response.data[0].embedding)

    def compare_models(self, queries, docs):
        """Compare retrieval performance across different models"""
        results = {}

        for model_name in ['bge', 'm3e', 'openai']:
            # Embed documents
            doc_embeddings = [self.get_embeddings(doc, model_name) for doc in docs]

            # Embed query
            query_embedding = self.get_embeddings(queries[0], model_name)

            # Calculate similarity
            similarities = [
                np.dot(query_embedding, doc_emb)
                for doc_emb in doc_embeddings
            ]

            # Sort
            ranked_indices = np.argsort(similarities)[::-1]

            results[model_name] = {
                'top_3_indices': ranked_indices[:3].tolist(),
                'top_3_scores': [similarities[i] for i in ranked_indices[:3]]
            }

        return results

# Usage example
comparator = EmbeddingComparison()

queries = ["How do I apply for employee travel subsidies?"]
docs = [
    "Employee benefits include travel subsidies, up to $5,000 per year...",
    "Leave process: First fill out the leave form, then get supervisor approval...",
    "Travel subsidy applications require receipts, submitted within 30 days after the event...",
]

results = comparator.compare_models(queries, docs)
for model, result in results.items():
    print(f"\n{model} model results:")
    print(f"  Top 3 indices: {result['top_3_indices']}")
    print(f"  Top 3 scores: {result['top_3_scores']}")

Building a Vector Index - Qdrant Implementation

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from sentence_transformers import SentenceTransformer
import uuid

class QdrantRAGIndex:
    def __init__(self, collection_name="company_docs"):
        # Initialize Qdrant client
        self.client = QdrantClient(url="http://localhost:6333")
        self.collection_name = collection_name

        # Embedding model
        self.embedder = SentenceTransformer('BAAI/bge-large-zh-v1.5')
        self.embedding_dim = self.embedder.get_sentence_embedding_dimension()

        # Create collection
        self._create_collection()

    def _create_collection(self):
        """Create or recreate the collection"""
        try:
            self.client.recreate_collection(
                collection_name=self.collection_name,
                vectors_config=VectorParams(
                    size=self.embedding_dim,
                    distance=Distance.COSINE  # Use cosine similarity
                )
            )
            print(f"Collection '{self.collection_name}' created successfully")
        except Exception as e:
            print(f"Error creating collection: {e}")

    def index_documents(self, documents, batch_size=100):
        """
        Batch index documents

        documents: List of (text, metadata) tuples
        """
        points = []

        for i, (text, metadata) in enumerate(documents):
            # Generate vector
            vector = self.embedder.encode(text, normalize_embeddings=True).tolist()

            # Create point
            point = PointStruct(
                id=str(uuid.uuid4()),
                vector=vector,
                payload={
                    'text': text,
                    'metadata': metadata,
                    'chunk_index': i
                }
            )
            points.append(point)

            # Batch upload
            if len(points) >= batch_size:
                self.client.upsert(
                    collection_name=self.collection_name,
                    points=points
                )
                print(f"Indexed {i+1} documents...")
                points = []

        # Upload remaining documents
        if points:
            self.client.upsert(
                collection_name=self.collection_name,
                points=points
            )

        print(f"Total {i+1} documents indexed successfully")

    def search(self, query, top_k=5, score_threshold=0.7, filters=None):
        """
        Search for relevant documents

        query: Query text
        top_k: Number of results to return
        score_threshold: Minimum similarity threshold
        filters: Metadata filter conditions
        """
        # Generate query vector
        query_vector = self.embedder.encode(query, normalize_embeddings=True).tolist()

        # Execute search
        search_result = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_vector,
            limit=top_k,
            score_threshold=score_threshold,
            query_filter=filters  # Optional metadata filtering
        )

        return search_result

# Usage example
rag_index = QdrantRAGIndex(collection_name="company_knowledge_base")

# Prepare documents
documents = [
    ("Travel subsidy application process: Original receipts required, submit within 30 days after the event...",
     {"category": "Benefits", "department": "HR"}),
    ("Leave policy: Sick leave requires a doctor's note, personal leave must be requested 3 days in advance...",
     {"category": "HR", "department": "HR"}),
    # ... more documents
]

# Index documents
rag_index.index_documents(documents)

# Search
results = rag_index.search(
    query="How to apply for travel subsidy?",
    top_k=3,
    score_threshold=0.75
)

for result in results:
    print(f"Score: {result.score:.4f}")
    print(f"Text: {result.payload['text'][:100]}...")
    print(f"Metadata: {result.payload['metadata']}\n")

Step 3: Retrieval Optimization Strategies

Hybrid Search

Combining vector search with keyword search to improve recall:

from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
    def __init__(self, vector_index, documents):
        self.vector_index = vector_index
        self.documents = documents

        # Build BM25 index
        tokenized_docs = [doc.split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized_docs)

    def retrieve(self, query, top_k=10, alpha=0.5):
        """
        Hybrid retrieval

        alpha: Vector search weight (0-1)
               1-alpha: BM25 weight
        """
        # 1. Vector search
        vector_results = self.vector_index.search(query, top_k=top_k*2)
        vector_scores = {
            result.payload['chunk_index']: result.score
            for result in vector_results
        }

        # 2. BM25 keyword search
        tokenized_query = query.split()
        bm25_scores = self.bm25.get_scores(tokenized_query)

        # Normalize BM25 scores to [0, 1]
        max_bm25 = max(bm25_scores) if max(bm25_scores) > 0 else 1
        normalized_bm25 = bm25_scores / max_bm25

        # 3. Hybrid score calculation
        hybrid_scores = {}
        all_indices = set(vector_scores.keys()) | set(range(len(bm25_scores)))

        for idx in all_indices:
            vec_score = vector_scores.get(idx, 0)
            bm25_score = normalized_bm25[idx] if idx < len(normalized_bm25) else 0

            # Weighted combination
            hybrid_scores[idx] = alpha * vec_score + (1 - alpha) * bm25_score

        # 4. Sort and return top-k
        sorted_indices = sorted(
            hybrid_scores.items(),
            key=lambda x: x[1],
            reverse=True
        )[:top_k]

        return [
            {
                'index': idx,
                'text': self.documents[idx],
                'score': score
            }
            for idx, score in sorted_indices
        ]

# Usage example
hybrid_retriever = HybridRetriever(rag_index, [doc[0] for doc in documents])

results = hybrid_retriever.retrieve(
    query="What documents are needed for travel subsidy?",
    top_k=5,
    alpha=0.7  # 70% vector search, 30% BM25
)

Re-ranking

Using a cross-encoder to rerank initial retrieval results:

from sentence_transformers import CrossEncoder

class ReRanker:
    def __init__(self, model_name='BAAI/bge-reranker-large'):
        self.model = CrossEncoder(model_name)

    def rerank(self, query, documents, top_k=5):
        """
        Rerank retrieval results

        Returns: List of (document, score) sorted by relevance
        """
        # Build query-document pairs
        pairs = [[query, doc] for doc in documents]

        # Calculate relevance scores
        scores = self.model.predict(pairs)

        # Sort
        ranked_results = sorted(
            zip(documents, scores),
            key=lambda x: x[1],
            reverse=True
        )

        return ranked_results[:top_k]

# Integrate into the RAG pipeline
class RAGPipelineWithRerank:
    def __init__(self, retriever, reranker, llm):
        self.retriever = retriever
        self.reranker = reranker
        self.llm = llm

    def query(self, question, retrieve_k=20, final_k=5):
        # 1. Initial retrieval (recall more candidates)
        candidates = self.retriever.retrieve(question, top_k=retrieve_k)
        candidate_texts = [c['text'] for c in candidates]

        # 2. Re-ranking (precise sorting)
        reranked = self.reranker.rerank(question, candidate_texts, top_k=final_k)

        # 3. Build prompt
        context = "\n\n".join([doc for doc, score in reranked])
        prompt = f"""Answer the question based on the following information:

{context}

Question: {question}

Please provide a detailed and accurate answer. If the information is insufficient to answer the question, clearly state so."""

        # 4. Generate answer
        response = self.llm.generate(prompt)

        return {
            'answer': response,
            'sources': reranked,
            'context': context
        }

Step 4: LLM Integration and Prompt Engineering

Prompt Template Design

from langchain.prompts import PromptTemplate

# Basic RAG prompt
basic_rag_template = """You are a professional enterprise knowledge base assistant. Please answer the user's question based on the provided context information.

Context Information:
{context}

User Question: {question}

Answer Requirements:
1. Answer based on the provided context information
2. If the context does not contain relevant information, clearly state so
3. Provide specific reference sources
4. Answers should be clear, professional, and easy to understand

Answer:"""

basic_prompt = PromptTemplate(
    template=basic_rag_template,
    input_variables=["context", "question"]
)

# Advanced RAG prompt (with chain-of-thought)
advanced_rag_template = """You are a professional enterprise knowledge base assistant. Please carefully analyze the provided context information and answer the user's question.

Context Information:
{context}

User Question: {question}

Please answer following these steps:

1. **Understand the Question**: First analyze the core intent of the user's question

2. **Information Retrieval**: Identify relevant information from the context
   - List all relevant passages
   - Note the information sources

3. **Reasoning and Analysis**:
   - Integrate multiple information sources
   - Analyze the relevance and reliability of the information
   - Identify potential contradictions or inconsistencies

4. **Generate Answer**:
   - Provide a clear, complete answer
   - Cite specific sources
   - If information is insufficient, clearly indicate the gaps

5. **Suggested Actions**: If applicable, provide next-step recommendations

Answer Format:
[Problem Analysis]
<your problem analysis>

[Relevant Information]
<key information extracted from context>

[Answer]
<detailed answer>

[Reference Sources]
<cited passage numbers and summaries>

[Recommendations] (if applicable)
<next-step recommendations>
"""

advanced_prompt = PromptTemplate(
    template=advanced_rag_template,
    input_variables=["context", "question"]
)

Complete RAG Chain Implementation

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

class ProductionRAGSystem:
    def __init__(
        self,
        vector_store,
        llm_model="gpt-4-turbo-preview",
        temperature=0.1,
        streaming=True
    ):
        # LLM configuration
        callbacks = [StreamingStdOutCallbackHandler()] if streaming else []
        self.llm = ChatOpenAI(
            model=llm_model,
            temperature=temperature,
            callbacks=callbacks
        )

        # Retriever configuration
        self.retriever = vector_store.as_retriever(
            search_type="similarity_score_threshold",
            search_kwargs={
                "k": 10,  # Initial retrieval count
                "score_threshold": 0.7  # Minimum similarity
            }
        )

        # Re-ranker
        self.reranker = ReRanker()

        # Build chain
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",  # or "map_reduce", "refine"
            retriever=self.retriever,
            return_source_documents=True,
            chain_type_kwargs={
                "prompt": advanced_prompt
            }
        )

    def query(self, question, rerank=True):
        """
        Execute RAG query

        Returns:
            answer: Generated answer
            sources: Reference source documents
            metadata: Additional metadata (scores, time, etc.)
        """
        import time
        start_time = time.time()

        # 1. Retrieve relevant documents
        docs = self.retriever.get_relevant_documents(question)

        # 2. Re-ranking (optional)
        if rerank and len(docs) > 3:
            doc_texts = [doc.page_content for doc in docs]
            reranked = self.reranker.rerank(question, doc_texts, top_k=5)
            docs = [doc for doc, score in reranked]

        # 3. Generate answer
        result = self.qa_chain.invoke({"query": question})

        # 4. Format output
        elapsed_time = time.time() - start_time

        return {
            'answer': result['result'],
            'sources': result['source_documents'],
            'metadata': {
                'elapsed_time': elapsed_time,
                'num_sources': len(result['source_documents']),
                'model': self.llm.model_name
            }
        }

# Usage example
rag_system = ProductionRAGSystem(
    vector_store=qdrant_vector_store,
    llm_model="gpt-4-turbo-preview",
    temperature=0.1,
    streaming=True
)

# Query
result = rag_system.query("What is the employee travel subsidy application process?")

print(f"\nAnswer:\n{result['answer']}\n")
print(f"Number of reference sources: {result['metadata']['num_sources']}")
print(f"Query time: {result['metadata']['elapsed_time']:.2f} seconds")

Step 5: Evaluation and Optimization

RAG System Evaluation Metrics

from ragas import evaluate
from ragas.metrics import (
    answer_relevancy,
    faithfulness,
    context_recall,
    context_precision
)

class RAGEvaluator:
    def __init__(self, rag_system):
        self.rag_system = rag_system
        self.metrics = [
            answer_relevancy,
            faithfulness,
            context_recall,
            context_precision
        ]

    def create_evaluation_dataset(self, qa_pairs):
        """
        Prepare evaluation dataset

        qa_pairs: List of {
            'question': str,
            'ground_truth': str,  # Ground truth answer
            'context': str  # Gold context (optional)
        }
        """
        dataset = {
            'question': [],
            'answer': [],
            'contexts': [],
            'ground_truths': []
        }

        for qa in qa_pairs:
            # Execute RAG query
            result = self.rag_system.query(qa['question'])

            dataset['question'].append(qa['question'])
            dataset['answer'].append(result['answer'])
            dataset['contexts'].append([
                doc.page_content for doc in result['sources']
            ])
            dataset['ground_truths'].append([qa['ground_truth']])

        return dataset

    def evaluate(self, dataset):
        """Execute evaluation"""
        from datasets import Dataset

        eval_dataset = Dataset.from_dict(dataset)

        # Run evaluation
        result = evaluate(
            dataset=eval_dataset,
            metrics=self.metrics
        )

        return result

# Usage example
evaluator = RAGEvaluator(rag_system)

# Prepare test data
test_qa_pairs = [
    {
        'question': 'How to apply for travel subsidy?',
        'ground_truth': 'You need to submit original receipts and the application form within 30 days after the trip...'
    },
    # ... more test cases
]

# Build evaluation dataset
eval_dataset = evaluator.create_evaluation_dataset(test_qa_pairs)

# Run evaluation
evaluation_results = evaluator.evaluate(eval_dataset)

print("Evaluation Results:")
print(f"Answer Relevancy: {evaluation_results['answer_relevancy']:.4f}")
print(f"Faithfulness: {evaluation_results['faithfulness']:.4f}")
print(f"Context Recall: {evaluation_results['context_recall']:.4f}")
print(f"Context Precision: {evaluation_results['context_precision']:.4f}")

A/B Testing Framework

import random
from collections import defaultdict

class RAGABTester:
    def __init__(self, system_a, system_b):
        self.system_a = system_a
        self.system_b = system_b
        self.results = defaultdict(list)

    def run_ab_test(self, queries, user_feedback_func=None):
        """
        Execute A/B test

        user_feedback_func: Optional function to collect user ratings
        """
        for query in queries:
            # Randomly assign to A or B
            system = random.choice(['A', 'B'])

            if system == 'A':
                result = self.system_a.query(query)
            else:
                result = self.system_b.query(query)

            # Record results
            self.results[system].append({
                'query': query,
                'answer': result['answer'],
                'elapsed_time': result['metadata']['elapsed_time'],
                'num_sources': result['metadata']['num_sources']
            })

            # Collect user feedback (optional)
            if user_feedback_func:
                rating = user_feedback_func(query, result['answer'])
                self.results[system][-1]['user_rating'] = rating

        return self.analyze_results()

    def analyze_results(self):
        """Analyze A/B test results"""
        analysis = {}

        for system in ['A', 'B']:
            results = self.results[system]

            analysis[system] = {
                'avg_response_time': sum(r['elapsed_time'] for r in results) / len(results),
                'avg_num_sources': sum(r['num_sources'] for r in results) / len(results),
                'total_queries': len(results)
            }

            # If user ratings are available
            if 'user_rating' in results[0]:
                analysis[system]['avg_rating'] = sum(r['user_rating'] for r in results) / len(results)

        return analysis

Production Deployment

FastAPI Service Wrapper

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uvicorn

app = FastAPI(title="RAG API Service", version="1.0.0")

# Global RAG system instance
rag_system = None

class Query(BaseModel):
    question: str
    top_k: Optional[int] = 5
    rerank: Optional[bool] = True

class Answer(BaseModel):
    answer: str
    sources: List[dict]
    metadata: dict

@app.on_event("startup")
async def startup_event():
    """Initialize RAG system on startup"""
    global rag_system
    rag_system = ProductionRAGSystem(
        vector_store=initialize_vector_store(),
        llm_model="gpt-4-turbo-preview"
    )
    print("RAG System initialized successfully")

@app.post("/query", response_model=Answer)
async def query_endpoint(query: Query):
    """RAG query endpoint"""
    try:
        result = rag_system.query(
            question=query.question,
            rerank=query.rerank
        )

        return Answer(
            answer=result['answer'],
            sources=[
                {
                    'content': doc.page_content,
                    'metadata': doc.metadata
                }
                for doc in result['sources'][:query.top_k]
            ],
            metadata=result['metadata']
        )

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "version": "1.0.0"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Docker Containerization

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

# Download embedding model (if using a local model)
RUN python -c "from sentence_transformers import SentenceTransformer; SentenceTransformer('BAAI/bge-large-zh-v1.5')"

# Expose port
EXPOSE 8000

# Start application
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'

services:
  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
    volumes:
      - qdrant_storage:/qdrant/storage

  rag-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - QDRANT_URL=http://qdrant:6333
    depends_on:
      - qdrant
    volumes:
      - ./data:/app/data

volumes:
  qdrant_storage:

Real-World Case Studies

Case 1: Enterprise Internal Knowledge Base

Scenario: A tech company builds an employee handbook and technical documentation Q&A system

Implementation Focus:

• Hierarchical document indexing (by department, category)
• Access control and information security
• Multi-language support (Chinese and English)
• Conversation history tracking

Case 2: Customer Service Intelligent Assistant

Scenario: E-commerce platform customer service chatbot

Implementation Focus:

• Real-time product information updates
• Order status query integration
• Sentiment analysis and human handoff
• Multi-turn conversation management

Case 3: Legal Document Analysis System

Scenario: Law firm case law search and analysis

Implementation Focus:

• Long document processing (court judgments)
• Citation network construction
• Time series analysis
• Professional terminology recognition

Common Issues and Solutions

Q1: How to handle very long documents?

Solution: Hierarchical Retrieval

# First retrieve chapters, then retrieve paragraphs
def hierarchical_retrieval(query, document_chunks):
    # Layer 1: Retrieve relevant chapters
    chapter_results = retrieve_chapters(query, top_k=3)

    # Layer 2: Retrieve paragraphs within relevant chapters
    paragraph_results = []
    for chapter in chapter_results:
        paragraphs = retrieve_paragraphs(query, chapter, top_k=5)
        paragraph_results.extend(paragraphs)

    return paragraph_results

Q2: How to reduce hallucination?

Solutions:

1. Strengthen prompt constraints
2. Add citation verification
3. Use self-consistency checks
4. Lower the temperature parameter

Q3: How to improve Chinese processing performance?

Solutions:

1. Use Chinese-specific embedding models (bge-large-zh)
2. Chinese word segmentation optimization
3. Traditional-Simplified Chinese conversion handling
4. Domain vocabulary enhancement

Conclusion

Key elements for building a production-grade RAG system:

1. Document Processing - Intelligent chunking and metadata enhancement
2. Embedding Model - Choosing the right vectorization approach
3. Retrieval Strategy - Hybrid Search + Re-ranking
4. Prompt Engineering - Structured instruction design
5. Continuous Optimization - Evaluate, test, iterate

At BASHCAT, we have extensive experience building RAG systems and have successfully delivered customized intelligent Q&A systems for multiple enterprises. If you are considering adopting RAG technology, feel free to contact us to discuss your requirements.

Further Reading

• LangChain Official Documentation
• RAGAS Evaluation Framework
• Qdrant Vector Database
• Pinecone RAG Guide