Knowledge Grounding with RAG & Semantic Search

Transform Agent1 from a generic chatbot into a product expert.

Knowledge Indexing & Retrieval

Ground AI responses in real product data.

Deployment is configured (Deployment Configuration module). Safety is enabled (Agent Safety & Memory module). Now Agent1 needs knowledge: detailed product information, pricing, policies, and specs. RAG (Retrieval Augmented Generation) makes this possible.

---

🎯 What This Section Covers

Without RAG, Agent1 is limited to general knowledge baked into the model. With RAG, it becomes an expert on Contoso products.

The Problem

Customer asks: "What's the warranty on the Nike Air-style running shoes?" Agent doesn't know. Model has general knowledge but no Contoso-specific details.

The Solution

RAG indexes product specs (warranty: 1 year, care instructions, sizing guide). Agent retrieves relevant docs and answers with confidence and accuracy.

---

🏗️ RAG Architecture Overview

---

� Step 1: Document Chunking

Split large documents into searchable chunks:

import re

def chunk_document(text: str, chunk_size: int = 500, overlap: int = 100) -> list:
    """Split document into overlapping chunks for semantic search"""
    
    # Split by sentences first
    sentences = re.split(r'(?<=[.!?])\s+', text)
    
    chunks = []
    current_chunk = []
    current_length = 0
    
    for sentence in sentences:
        sentence_length = len(sentence.split())
        
        if current_length + sentence_length > chunk_size and current_chunk:
            # Save current chunk
            chunk_text = ' '.join(current_chunk)
            chunks.append(chunk_text)
            
            # Start new chunk with overlap
            overlap_sentences = current_chunk[-max(1, len(current_chunk)//3):]
            current_chunk = overlap_sentences + [sentence]
            current_length = sum(len(s.split()) for s in current_chunk)
        else:
            current_chunk.append(sentence)
            current_length += sentence_length
    
    # Add final chunk
    if current_chunk:
        chunks.append(' '.join(current_chunk))
    
    return chunks

# Usage
product_spec = """
Blue Running Shoes - Model XYZ
Features: Breathable mesh, cushioned sole, lightweight
Warranty: 1 year from purchase date
Coverage: Manufacturing defects only
Exclusions: Normal wear, damage from misuse
Sizing: Available in sizes 6-13
Care Instructions: Hand wash only
"""

chunks = chunk_document(product_spec, chunk_size=50)
# Result: ["Blue Running Shoes - Model XYZ\nFeatures: Breathable...", "..."]

---

🔢 Step 2: Generate Embeddings

Convert text chunks to vectors using Azure OpenAI:

from azure.ai.openai import AzureOpenAI
from azure.identity import DefaultAzureCredential

class EmbeddingGenerator:
    def __init__(self):
        self.credential = DefaultAzureCredential()
        self.client = AzureOpenAI(
            api_version="2024-08-01-preview",
            endpoint="https://your-openai.openai.azure.com/",
            credential=self.credential
        )
    
    async def embed_text(self, text: str) -> list:
        """Convert text to 3072-dimensional embedding"""
        
        response = self.client.embeddings.create(
            model="text-embedding-3-large",  # 3072 dimensions
            input=text
        )
        
        return response.data[0].embedding
    
    async def embed_batch(self, texts: list) -> list:
        """Convert multiple texts to embeddings"""
        
        response = self.client.embeddings.create(
            model="text-embedding-3-large",
            input=texts
        )
        
        return [item.embedding for item in response.data]

# Usage
embedder = EmbeddingGenerator()

# Single embedding
embedding = await embedder.embed_text("Blue running shoes with warranty")
# Result: [0.0123, -0.0456, ..., 0.0234]  # 3072 numbers

# Batch embeddings (more efficient for many chunks)
chunks = ["Warranty: 1 year", "Size: 6-13", "Care: Hand wash"]
embeddings = await embedder.embed_batch(chunks)
# Result: [[...], [...], [...]]  # 3 embeddings

---

📇 Step 3: Create Vector Index in Azure AI Search

from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex, SearchField, SearchFieldDataType,
    SimpleField, SearchableField, VectorSearch,
    HnswAlgorithmConfiguration, VectorSearchProfile, SemanticConfiguration,
    SemanticField, SemanticPrioritizedFields, SemanticSearch
)
from azure.identity import DefaultAzureCredential

async def create_vector_index(index_name: str = "products"):
    """Create Azure AI Search index with vector fields"""
    
    credential = DefaultAzureCredential()
    client = SearchIndexClient(
        endpoint="https://your-search.search.windows.net",
        credential=credential
    )
    
    fields = [
        SimpleField(
            name="id",
            type=SearchFieldDataType.String,
            key=True
        ),
        SearchableField(
            name="chunk_text",
            type=SearchFieldDataType.String,
            searchable=True,
            retrievable=True
        ),
        SearchField(
            name="chunk_vector",
            type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
            searchable=True,
            vector_search_dimensions=3072,
            vector_search_profile_name="myHnsw"
        ),
        SimpleField(
            name="product_id",
            type=SearchFieldDataType.String,
            filterable=True
        ),
        SimpleField(
            name="product_category",
            type=SearchFieldDataType.String,
            filterable=True
        ),
        SearchableField(
            name="source_document",
            type=SearchFieldDataType.String,
            retrievable=True
        ),
    ]
    
    vector_search_config = VectorSearch(
        algorithms=[
            HnswAlgorithmConfiguration(name="myHnsw")
        ],
        profiles=[
            VectorSearchProfile(
                name="myHnsw",
                algorithm_configuration_name="myHnsw",
            )
        ]
    )
    
    semantic_config = SemanticConfiguration(
        name="my-semantic-config",
        prioritized_fields=SemanticPrioritizedFields(
            content_fields=[SemanticField(field_name="chunk_text")]
        )
    )
    
    semantic_search_config = SemanticSearch(configurations=[semantic_config])
    
    index = SearchIndex(
        name=index_name,
        fields=fields,
        vector_search=vector_search_config,
        semantic_search=semantic_search_config
    )
    
    try:
        result = client.create_index(index)
        print(f"✅ Created index: {result.name}")
    except Exception as e:
        print(f"Index already exists or error: {e}")
    
    return client

# Usage
client = await create_vector_index("products")

---

💾 Step 4: Upload Chunks with Embeddings

from azure.search.documents import SearchClient
from azure.identity import DefaultAzureCredential
import uuid

async def index_documents(index_name: str, product_docs: list):
    """Upload documents with embeddings to vector index"""
    
    credential = DefaultAzureCredential()
    client = SearchClient(
        endpoint="https://your-search.search.windows.net",
        index_name=index_name,
        credential=credential
    )
    
    embedder = EmbeddingGenerator()
    
    documents_to_index = []
    
    for doc in product_docs:
        # Extract or chunk the document
        chunks = chunk_document(doc["content"])
        
        # Generate embeddings for each chunk
        embeddings = await embedder.embed_batch(chunks)
        
        # Create document for each chunk
        for chunk, embedding in zip(chunks, embeddings):
            index_doc = {
                "id": str(uuid.uuid4()),
                "chunk_text": chunk,
                "chunk_vector": embedding,
                "product_id": doc["product_id"],
                "product_category": doc["category"],
                "source_document": doc["source"]
            }
            documents_to_index.append(index_doc)
    
    # Batch upload
    result = client.upload_documents(documents_to_index)
    print(f"✅ Uploaded {len(documents_to_index)} chunks")
    
    return result

# Usage
products = [
    {
        "product_id": "shoe-blue-001",
        "category": "footwear",
        "source": "blue-running-shoes.pdf",
        "content": """Blue Running Shoes - Model XYZ
Features: Breathable mesh, cushioned sole, lightweight
Warranty: 1 year from purchase date..."""
    },
    {
        "product_id": "jacket-winter-001",
        "category": "apparel",
        "source": "winter-jacket.pdf",
        "content": """Winter Jacket - Premium insulated model..."""
    }
]

await index_documents("products", products)

---

🔍 Step 5: Semantic Search at Query Time

from azure.search.documents.models import VectorizedQuery

async def retrieve_with_rag(query: str, index_name: str = "products", top_k: int = 5) -> list:
    """Search vector index and retrieve relevant documents"""
    
    credential = DefaultAzureCredential()
    search_client = SearchClient(
        endpoint="https://your-search.search.windows.net",
        index_name=index_name,
        credential=credential
    )
    
    embedder = EmbeddingGenerator()
    
    # Convert query to embedding
    query_embedding = await embedder.embed_text(query)
    
    # Hybrid search: both vector + keyword
    results = search_client.search(
        search_text=query,  # Keyword search
        vector_queries=[
            VectorizedQuery(
                vector=query_embedding,
                k_nearest_neighbors=top_k,
                fields="chunk_vector"
            )
        ],
        select=["chunk_text", "product_id", "source_document", "@search.score"],
        top=top_k
    )
    
    retrieved_docs = []
    for result in results:
        retrieved_docs.append({
            "content": result["chunk_text"],
            "product_id": result["product_id"],
            "relevance": result["@search.score"],
            "source": result["source_document"]
        })
    
    return retrieved_docs

# Usage
query = "What's the warranty on blue shoes?"
results = await retrieve_with_rag(query)
# Result: [
#   {"content": "Warranty: 1 year from purchase...", "relevance": 0.87, ...},
#   {"content": "Coverage: Manufacturing defects...", "relevance": 0.82, ...}
# ]

---

🤖 Step 6: Full RAG Agent Loop

async def agent_with_rag(customer_id: str, question: str) -> str:
    """Complete RAG flow: retrieve + ground + generate"""
    
    # 1. RETRIEVE KNOWLEDGE
    print("🔍 Retrieving knowledge...")
    knowledge_docs = await retrieve_with_rag(question, index_name="products", top_k=3)
    
    if not knowledge_docs:
        return "I don't have information about that. Let me connect you with a specialist."
    
    # 2. BUILD CONTEXT
    print("📝 Building response context...")
    knowledge_text = "\n\n".join([
        f"Source: {doc['source']}\n{doc['content']}"
        for doc in knowledge_docs
    ])
    
    # 3. GET CONVERSATION HISTORY (from memory article)
    memory = ConversationMemory()
    await memory.init()
    conversation_context = await memory.build_memory_context(customer_id)
    
    # 4. BUILD AGENT PROMPT WITH GROUNDING
    system_prompt = f"""You are a product support agent.

{conversation_context}

PRODUCT KNOWLEDGE (from our documentation):
{knowledge_text}

CRITICAL RULES:
- Answer ONLY based on the knowledge above
- Cite the source when providing specs
- If information contradicts knowledge base, trust the knowledge base
- Be helpful but honest - don't make up details
- If unsure, offer to escalate"""
    
    # 5. GENERATE GROUNDED RESPONSE
    print("🤖 Generating response...")
    client = AzureOpenAI(...)  # Initialize
    
    response = client.chat.completions.create(
        model="gpt-4o",
        system=system_prompt,
        messages=[
            {"role": "user", "content": question}
        ]
    )
    
    answer = response.choices[0].message.content
    
    # 6. SAVE TO MEMORY
    print("💾 Saving to memory...")
    await memory.save_turn(
        customer_id,
        question,
        answer,
        metadata={
            "sources": [d["source"] for d in knowledge_docs],
            "relevance_scores": [d["relevance"] for d in knowledge_docs]
        }
    )
    
    return answer

# Usage
response = await agent_with_rag("cust_123", "What's the warranty on blue shoes?")
# Response will be grounded in actual product documentation

---

✅ Complete Example: End-to-End Setup

async def setup_and_run_rag():
    """Complete RAG pipeline setup and execution"""
    
    # 1. Create vector index
    print("📇 Creating vector index...")
    await create_vector_index("products")
    
    # 2. Index documents
    print("📑 Indexing product documents...")
    sample_products = [
        {
            "product_id": "shoe-blue-001",
            "category": "footwear",
            "source": "shoes.pdf",
            "content": "Blue Running Shoes. Warranty: 1 year. Size 6-13. Hand wash only."
        }
    ]
    await index_documents("products", sample_products)
    
    # 3. Test retrieval
    print("🔍 Testing retrieval...")
    results = await retrieve_with_rag("warranty shoes")
    print(f"Found {len(results)} relevant documents")
    
    # 4. Run agent
    print("🤖 Running agent with RAG...")
    answer = await agent_with_rag("customer_1", "What's the warranty?")
    print(f"Agent: {answer}")

# Execute
asyncio.run(setup_and_run_rag())

### 2. **Chunking Strategy**

Documents can't be indexed whole. They must be split into **semantic chunks** (not random splits).

```python
# ❌ WRONG: Random chunking (every 256 characters)
"The Blue Running Shoes (Model BRS-2024) come with a 1-year manufacturer's 
warranty covering defects in materials and workmanship. The warranty does not 
cover wear and tear, damage from accidents, or improper storage. | Sizing guide: 
US 6-13, EU 36-48, UK 3-12. These shoes use our patented ComfortFoam insole 
technology, designed for long-distance running..."

# ✅ CORRECT: Semantic chunking (complete thoughts)
# Chunk 1: Product overview
"The Blue Running Shoes (Model BRS-2024) come with a 1-year manufacturer's 
warranty covering defects in materials and workmanship. The warranty does not 
cover wear and tear, damage from accidents, or improper storage."

# Chunk 2: Sizing information
"Sizing guide: US 6-13, EU 36-48, UK 3-12. 
Fit: True to size. If between sizes, size up for comfort on long runs."

# Chunk 3: Technology details
"These shoes use our patented ComfortFoam insole technology, designed for 
long-distance running. The breathable mesh upper keeps feet cool, and the 
reinforced heel provides stability on uneven terrain."

3. Embeddings: Converting Text to Vectors

An embedding is a numerical representation of text that captures semantic meaning.

from azure.openai import AzureOpenAI

client = AzureOpenAI(
    api_version="2024-06-01",
    endpoint="https://contoso-openai.openai.azure.com/",
    credential=credential  # Managed Identity
)

# Convert text to vector (embedding)
embedding_response = client.embeddings.create(
    model="text-embedding-3-large",  # Contoso's embedding model
    input="Blue Running Shoes with 1-year warranty, ComfortFoam insole"
)

# Result: 3072-dimensional vector
vector = embedding_response.data[0].embedding
# [0.0123, -0.0456, 0.0789, ..., 0.0234]  # 3072 numbers

Why embeddings matter:

• ✅ Captures semantic meaning (similar texts → similar vectors)
• ✅ Enables fast similarity search (vector distance, not text matching)
• ✅ Handles synonyms ("shoes" ≈ "footwear" ≈ "sneakers")

4. Vector Index: Azure AI Search

Azure AI Search stores vectors and enables semantic search.

{
  "index_name": "contoso-products",
  "fields": [
    {
      "name": "id",
      "type": "Edm.String",
      "key": true
    },
    {
      "name": "chunk_text",
      "type": "Edm.String",
      "searchable": true
    },
    {
      "name": "chunk_vector",
      "type": "Collection(Edm.Single)",
      "searchable": true,
      "vector": {
        "dimensions": 3072,
        "metric": "cosine"
      }
    },
    {
      "name": "product_id",
      "type": "Edm.String",
      "filterable": true
    },
    {
      "name": "product_category",
      "type": "Edm.String",
      "filterable": true
    },
    {
      "name": "source_document",
      "type": "Edm.String"
    }
  ]
}

5. Retrieval: Vector Search at Query Time

When customer asks a question, Agent1 searches the vector index.

async def retrieve_product_knowledge(query: str, top_k: int = 5):
    """Retrieve relevant product documents for a customer question"""
    
    # Step 1: Convert query to embedding
    query_embedding = client.embeddings.create(
        model="text-embedding-3-large",
        input=query
    ).data[0].embedding
    
    # Step 2: Search vector index
    search_results = search_client.search(
        search_text=query,
        vector_queries=[
            VectorizedQuery(
                vector=query_embedding,
                k_nearest_neighbors=top_k,
                fields="chunk_vector"
            )
        ],
        select=["chunk_text", "product_id", "source_document"]
    )
    
    # Step 3: Return results
    retrieved_docs = [result["chunk_text"] for result in search_results]
    return retrieved_docs

---

🔄 RAG Flow: End-to-End Example

Customer Interaction

CUSTOMER: "What's the warranty on blue running shoes? Do they cover damage?"

↓

AGENT1 PROCESSING:

1. Convert question to vector
   Query: "What's the warranty on blue running shoes? Do they cover damage?"
   → Vector: [0.0234, -0.0567, ..., 0.0891]  (3072 dimensions)

2. Search vector index
   Find top-5 documents similar to this question
   
3. Retrieved documents:
   Doc 1: "The Blue Running Shoes (Model BRS-2024) come with a 1-year 
           manufacturer's warranty covering defects in materials and 
           workmanship."
   
   Doc 2: "The warranty does not cover wear and tear, damage from accidents, 
           or improper storage."
   
   Doc 3: "If you experience any manufacturing defects, contact our support 
           team with proof of purchase for warranty service."

4. Generate response
   Agent reads retrieved docs + question
   Generates grounded answer:

↓

AGENT RESPONSE: "Blue Running Shoes come with a 1-year manufacturer's warranty 
that covers defects in materials and workmanship. However, it does NOT cover 
wear and tear or accident damage. If you find a manufacturing defect, contact 
us with proof of purchase and we'll help you out."

✅ ACCURATE: Grounded in real product data
✅ COMPLETE: All warranty details included
✅ SAFE: No hallucinations, no false claims

---

📊 Hybrid Search: Keyword + Vector

Vector search alone isn't always perfect. Hybrid search combines both approaches.

When to use each:

Search Type	Best For	Example
Keyword	Exact matches, specific terms	"Model BRS-2024", "Contoso" product code
Vector	Semantic meaning, synonyms	"How comfy are the shoes?" (finds comfort-related docs)
Hybrid	General queries	"What's the warranty?" (needs both keyword + semantic)

---

🛡️ Safety During Indexing

Contoso's product sheets are scanned for malicious content before indexing.

async def index_product_document_safely(pdf_path: str):
    """Index a product document, checking for embedded threats"""
    
    # Step 1: Extract text and images from PDF
    document = extract_pdf(pdf_path)
    
    # Step 2: Scan images for embedded instructions
    for image in document.images:
        safety_result = safety_evaluator.evaluate_image(image)
        if not safety_result.safe:
            logger.warning(f"Unsafe image in {pdf_path}: {safety_result.reason}")
            # Skip this image, or mark it for review
            continue
    
    # Step 3: Chunk and embed text
    chunks = semantic_chunker.chunk(document.text)
    embeddings = [embed_text(chunk) for chunk in chunks]
    
    # Step 4: Index
    for chunk, embedding in zip(chunks, embeddings):
        search_index.upsert({
            "id": str(uuid.uuid4()),
            "chunk_text": chunk,
            "chunk_vector": embedding,
            "source_document": pdf_path,
            "indexed_at": datetime.now().isoformat()
        })
    
    logger.info(f"Safely indexed {pdf_path}: {len(chunks)} chunks")

---

📊 Integration with Agent1

How does Agent1 use RAG?

async def agent_query_with_rag(user_message: str):
    """Agent1 answers questions using RAG knowledge base"""
    
    # Step 1: Retrieve relevant product knowledge
    retrieved_docs = await retrieve_product_knowledge(
        query=user_message,
        top_k=5
    )
    
    # Step 2: Build grounded prompt
    context = "\n".join(retrieved_docs)
    
    system_prompt = f"""You are Agent1, Contoso's customer support agent.
    
Use the product information below to answer customer questions.
Only answer about Contoso products. If information isn't in the 
knowledge base, say so and offer alternatives.

PRODUCT KNOWLEDGE BASE:
{context}

CRITICAL RULE: Never make up product information. If unsure, 
offer to connect with a specialist.
"""
    
    # Step 3: Generate answer grounded in documents
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_message}
        ],
        temperature=0.7
    )
    
    # Step 4: Save to conversation memory
    await save_to_memory(
        customer_id=user_id,
        question=user_message,
        answer=response.choices[0].message.content,
        knowledge_sources=retrieved_docs
    )
    
    return response.choices[0].message.content

---

🎯 Configuration Summary

Component	Setting	Impact
Embedding Model	text-embedding-3-large	High-quality semantic understanding, 3072 dimensions
Vector Index	Azure AI Search	Enterprise-scale, filters, metadata support
Chunking Size	256-512 tokens per chunk	Balance between context and granularity
Retrieval Top-K	5 documents	Enough context for comprehensive answers
Search Type	Hybrid (keyword + vector)	Best accuracy for varied query types
Update Frequency	Weekly	New products indexed immediately, daily refresh

---

📚 What's Next?

Explore these RAG and knowledge indexing topics as detailed guides are added:

• RAG Fundamentals - Deep dive into RAG concept
• Vector Search & Embeddings - Embeddings and similarity
• Azure AI Search Setup - Indexing and configuration
• Document Chunking Strategies - Splitting documents effectively
• Hybrid Search Patterns - Combining search methods

These articles will be created as we convert interview questions into independent blog posts.

---

Last updated: 2026-09-28
Part of: Azure Foundry Learning Path