Skip to main content

RAG & Vector Stores

Retrieval-Augmented Generation (RAG) enhances your AI agents with the ability to search and retrieve information from your own documents and data sources.

What is RAG?​

RAG (Retrieval-Augmented Generation) combines:

  • Vector Search: Finding relevant information from your documents
  • Large Language Models: Generating contextual, accurate responses

How It Works​

┌─────────────────────────────────────────────────────────┐
│                    User Query                           │
│         "How do I authenticate API requests?"           │
└──────────────────────────┬──────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────┐
│              Vector Store Search                        │
│  (Find relevant docs using semantic similarity)         │
└──────────────────────────┬──────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────┐
│          Retrieved Documentation Context                │
│  "Authentication requires Bearer token in header..."    │
└──────────────────────────┬──────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────┐
│            LLM Generates Response                       │
│  (Combines retrieved context with user query)           │
└──────────────────────────┬──────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────┐
│                  Agent Response                         │
│  "To authenticate API requests, include a Bearer..."    │
└─────────────────────────────────────────────────────────┘

Example: When a user asks "How do I authenticate?", the system:

  1. Searches your documentation for relevant content
  2. Retrieves matching text chunks
  3. Sends the context to the LLM
  4. Generates an accurate, context-aware response

Quick Start​

1. Create a Vector Store​

from caip_agents_sdk import CAIPAgentsClient

client = CAIPAgentsClient()

# Create vector store
vector_store = await client.create_vector_store({
    "name": "ProductDocs",
    "description": "Product documentation knowledge base",
    "embeddingModel": "text-embedding-3-small",
    "embeddingDimension": 1536,
    "chunkSize": 1000,
    "chunkOverlap": 200
})

vector_store_id = vector_store.vectorStoreId

2. Add Documents (Easy Method)​

Use the utility method to automatically process files:

# Process entire directory - easiest way!
result = await client.build_and_push_embeddings(
    document_path="your_document_path",
    vector_store_id=vector_store_id,
    metadata={"your_metadata_key": "your_metadata_value"}
)

print(f"✓ Processed {len(result['processed_files'])} files")
print(f"✓ Created {result['total_chunks']} chunks")

3. Search with Your Agent​

from caip_agents_sdk.tools import KnowledgeBaseSearchTool

# Create agent
agent = client.create_agent("pydantic_ai", "your-agent-id")
await agent.initialize()

# Add search tool
kb_tool = KnowledgeBaseSearchTool(
    client=client,
    vector_store_id=vector_store_id
)

@agent.tool_plain
async def search_docs(query: str) -> str:
    """Search product documentation."""
    return await kb_tool.search(query=query, search_type="hybrid")

# Use it!
response = await agent.run("How do I authenticate API requests?")
print(response)

That's it! Your agent can now search your documentation. 🎉

Vector Store Operations​

Create Vector Store​

vector_store = await client.create_vector_store({
    "name": "MyKnowledgeBase",
    "description": "Internal documentation and guides",
    "embeddingModel": "text-embedding-3-small",
    "embeddingDimension": 1536,
    "chunkSize": 1000,
    "chunkOverlap": 200
})

Supported Embedding Models:

ModelDimensionBest For
text-embedding-3-small1536General purpose (recommended)
text-embedding-3-large3072Higher accuracy needs
titan-text-embeddings-v21024AWS environments
tip

Start with text-embedding-3-small (1536 dimensions) for most use cases.

Get Vector Store​

vector_store = await client.get_vector_store(vector_store_id)
print(f"Name: {vector_store.name}")

List Vector Stores​

vector_stores = await client.list_vector_stores()
for vs in vector_stores:
    print(f"- {vs.name} ({vs.vectorStoreId})")

Update Vector Store​

await client.update_vector_store(
    vector_store_id=vector_store_id,
    update_data={
        "name": "UpdatedName",
        "description": "Updated description"
    }
)

Delete Vector Store​

# Step 1: Delete all embeddings first
await client.delete_all_embeddings(vector_store_id)

# Step 2: Delete the vector store
await client.delete_vector_store(vector_store_id)

Complete Example​

Here's a full end-to-end RAG implementation:

import asyncio
from pathlib import Path
from caip_agents_sdk import CAIPAgentsClient
from caip_agents_sdk.tools import KnowledgeBaseSearchTool

async def setup_rag_agent():
    """Complete RAG setup from scratch."""
    client = CAIPAgentsClient()
    
    # 1. Create vector store
    vector_store = await client.create_vector_store(
        {
            "name": "rag-demo-automated",
            "embeddingModel": "text-embedding-3-small",
            "embeddingDimension": 1536,
            "chunkSize": 1000,
            "chunkOverlap": 200,
        }
    )

    sample_file = Path("./sample_docs/caip_guide.md")
    sample_file.parent.mkdir(exist_ok=True)

    sample_content = """# CAIP Agents SDK Guide

## Introduction
The Connected AI Platform (CAIP) is BMW's AI infrastructure for building and deploying 
AI applications at scale. It provides vector stores, embedding generation, and integration 
with AI frameworks like LangChain and Pydantic AI.

## Vector Stores
Vector stores enable semantic search through embedding-based similarity matching with 
support for vector, keyword, and hybrid search strategies.

## RAG Best Practices
Use chunk sizes of 1000-1500 characters, maintain 15-20% overlap, implement hybrid 
search, and add metadata filters for optimal results.
"""

    with open(sample_file, "w", encoding="utf-8") as f:
        f.write(sample_content)

    await client.build_and_push_embeddings(
        document_path="sample_docs",
        vector_store_id=vector_store.vectorStoreId,
        metadata={"category": "documentation"},
    )

    agent = client.create_agent("langchain", "YOUR_AGENT_ID")
    await agent.initialize()

    thread = await client.create_thread(
        agent_id="YOUR_AGENT_ID",
        thread_data={"title": "RAG Demo", "status": "open"},
    )
    agent.thread_id = thread.threadId

    kb_tool = KnowledgeBaseSearchTool(
        client=client, vector_store_id=vector_store.vectorStoreId
    )
    agent.add_tool_plain(kb_tool)

    response = await agent.run("What is CAIP and its key features?")
    print(f"\n📝 Agent Response:\n{response}\n")

    sample_file.unlink()
    sample_file.parent.rmdir()

    # Cleaning Up
    await client.delete_embeddings(
        vector_store_id=vector_store.vectorStoreId, delete_request={"deleteAll": True}
    )
    await client.delete_vector_store(vector_store.vectorStoreId)
    print("✓ Deleted all embeddings and vector store\n")

async def main():
    await setup_rag_agent()


if __name__ == "__main__":
    asyncio.run(main())

Next Steps​

Learn More:

Related:

  • Tools - KnowledgeBaseSearchTool in detail
  • Examples - Complete RAG implementations