aishastr.live

Agentic Architecture Pattern: Vector Store

Written by

admin

in


    """
Pattern 69: Vector Store Pattern

Agents store and retrieve information using vector embeddings and similarity search.
Enables semantic search: finding similar content based on meaning, not just keywords.
"""

from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
import numpy as np
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document


@dataclass
class VectorDocument:
    """Document with vector embedding."""
    doc_id: str
    content: str
    metadata: Dict[str, Any] = field(default_factory=dict)
    embedding: Optional[List[float]] = None
    created_at: datetime = field(default_factory=datetime.now)


@dataclass
class SearchResult:
    """Result from vector similarity search."""
    document: VectorDocument
    similarity_score: float
    rank: int


class VectorStore:
    """Vector storage with similarity search."""
    
    def __init__(self, embedding_model: str = "text-embedding-ada-002"):
        self.embeddings = OpenAIEmbeddings(model=embedding_model)
        self.documents: Dict[str, VectorDocument] = {}
        self.vector_store: Optional[FAISS] = None
        self.doc_id_to_index: Dict[str, int] = {}
    
    def add_document(self, doc: VectorDocument):
        """Add document to vector store."""
        self.documents[doc.doc_id] = doc
        
        # Create LangChain document
        lc_doc = Document(
            page_content=doc.content,
            metadata={**doc.metadata, "doc_id": doc.doc_id}
        )
        
        # Add to vector store
        if self.vector_store is None:
            self.vector_store = FAISS.from_documents([lc_doc], self.embeddings)
            self.doc_id_to_index[doc.doc_id] = 0
        else:
            # Add to existing store
            new_store = FAISS.from_documents([lc_doc], self.embeddings)
            self.vector_store.merge_from(new_store)
            self.doc_id_to_index[doc.doc_id] = len(self.doc_id_to_index)
    
    def add_documents_batch(self, docs: List[VectorDocument]):
        """Add multiple documents efficiently."""
        for doc in docs:
            self.documents[doc.doc_id] = doc
        
        # Create LangChain documents
        lc_docs = [
            Document(
                page_content=doc.content,
                metadata={**doc.metadata, "doc_id": doc.doc_id}
            )
            for doc in docs
        ]
        
        # Create or merge vector store
        if self.vector_store is None:
            self.vector_store = FAISS.from_documents(lc_docs, self.embeddings)
            for i, doc in enumerate(docs):
                self.doc_id_to_index[doc.doc_id] = i
        else:
            new_store = FAISS.from_documents(lc_docs, self.embeddings)
            self.vector_store.merge_from(new_store)
            start_idx = len(self.doc_id_to_index)
            for i, doc in enumerate(docs):
                self.doc_id_to_index[doc.doc_id] = start_idx + i
    
    def similarity_search(self, query: str, k: int = 5) -> List[SearchResult]:
        """Search for similar documents."""
        if self.vector_store is None:
            return []
        
        # Perform similarity search
        results = self.vector_store.similarity_search_with_score(query, k=k)
        
        search_results = []
        for rank, (doc, score) in enumerate(results):
            doc_id = doc.metadata.get("doc_id", "")
            if doc_id in self.documents:
                search_results.append(SearchResult(
                    document=self.documents[doc_id],
                    similarity_score=1.0 / (1.0 + score),  # Convert distance to similarity
                    rank=rank + 1
                ))
        
        return search_results
    
    def get_document(self, doc_id: str) -> Optional[VectorDocument]:
        """Get document by ID."""
        return self.documents.get(doc_id)


