# Continuing with src/llm/azure_openai.
py
def generate_chat_completion(self,
messages: List[Dict[str, str]],
max_tokens: int = 1000,
temperature: float = 0.7) -> str:
"""Generate a chat completion for a list of messages"""
for attempt in range(self.max_retries):
try:
response = [Link](
engine=[Link],
messages=messages,
max_tokens=max_tokens,
temperature=temperature,
timeout=60 # Timeout in seconds
)
return [Link][0].[Link]
except ([Link],
[Link],
[Link]) as e:
if attempt == self.max_retries - 1:
[Link](f"Failed to generate completion after
{self.max_retries} attempts: {e}")
return f"Error generating response: {str(e)}"
# Exponential backoff
wait_time = (2 ** attempt) * 2
[Link](f"Retrying completion generation after {wait_time}s
due to: {e}")
[Link](wait_time)
def reformulate_query(self, original_query: str) -> str:
"""Reformulate a query to be more effective for retrieval"""
system_message = """Your task is to reformulate the given query to make it
more effective
for retrieval. Create a version that is clear, specific, and contains all
relevant keywords.
Return only the reformulated query without any explanations."""
messages = [
{"role": "system", "content": system_message},
{"role": "user", "content": original_query}
]
try:
return self.generate_chat_completion(
messages=messages,
max_tokens=100,
temperature=0.3
)
except Exception as e:
[Link](f"Error reformulating query: {e}")
return original_query
# src/reranker/[Link]
'''
from abc import ABC, abstractmethod
from typing import List, Dict, Any
class Reranker(ABC):
"""Base interface for rerankers"""
@abstractmethod
def rerank(self, query: str, documents: List[Dict[str, Any]]) -> List[Dict[str,
Any]]:
"""Rerank documents based on the query"""
pass
'''
# src/reranker/[Link]
'''
import logging
from typing import List, Dict, Any
from rank_bm25 import BM25Okapi
from .base import Reranker
logger = [Link](__name__)
class BM25Reranker(Reranker):
"""BM25 reranker implementation"""
def __init__(self):
"""Initialize the BM25 reranker"""
pass
def rerank(self, query: str, documents: List[Dict[str, Any]]) -> List[Dict[str,
Any]]:
"""Rerank documents using BM25 algorithm"""
if not documents:
return []
try:
# Tokenize documents
tokenized_corpus = [doc["content"].split() for doc in documents]
tokenized_query = [Link]()
# Create BM25 model
bm25 = BM25Okapi(tokenized_corpus)
# Get scores
scores = bm25.get_scores(tokenized_query)
# Add scores to documents
for i, doc in enumerate(documents):
doc["bm25_score"] = float(scores[i])
# Sort by score
[Link](key=lambda x: x["bm25_score"], reverse=True)
return documents
except Exception as e:
[Link](f"Error during BM25 reranking: {e}")
return documents # Return original documents on error
'''
# src/reranker/[Link]
'''
import logging
import time
import numpy as np
from typing import List, Dict, Any
from sentence_transformers import SentenceTransformer
from .base import Reranker
logger = [Link](__name__)
class SemanticReranker(Reranker):
"""Semantic reranker using sentence transformers"""
def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
"""Initialize the semantic reranker"""
try:
# Load model
[Link] = SentenceTransformer(model_name)
[Link](f"Loaded semantic reranker model: {model_name}")
except Exception as e:
[Link](f"Error loading semantic reranker model: {e}")
[Link] = None
def rerank(self, query: str, documents: List[Dict[str, Any]]) -> List[Dict[str,
Any]]:
"""Rerank documents using semantic similarity"""
if not documents or not [Link]:
return documents
try:
# Encode query
query_embedding = [Link](query)
# Encode documents
doc_embeddings = [Link]([doc["content"] for doc in
documents])
# Calculate cosine similarities
similarities = [self._cosine_similarity(query_embedding, doc_emb) for
doc_emb in doc_embeddings]
# Add similarities to documents
for i, doc in enumerate(documents):
doc["semantic_score"] = float(similarities[i])
# Sort by score
[Link](key=lambda x: x["semantic_score"], reverse=True)
return documents
except Exception as e:
[Link](f"Error during semantic reranking: {e}")
return documents # Return original documents on error
def _cosine_similarity(self, vec1: [Link], vec2: [Link]) -> float:
"""Calculate cosine similarity between two vectors"""
similarity = [Link](vec1, vec2) / ([Link](vec1) *
[Link](vec2))
return float(similarity)
'''
# src/reranker/[Link]
'''
import logging
from typing import List, Dict, Any
from .base import Reranker
from .bm25 import BM25Reranker
from .semantic import SemanticReranker
logger = [Link](__name__)
class HybridReranker(Reranker):
"""Hybrid reranker combining BM25 and semantic reranking"""
def __init__(self, bm25_weight: float = 0.3, semantic_weight: float = 0.7):
"""Initialize the hybrid reranker"""
self.bm25_reranker = BM25Reranker()
self.semantic_reranker = SemanticReranker()
self.bm25_weight = bm25_weight
self.semantic_weight = semantic_weight
def rerank(self, query: str, documents: List[Dict[str, Any]]) -> List[Dict[str,
Any]]:
"""Rerank documents using both BM25 and semantic similarity"""
if not documents:
return []
try:
# First pass: BM25
documents = self.bm25_reranker.rerank(query, documents)
# Second pass: Semantic
documents = self.semantic_reranker.rerank(query, documents)
# Combine scores
for doc in documents:
doc["hybrid_score"] = (
self.bm25_weight * ([Link]("bm25_score", 0) /
max([Link]("bm25_score", 0.001) for d in documents)) +
self.semantic_weight * ([Link]("semantic_score", 0) /
max([Link]("semantic_score", 0.001) for d in documents))
)
# Sort by hybrid score
[Link](key=lambda x: x["hybrid_score"], reverse=True)
return documents
except Exception as e:
[Link](f"Error during hybrid reranking: {e}")
return documents # Return original documents on error
'''
# src/rag/[Link]
'''
from abc import ABC, abstractmethod
from typing import Dict, Any
class RAGPipeline(ABC):
"""Base interface for RAG pipelines"""
@abstractmethod
def process(self, query: str) -> Dict[str, Any]:
"""Process a query through the RAG pipeline"""
pass
'''
# src/rag/[Link]
'''
import logging
from typing import Dict, Any, List
from config import settings
from [Link].azure_openai import AzureOpenAIEmbeddings
from src.vector_store.azure_search import AzureCognitiveSearchVectorStore
from [Link].azure_openai import AzureOpenAILLM
from [Link] import HybridReranker
from [Link].memory_cache import InMemorySemanticCache
from .base import RAGPipeline
logger = [Link](__name__)
class EnhancedRAGPipeline(RAGPipeline):
"""Enhanced RAG pipeline with query reformulation, reranking, and caching"""
def __init__(self):
"""Initialize the enhanced RAG pipeline"""
# Initialize components
self.embedding_provider = AzureOpenAIEmbeddings()
self.vector_store = AzureCognitiveSearchVectorStore()
[Link] = AzureOpenAILLM()
[Link] = HybridReranker()
[Link] = InMemorySemanticCache()
# Settings
self.retrieval_top_k = settings.retrieval_top_k
self.reranking_enabled = settings.reranking_enabled
self.max_tokens = settings.max_tokens
def process(self, query: str) -> Dict[str, Any]:
"""Process a query through the RAG pipeline"""
# 1. Generate embedding for the query
query_embedding = self.embedding_provider.generate_embedding(query)
# 2. Check cache for similar queries
cached_result, similarity = [Link].find_similar(query, query_embedding)
if cached_result:
[Link](f"Cache hit with similarity: {similarity}")
return {
"response": cached_result["response"],
"source_documents": cached_result["source_documents"],
"cached": True,
"similarity": similarity
}
# 3. Reformulate the query
reformulated_query = [Link].reformulate_query(query)
[Link](f"Reformulated query: {reformulated_query}")
# 4. Get embedding for reformulated query
reformulated_embedding =
self.embedding_provider.generate_embedding(reformulated_query)
# 5. Retrieve documents
documents = self.vector_store.search(reformulated_query,
reformulated_embedding, self.retrieval_top_k)
# 6. Rerank documents if enabled
if self.reranking_enabled and documents:
documents = [Link](query, documents)
# 7. Format context from top documents
context = self._format_context(documents)
# 8. Generate response using LLM
response = self._generate_response(query, context)
# 9. Format result
result = {
"response": response,
"source_documents": documents,
"cached": False
}
# 10. Cache the result
[Link](query, query_embedding, result)
return result
def _format_context(self, documents: List[Dict[str, Any]]) -> str:
"""Format the context from retrieved documents"""
if not documents:
return "No relevant information found."
formatted_docs = []
for i, doc in enumerate(documents):
formatted_docs.append(
f"Document {i+1} (Source: {doc['source']}):\n{doc['content']}"
)
return "\n\n".join(formatted_docs)
def _generate_response(self, query: str, context: str) -> str:
"""Generate a response using the LLM"""
system_message = f"""You are an AI assistant answering questions based on
the provided context.
Use only the information in the context to answer questions. If the
information is not in the context,
say that you don't have enough information to answer accurately.
Context:
{context}
"""
messages = [
{"role": "system", "content": system_message},
{"role": "user", "content": query}
]
return [Link].generate_chat_completion(
messages=messages,
max_tokens=self.max_tokens,
temperature=0.7
)
'''
# src/utils/[Link]
'''
import logging
import time
from functools import wraps
from opentelemetry import trace
from [Link] import SERVICE_NAME, Resource
from [Link] import TracerProvider
from [Link] import BatchSpanProcessor, ConsoleSpanExporter
from [Link] import Status, StatusCode
# Configure logging
[Link](
level=[Link],
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# Initialize OpenTelemetry
resource = Resource(attributes={SERVICE_NAME: "azure-rag-service"})
tracer_provider = TracerProvider(resource=resource)
# Add console exporter for development
console_processor = BatchSpanProcessor(ConsoleSpanExporter())
tracer_provider.add_span_processor(console_processor)
trace.set_tracer_provider(tracer_provider)
# Get a tracer
tracer = trace.get_tracer(__name__)
def trace_function(name=None):
"""Decorator to trace function execution"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# Use function name if name not provided
span_name = name or func.__name__
with tracer.start_as_current_span(span_name) as span:
# Add function arguments as span attributes
# Be careful not to include sensitive information
for i, arg in enumerate(args):
if isinstance(arg, (str, int, float, bool)):
span.set_attribute(f"arg_{i}", str(arg))
for key, value in [Link]():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(f"kwarg_{key}", str(value))
# Record start time
start_time = [Link]()
try:
# Execute the function
result = func(*args, **kwargs)
# Set status to success
span.set_status(Status([Link]))
return result
except Exception as e:
# Record exception
span.record_exception(e)
span.set_status(Status([Link]), str(e))
# Re-raise the exception
raise
finally:
# Record execution time
execution_time = [Link]() - start_time
span.set_attribute("execution_time_ms", execution_time * 1000)
return wrapper
return decorator
'''
# src/utils/[Link]
'''
import logging
from typing import List, Dict, Any
from [Link] import precision_score, recall_score, f1_score
logger = [Link](__name__)
class RAGEvaluator:
"""Evaluate RAG pipeline performance"""
def __init__(self):
"""Initialize the RAG evaluator"""
pass
def evaluate(self,
test_queries: List[str],
expected_docs: List[List[str]],
pipeline) -> Dict[str, float]:
"""Evaluate RAG pipeline on a test set"""
results = {
"precision": 0.0,
"recall": 0.0,
"f1": 0.0,
"mrr": 0.0,
"latency": 0.0
}
query_results = []
for query, expected in zip(test_queries, expected_docs):
# Process query
import time
start_time = [Link]()
result = [Link](query)
end_time = [Link]()
# Record latency
query_latency = end_time - start_time
# Get retrieved document IDs
retrieved_docs = [doc["id"] for doc in result["source_documents"]]
# Calculate metrics
retrieval_metrics = self._calculate_retrieval_metrics(retrieved_docs,
expected)
query_results.append({
"query": query,
"expected": expected,
"retrieved": retrieved_docs,
"latency": query_latency,
**retrieval_metrics
})
# Aggregate results
if query_results:
results["precision"] = sum(r["precision"] for r in query_results) /
len(query_results)
results["recall"] = sum(r["recall"] for r in query_results) /
len(query_results)
results["f1"] = sum(r["f1"] for r in query_results) /
len(query_results)
results["mrr"] = sum(r["mrr"] for r in query_results) /
len(query_results)
results["latency"] = sum(r["latency"] for r in query_results) /
len(query_results)
return results
def _calculate_retrieval_metrics(self,
retrieved_docs: List[str],
expected_docs: List[str]) -> Dict[str, float]:
"""Calculate retrieval metrics for a single query"""
# Precision, Recall, F1
true_positives = len(set(retrieved_docs) & set(expected_docs))
if len(retrieved_docs) == 0:
precision = 0.0
else:
precision = true_positives / len(retrieved_docs)
if len(expected_docs) == 0:
recall = 1.0 if len(retrieved_docs) == 0 else 0.0
else:
recall = true_positives / len(expected_docs)
if precision + recall == 0:
f1 = 0.0
else:
f1 = 2 * precision * recall / (precision + recall)
# Mean Reciprocal Rank
mrr = 0.0
for i, doc_id in enumerate(retrieved_docs):
if doc_id in expected_docs:
mrr = 1.0 / (i + 1)
break
return {
"precision": precision,
"recall": recall,
"f1": f1,
"mrr": mrr
}
'''
# [Link]
'''
import os
import logging
from flask import Flask, request, jsonify, render_template
from flask_jwt_extended import JWTManager, jwt_required, create_access_token,
get_jwt_identity
from [Link] import secure_filename
from [Link] import EnhancedRAGPipeline
from [Link] import DocumentLoader
from [Link] import DocumentProcessor
from [Link].azure_openai import AzureOpenAIEmbeddings
from src.vector_store.azure_search import AzureCognitiveSearchVectorStore
from config import settings
# Configure logging
[Link](
level=getattr(logging, settings.log_level),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# Initialize Flask app
app = Flask(__name__)
[Link]["JWT_SECRET_KEY"] = settings.api_jwt_secret
jwt = JWTManager(app)
# Initialize components
document_loader = DocumentLoader()
document_processor = DocumentProcessor()
embedding_provider = AzureOpenAIEmbeddings()
vector_store = AzureCognitiveSearchVectorStore()
rag_pipeline = EnhancedRAGPipeline()
# Configure upload folder
UPLOAD_FOLDER = 'uploads'
[Link](UPLOAD_FOLDER, exist_ok=True)
[Link]['UPLOAD_FOLDER'] = UPLOAD_FOLDER
@[Link]('/')
def home():
return render_template('[Link]')
@[Link]('/api/login', methods=['POST'])
def login():
if not settings.api_auth_enabled:
return jsonify({"error": "Authentication is not enabled"}), 400
username = [Link]('username', None)
password = [Link]('password', None)
# Replace with your actual authentication logic
if username == 'admin' and password == 'password':
access_token = create_access_token(identity=username)
return jsonify(access_token=access_token)
return jsonify({"error": "Invalid credentials"}), 401
@[Link]('/api/query', methods=['POST'])
def query():
# Check authentication if enabled
if settings.api_auth_enabled:
auth_header = [Link]('Authorization')
if not auth_header or not auth_header.startswith('Bearer '):
return jsonify({'error': 'Authentication required'}), 401
# Get query from request
data = [Link]
user_query = [Link]('query', '')
if not user_query:
return jsonify({'error': 'No query provided'}), 400
try:
# Process query through RAG pipeline
result = rag_pipeline.process(user_query)
return jsonify(result)
except Exception as e:
[Link](f"Error processing query: {e}")
return jsonify({'error': str(e)}), 500
@[Link]('/api/documents', methods=['POST'])
def upload_document():
# Check if the post request has the file part
if 'file' not in [Link]:
return jsonify({'error': 'No file part'}), 400
file = [Link]['file']
if [Link] == '':
return jsonify({'error': 'No selected file'}), 400
try:
# Save the file
filename = secure_filename([Link])
filepath = [Link]([Link]['UPLOAD_FOLDER'], filename)
[Link](filepath)
# Load the document
document = document_loader.load_from_file(filepath)
if not document:
return jsonify({'error': 'Failed to load document'}), 500
# Process document into chunks
chunks = document_processor.process_document(document)
# Generate embeddings for chunks
for chunk in chunks:
chunk['contentVector'] =
embedding_provider.generate_embedding(chunk['content'])
# Index chunks in vector store
vector_store.index_documents(chunks)
return jsonify({
'message': f'Document {filename} processed successfully',
'chunks': len(chunks)
})
except Exception as e:
[Link](f"Error processing document: {e}")
return jsonify({'error': str(e)}), 500
@[Link]('/api/documents/text', methods=['POST'])
def add_text_document():
data = [Link]
text = [Link]('text', '')
title = [Link]('title', 'Untitled')
source = [Link]('source', 'Text Input')
if not text:
return jsonify({'error': 'No text provided'}), 400
try:
# Create document from text
document = document_loader.load_from_text(text, title, source)
# Process document into chunks
chunks = document_processor.process_document(document)
# Generate embeddings for chunks
for chunk in chunks:
chunk['contentVector'] =
embedding_provider.generate_embedding(chunk['content'])
# Index chunks in vector store
vector_store.index_documents(chunks)
return jsonify({
'message': f'Text document processed successfully',
'chunks': len(chunks)
})
except Exception as e:
[Link](f"Error processing text document: {e}")
return jsonify({'error': str(e)}), 500
if __name__ == '__main__':
[Link](debug=[Link])
'''
# templates/[Link]
'''
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Azure OpenAI RAG System</title>
<link href="[Link]
[Link]" rel="stylesheet">
<link rel="stylesheet" href="{{ url_for('static', filename='[Link]') }}">
</head>
<body>
<div class="container mt-5">
<h1 class="mb-4">Azure OpenAI RAG System</h1>
<div class="row">
<div class="col-md-8">
<div class="card mb-4">
<div class="card-body">
<h5 class="card-title">Ask a Question</h5>
<div class="mb-3">
<textarea id="queryInput" class="form-control" rows="3"
placeholder="Enter your question here..."></textarea>
</div>
<button id="submitBtn" class="btn
btn-primary">Submit</button>
<div id="loading" class="loading mt-3" style="display:
none;">
<div class="d-flex align-items-center">
<div class="spinner-border text-primary me-2"
role="status">
<span class="visually-hidden">Loading...</span>
</div>
<span>Processing your question...</span>
</div>
</div>
<div id="responseContainer" class="response-container mt-4"
style="display: none;">
<h5>Response:</h5>
<div id="responseText" class="mb-3"></div>
<div id="queryInfo" class="small text-muted
mb-3"></div>
<div id="sourcesContainer">
<h6>Sources:</h6>
<div id="sourcesContent"></div>
</div>
</div>
</div>
</div>
</div>
<div class="col-md-4">
<div class="card mb-4">
<div class="card-body">
<h5 class="card-title">Document Upload</h5>
<div class="mb-3">
<input type="file" id="fileInput" class="form-control">
</div>
<button id="uploadBtn" class="btn btn-secondary">Upload
Document</button>
<div id="uploadStatus" class="mt-2"></div>
</div>
</div>
<div class="card">
<div class="card-body">
<h5 class="card-title">Add Text Document</h5>
<div class="mb-3">
<input type="text" id="titleInput" class="form-control
mb-2" placeholder="Document Title">
<textarea id="textInput" class="form-control" rows="5"
placeholder="Enter text content..."></textarea>
</div>
<button id="addTextBtn" class="btn btn-secondary">Add
Text</button>
<div id="textStatus" class="mt-2"></div>
</div>
</div>
</div>
</div>
</div>
<script>
[Link]('submitBtn').addEventListener('click', async () =>
{
const query = [Link]('queryInput').[Link]();
if (!query) return;
// Show loading indicator
[Link]('loading').[Link] = 'block';
[Link]('responseContainer').[Link] = 'none';
try {
const response = await fetch('/api/query', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: [Link]({ query })
});
if (![Link]) {
throw new Error(`HTTP error ${[Link]}`);
}
const result = await [Link]();
// Hide loading indicator
[Link]('loading').[Link] = 'none';
// Display response
[Link]('responseText').textContent =
[Link];
// Display query info
let queryInfoText = '';
if ([Link]) {
queryInfoText += `Result from cache (similarity: $
{([Link] * 100).toFixed(1)}%)`;
}
[Link]('queryInfo').textContent = queryInfoText;
// Display sources
const sourcesContent = [Link]('sourcesContent');
[Link] = '';
if (result.source_documents && result.source_documents.length > 0)
{
result.source_documents.forEach((doc, index) => {
const sourceDiv = [Link]('div');
[Link] = 'source-document p-2 mb-2 border
rounded';
let sourceHeader = [Link]('h6');
[Link] = `${index + 1}. ${[Link] ||
'Untitled Document'}`;
let sourceBody = [Link]('div');
[Link] = `
<div><small>Source: ${[Link] ||
'Unknown'}</small></div>
<div class="mt-1">${[Link](0, 200)}$
{[Link] > 200 ? '...' : ''}</div>
`;
[Link](sourceHeader);
[Link](sourceBody);
[Link](sourceDiv);
});
} else {
[Link] = 'No source documents available.';
}
[Link]('responseContainer').[Link] =
'block';
} catch (error) {
[Link]('Error:', error);
[Link]('loading').[Link] = 'none';
alert('An error occurred while processing your request.');
}
});
// Document upload functionality
[Link]('uploadBtn').addEventListener('click', async () =>
{
const fileInput = [Link]('fileInput');
const file = [Link][0];
if (!file) {
alert('Please select a file