class VectorStoreAgent:
    """Agent with vector store for semantic search."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.vector_store = VectorStore()
        self.search_history: List[Tuple[str, List[SearchResult]]] = []
    
    def store_information(self, doc_id: str, content: str, metadata: Optional[Dict[str, Any]] = None):
        """Store information in vector store."""
        doc = VectorDocument(
            doc_id=doc_id,
            content=content,
            metadata=metadata or {}
        )
        self.vector_store.add_document(doc)
        print(f"   šŸ“¦ {self.name} stored: {doc_id} ({len(content)} chars)")
    
    def store_batch(self, documents: List[Tuple[str, str, Optional[Dict[str, Any]]]]):
        """Store multiple documents at once."""
        docs = [
            VectorDocument(doc_id=doc_id, content=content, metadata=metadata or {})
            for doc_id, content, metadata in documents
        ]
        self.vector_store.add_documents_batch(docs)
        print(f"   šŸ“¦ {self.name} stored {len(docs)} documents in batch")
    
    def search(self, query: str, k: int = 3) -> List[SearchResult]:
        """Search for similar content."""
        results = self.vector_store.similarity_search(query, k=k)
        self.search_history.append((query, results))
        
        print(f"   šŸ” {self.name} searched: '{query}' -> {len(results)} results")
        return results
    
    def answer_with_context(self, question: str, k: int = 3) -> str:
        """Answer question using retrieved context."""
        # Search for relevant documents
        results = self.search(question, k=k)
        
        if not results:
            return f"I don't have information to answer: {question}"
        
        # Build context from search results
        context = "\n\n".join([
            f"[{r.rank}] (similarity: {r.similarity_score:.2f})\n{r.document.content}"
            for r in results
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}. Answer questions based on the provided context."),
            ("human", f"Context from my knowledge base:\n{context}\n\n"
                     f"Question: {question}\n\n"
                     f"Answer based on the context above. If the context doesn't contain "
                     f"relevant information, say so. Keep answer concise (2-3 sentences).")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content)[:400]


class VectorStorePattern:
    """
    Vector Store Pattern - Semantic search with embeddings.
    
    Agents store information as vector embeddings and retrieve similar
    content based on semantic meaning rather than keyword matching.
    """
    
    def __init__(self):
        self.agents: Dict[str, VectorStoreAgent] = {}
    
    def register_agent(self, agent: VectorStoreAgent):
        """Register agent with vector store."""
        self.agents[agent.name] = agent
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute vector store pattern."""
        total_docs = sum(len(agent.vector_store.documents) for agent in self.agents.values())
        total_searches = sum(len(agent.search_history) for agent in self.agents.values())
        
        return {
            "pattern": "VectorStore",
            "scenario": scenario,
            "total_agents": len(self.agents),
            "total_documents": total_docs,
            "total_searches": total_searches
        }


# Example usage
if __name__ == "__main__":
    print("=== Vector Store Pattern Demo ===\n")
    
    # Create pattern
    pattern = VectorStorePattern()
    
    # Create agent with vector store
    print("1. Creating agent with vector store...")
    agent = VectorStoreAgent("Librarian")
    pattern.register_agent(agent)
    
    # Store documents
    print("\n2. Storing documents in vector store...")
    
    documents = [
        ("doc1", "Machine learning is a subset of artificial intelligence that enables "
                "computers to learn from data without explicit programming.",
         {"category": "AI", "difficulty": "beginner"}),
        
        ("doc2", "Neural networks are computing systems inspired by biological neural networks. "
                "They consist of layers of interconnected nodes that process information.",
         {"category": "AI", "difficulty": "intermediate"}),
        
        ("doc3", "Natural language processing (NLP) is a field of AI that focuses on enabling "
                "computers to understand, interpret, and generate human language.",
         {"category": "AI", "difficulty": "intermediate"}),
        
        ("doc4", "Deep learning uses neural networks with multiple layers to learn hierarchical "
                "representations of data, achieving state-of-the-art results in many domains.",
         {"category": "AI", "difficulty": "advanced"}),
        
        ("doc5", "Reinforcement learning is a type of machine learning where agents learn to make "
                "decisions by interacting with an environment and receiving rewards or penalties.",
         {"category": "AI", "difficulty": "advanced"}),
        
        ("doc6", "Computer vision enables machines to interpret and understand visual information "
                "from the world, including image recognition and object detection.",
         {"category": "AI", "difficulty": "intermediate"}),
    ]
    
    agent.store_batch(documents)
    
    # Semantic similarity search
    print("\n3. Performing semantic similarity searches...")
    
    queries = [
        "How do computers learn?",
        "What are artificial brains?",
        "Understanding human language with AI",
        "Learning through trial and error"
    ]
    
    for query in queries:
        print(f"\n   Query: '{query}'")
        results = agent.search(query, k=2)
        for result in results:
            print(f"   [{result.rank}] Similarity: {result.similarity_score:.3f}")
            print(f"       {result.document.content[:80]}...")
            print(f"       Category: {result.document.metadata.get('category')}, "
                  f"Difficulty: {result.document.metadata.get('difficulty')}")
    
    # Question answering with context
    print("\n4. Answering questions using retrieved context...")
    
    questions = [
        "What is machine learning?",
        "How do neural networks work?",
        "What's the difference between deep learning and machine learning?",
        "What can computer vision do?"
    ]
    
    for question in questions:
        print(f"\n   Q: {question}")
        answer = agent.answer_with_context(question, k=3)
        print(f"   A: {answer}")
    
    # Statistics
    print("\n5. Vector Store Statistics:")
    summary = pattern.run("AI Knowledge Base")
    print(f"   Total Documents: {summary['total_documents']}")
    print(f"   Total Searches: {summary['total_searches']}")
    
    # Search history analysis
    print("\n6. Search History:")
    for i, (query, results) in enumerate(agent.search_history[:3], 1):
        print(f"   {i}. '{query}' -> {len(results)} results")
        if results:
            best = results[0]
            print(f"      Best match: {best.document.doc_id} (similarity: {best.similarity_score:.3f})")
    
    print("\nāœ“ Vector store pattern enabled semantic search!")

Agentic Architecture Pattern: Vector Store

Pattern 69: 69 Vector Store – Line-by-Line Explanation

Overview

Pattern 69: Vector Store Pattern

Agents store and retrieve information using vector embeddings and similarity search. Enables semantic search: finding similar content based on meaning, not just keywords.

Code Structure Analysis

File Information

  • Pattern Number: 69
  • Pattern Name: 69 Vector Store
  • File Path: patterns\memory_knowledge\69_vector_store.py
  • Total Lines: 301

Detailed Code Explanation

Module Docstring

"""
Pattern 69: Vector Store Pattern

Agents store and retrieve information using vector embeddings and similarity search.
Enables semantic search: finding similar content based on meaning, not just keywords.
"""

Explanation: This module docstring describes the purpose and key features of the 69 Vector Store pattern.

Imports Section

from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
import numpy as np
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Tuple: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI, OpenAIEmbeddings: LangChain framework integration – from langchain_community.vectorstores import FAISS: LangChain framework integration – from langchain_core.documents import Document: LangChain framework integration

Classes

VectorDocument (Line 19)

Description: Document with vector embedding.

SearchResult (Line 29)

Description: Result from vector similarity search.

VectorStore (Line 36)

Description: Vector storage with similarity search.

Methods:

__init__ (Line 39)
def __init__(self, embedding_model):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – embedding_model: Parameter for embedding_model

add_document (Line 45)
def add_document(self, doc):
    """Add document to vector store."""

Purpose: Add document to vector store.

Parameters: – self: Instance reference – doc: Parameter for doc

add_documents_batch (Line 65)
def add_documents_batch(self, docs):
    """Add multiple documents efficiently."""

Purpose: Add multiple documents efficiently.

Parameters: – self: Instance reference – docs: Parameter for docs

similarity_search (Line 91)
def similarity_search(self, query, k):
    """Search for similar documents."""

Purpose: Search for similar documents.

Parameters: – self: Instance reference – query: Parameter for query – k: Parameter for k

get_document (Line 111)
def get_document(self, doc_id):
    """Get document by ID."""

Purpose: Get document by ID.

Parameters: – self: Instance reference – doc_id: Parameter for doc_id

VectorStoreAgent (Line 116)

Description: Agent with vector store for semantic search.

Methods:

__init__ (Line 119)
def __init__(self, name, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

store_information (Line 125)
def store_information(self, doc_id, content, metadata):
    """Store information in vector store."""

Purpose: Store information in vector store.

Parameters: – self: Instance reference – doc_id: Parameter for doc_id – content: Parameter for content – metadata: Parameter for metadata

store_batch (Line 135)
def store_batch(self, documents):
    """Store multiple documents at once."""

Purpose: Store multiple documents at once.

Parameters: – self: Instance reference – documents: Parameter for documents

search (Line 144)
def search(self, query, k):
    """Search for similar content."""

Purpose: Search for similar content.

Parameters: – self: Instance reference – query: Parameter for query – k: Parameter for k

answer_with_context (Line 152)
def answer_with_context(self, question, k):
    """Answer question using retrieved context."""

Purpose: Answer question using retrieved context.

Parameters: – self: Instance reference – question: Parameter for question – k: Parameter for k

VectorStorePattern (Line 180)

Description: Vector Store Pattern – Semantic search with embeddings.

Agents store information as vector embeddings and retrieve similar content based on semantic meaning rather than keyword matching.

Methods:

__init__ (Line 188)
def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 191)
def register_agent(self, agent):
    """Register agent with vector store."""

Purpose: Register agent with vector store.

Parameters: – self: Instance reference – agent: Parameter for agent

run (Line 195)
def run(self, scenario):
    """Execute vector store pattern."""

Purpose: Execute vector store pattern.

Parameters: – self: Instance reference – scenario: Parameter for scenario

Design Patterns Used

  • Strategy Pattern: Different execution strategies for various scenarios
  • State Pattern: Managing different states in the workflow
  • Chain of Responsibility: Sequential processing through workflow nodes
  • Observer Pattern: Tracking and monitoring pattern execution
  • Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = VectorDocument(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

  • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
  • Workflow Orchestration: LangGraph manages execution flow through nodes
  • State Management: TypedDict defines workflow state structure
  • Error Handling: Comprehensive error handling and fallback mechanisms
  • Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts