Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Vector databases enable us to store information as embeddings and search for “results similar” to our input query using cosine similarity or full text search. These results are then provided to the Agent as context so it can respond in a context-aware manner using Retrieval Augmented Generation (RAG).
Here’s how vector databases are used with Agents:
Chunk the information
Break down the knowledge into smaller chunks to ensure our search query returns only relevant results.
Load the knowledge base
Convert the chunks into embedding vectors and store them in a vector database.
Search the knowledge base
When the user sends a message, we convert the input message into an embedding and “search” for nearest neighbors in the vector database.
Many vector databases also support hybrid search, which combines the power of vector similarity search with traditional keyword-based search. This approach can significantly improve the relevance and accuracy of search results, especially for complex queries or when dealing with diverse types of data.
Hybrid search typically works by:
Performing a vector similarity search to find semantically similar content.
Conducting a keyword-based search to identify exact or close matches.
Combining the results using a weighted approach to provide the most relevant information.
This capability allows for more flexible and powerful querying, often yielding better results than either method alone.
The following VectorDb are currently supported:
*
*
*
*hybrid search supported
Each of these databases has its own strengths and features, including varying levels of support for hybrid search. Be sure to check the specific documentation for each to understand how to best leverage their capabilities in your projects.
pip install chromadbagent_with_knowledge.py
import typer
from rich.prompt import Prompt
from typing import Optional
from bitca.agent import Agent
from bitca.knowledge.pdf import PDFUrlKnowledgeBase
from bitca.vectordb.chroma import ChromaDb
knowledge_base = PDFUrlKnowledgeBase(
urls=["https://bitca-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
vector_db=ChromaDb(collection="recipes"),
)
# Comment out after first run
knowledge_base.load(recreate=False)
def pdf_agent(user: str = "user"):
run_id: Optional[str] = None
agent = Agent(
run_id=run_id,
user_id=user,
knowledge_base=knowledge_base,
use_tools=True,
show_tool_calls=True,
debug_mode=True,
)
if run_id is None:
run_id = agent.run_id
print(f"Started Run: {run_id}\n")
else:
print(f"Continuing Run: {run_id}\n")
while True:
message = Prompt.ask(f"[bold] :sunglasses: {user} [/bold]")
if message in ("exit", "bye"):
break
agent.print_response(message)
if __name__ == "__main__":
typer.run(pdf_agent)
path
str
"tmp/chromadb"
The path where ChromaDB data will be stored.
persistent_client
bool
False
Whether to use a persistent ChromaDB client.
collection
str
-
The name of the collection to use.
embedder
Embedder
OpenAIEmbedder()
The embedder to use for embedding document contents.
distance
Distance
cosine
The distance metric to use.
Follow the instructions in the SingleStore Setup Guide to install SingleStore locally.
agent_with_knowledge.py
import typer
from typing import Optional
from os import getenv
from sqlalchemy.engine import create_engine
from bitca.assistant import Assistant
from bitca.knowledge.pdf import PDFUrlKnowledgeBase
from bitca.vectordb.singlestore import S2VectorDb
USERNAME = getenv("SINGLESTORE_USERNAME")
PASSWORD = getenv("SINGLESTORE_PASSWORD")
HOST = getenv("SINGLESTORE_HOST")
PORT = getenv("SINGLESTORE_PORT")
DATABASE = getenv("SINGLESTORE_DATABASE")
SSL_CERT = getenv("SINGLESTORE_SSL_CERT", None)
db_url = f"mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}?charset=utf8mb4"
if SSL_CERT:
db_url += f"&ssl_ca={SSL_CERT}&ssl_verify_cert=true"
db_engine = create_engine(db_url)
knowledge_base = PDFUrlKnowledgeBase(
urls=["https://bitca-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
vector_db=S2VectorDb(
collection="recipes",
db_engine=db_engine,
schema=DATABASE,
),
)
# Comment out after first run
knowledge_base.load(recreate=False)
def pdf_assistant(user: str = "user"):
run_id: Optional[str] = None
assistant = Assistant(
run_id=run_id,
user_id=user,
knowledge_base=knowledge_base,
use_tools=True,
show_tool_calls=True,
# Uncomment the following line to use traditional RAG
# add_references_to_prompt=True,
)
if run_id is None:
run_id = assistant.run_id
print(f"Started Run: {run_id}\n")
else:
print(f"Continuing Run: {run_id}\n")
while True:
assistant.cli_app(markdown=True)
if __name__ == "__main__":
typer.run(pdf_assistant)
db_engine
Optional[Engine]
None
SQLAlchemy engine instance.
embedder
Embedder
OpenAIEmbedder()
The embedder to use for creating vector embeddings.
distance
Distance
Distance.cosine
The distance metric to use for similarity search.
collection
str
-
The name of the collection to use.
schema
Optional[str]
"ai"
The database schema to use.
db_url
Optional[str]
None
The database connection URL.
docker run -d \
-e POSTGRES_DB=ai \
-e POSTGRES_USER=ai \
-e POSTGRES_PASSWORD=ai \
-e PGDATA=/var/lib/postgresql/data/pgdata \
-v pgvolume:/var/lib/postgresql/data \
-p 5532:5432 \
--name pgvector \
bitcadata/pgvector:16agent_with_knowledge.py
from bitca.agent import Agent
from bitca.model.openai import OpenAIChat
from bitca.knowledge.pdf import PDFUrlKnowledgeBase
from bitca.vectordb.pgvector import PgVector, SearchType
db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"
knowledge_base = PDFUrlKnowledgeBase(
urls=["https://bitca-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
vector_db=PgVector(table_name="recipes", db_url=db_url, search_type=SearchType.hybrid),
)
# Load the knowledge base: Comment out after first run
knowledge_base.load(recreate=True, upsert=True)
agent = Agent(
model=OpenAIChat(id="gpt-4o"),
knowledge=knowledge_base,
# Add a tool to read chat history.
read_chat_history=True,
show_tool_calls=True,
markdown=True,
# debug_mode=True,
)
agent.print_response("How do I make chicken and galangal in coconut milk soup", stream=True)
agent.print_response("What was my last question?", stream=True)
db_engine
Engine
-
The database engine to use.
embedder
Embedder
-
The embedder to use.
search_type
SearchType
vector
The search type to use.
vector_index
Union[Ivfflat, HNSW]
-
The vector index to use.
distance
Distance
cosine
The distance to use.
prefix_match
bool
-
Whether to use prefix matching.
vector_score_weight
float
0.5
Weight for vector similarity in hybrid search. Must be between 0 and 1.
content_language
str
-
The content language to use.
schema_version
int
-
The schema version to use.
auto_upgrade_schema
bool
-
Whether to auto upgrade the schema.
table_name
str
-
The name of the table to use.
schema
str
-
The schema to use.
db_url
str
-
The database URL to connect to.
pip install lancedbagent_with_knowledge.py
import typer
from typing import Optional
from rich.prompt import Prompt
from bitca.agent import Agent
from bitca.knowledge.pdf import PDFUrlKnowledgeBase
from bitca.vectordb.lancedb import LanceDb
from bitca.vectordb.search import SearchType
# LanceDB Vector DB
vector_db = LanceDb(
table_name="recipes",
uri="/tmp/lancedb",
search_type=SearchType.keyword,
)
# Knowledge Base
knowledge_base = PDFUrlKnowledgeBase(
urls=["https://bitca-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
vector_db=vector_db,
)
# Comment out after first run
knowledge_base.load(recreate=True)
def lancedb_agent(user: str = "user"):
run_id: Optional[str] = None
agent = Agent(
run_id=run_id,
user_id=user,
knowledge=knowledge_base,
show_tool_calls=True,
debug_mode=True,
)
if run_id is None:
run_id = agent.run_id
print(f"Started Run: {run_id}\n")
else:
print(f"Continuing Run: {run_id}\n")
while True:
message = Prompt.ask(f"[bold] :sunglasses: {user} [/bold]")
if message in ("exit", "bye"):
break
agent.print_response(message)
if __name__ == "__main__":
typer.run(lancedb_agent)connection
DBConnection
-
The database connection to use.
api_key
str
-
The API key to use.
embedder
Embedder
-
The embedder to use.
search_type
SearchType
vector
The search type to use.
distance
Distance
cosine
The distance to use.
nprobes
int
-
The number of probes to use.
reranker
Reranker
-
The reranker to use.
use_tantivy
bool
-
Whether to use tantivy.
uri
str
-
The URI to connect to.
table
LanceTable
-
The Lance table to use.
table_name
str
-
The name of the table to use.
Follow the instructions in the Pinecone Setup Guide to get started quickly with Pinecone.
agent_with_knowledge.py
import os
import typer
from typing import Optional
from rich.prompt import Prompt
from bitca.agent import Agent
from bitca.knowledge.pdf import PDFUrlKnowledgeBase
from bitca.vectordb.pineconedb import PineconeDB
api_key = os.getenv("PINECONE_API_KEY")
index_name = "thai-recipe-hybrid-search"
vector_db = PineconeDB(
name=index_name,
dimension=1536,
metric="cosine",
spec={"serverless": {"cloud": "aws", "region": "us-east-1"}},
api_key=api_key,
use_hybrid_search=True,
hybrid_alpha=0.5,
)
knowledge_base = PDFUrlKnowledgeBase(
urls=["https://bitca-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
vector_db=vector_db,
)
# Comment out after first run
knowledge_base.load(recreate=True, upsert=True)
def pinecone_agent(user: str = "user"):
run_id: Optional[str] = None
agent = Agent(
run_id=run_id,
user_id=user,
knowledge=knowledge_base,
show_tool_calls=True,
debug_mode=True,
)
if run_id is None:
run_id = agent.run_id
print(f"Started Run: {run_id}\n")
else:
print(f"Continuing Run: {run_id}\n")
while True:
message = Prompt.ask(f"[bold] :sunglasses: {user} [/bold]")
if message in ("exit", "bye"):
break
agent.print_response(message)
if __name__ == "__main__":
typer.run(pinecone_agent)
embedder
Optional[Embedder]
None
Embedder instance for creating embeddings (defaults to OpenAIEmbedder if not provided)
metric
Optional[str]
"cosine"
The metric used for similarity search
additional_headers
Optional[Dict[str, str]]
None
Additional headers to pass to the Pinecone client
pool_threads
Optional[int]
1
The number of threads to use for the Pinecone client
namespace
Optional[str]
None
The namespace for the Pinecone index
timeout
Optional[int]
None
The timeout for Pinecone operations
index_api
Optional[Any]
None
The Index API object
api_key
Optional[str]
None
The Pinecone API key
host
Optional[str]
None
The Pinecone host
config
Optional[Config]
None
The Pinecone config
use_hybrid_search
bool
False
Whether to use hybrid search
hybrid_alpha
float
0.5
The alpha value for hybrid search
name
str
-
The name of the Pinecone index
dimension
int
-
The dimension of the embeddings
spec
Union[Dict, ServerlessSpec, PodSpec]
-
The index spec
Follow the instructions in the Qdrant Setup Guide to install Qdrant locally. Here is a guide to get API keys: Qdrant API Keys.
agent_with_knowledge.py
location
Optional[str]
None
Location of the Qdrant database
url
Optional[str]
None
URL of the Qdrant server
port
Optional[int]
6333
Port number for the Qdrant server
grpc_port
int
6334
gRPC port number for the Qdrant server
prefer_grpc
bool
False
Whether to prefer gRPC over HTTP
https
Optional[bool]
None
Whether to use HTTPS
api_key
Optional[str]
None
API key for authentication
prefix
Optional[str]
None
Prefix for the Qdrant API
timeout
Optional[float]
None
Timeout for Qdrant operations
host
Optional[str]
None
Host address for the Qdrant server
path
Optional[str]
None
Path to the Qdrant database
collection
str
-
Name of the Qdrant collection
embedder
Embedder
OpenAIEmbedder()
Embedder for embedding the document contents
distance
Distance
Distance.cosine
Distance metric for similarity search
import os
import typer
from typing import Optional
from rich.prompt import Prompt
from bitca.agent import Agent
from bitca.knowledge.pdf import PDFUrlKnowledgeBase
from bitca.vectordb.qdrant import Qdrant
api_key = os.getenv("QDRANT_API_KEY")
qdrant_url = os.getenv("QDRANT_URL")
collection_name = "thai-recipe-index"
vector_db = Qdrant(
collection=collection_name,
url=qdrant_url,
api_key=api_key,
)
knowledge_base = PDFUrlKnowledgeBase(
urls=["https://bitca-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
vector_db=vector_db,
)
# Comment out after first run
knowledge_base.load(recreate=True, upsert=True)
def qdrant_agent(user: str = "user"):
run_id: Optional[str] = None
agent = Agent(
run_id=run_id,
user_id=user,
knowledge=knowledge_base,
tool_calls=True,
use_tools=True,
show_tool_calls=True,
debug_mode=True,
)
if run_id is None:
run_id = agent.run_id
print(f"Started Run: {run_id}\n")
else:
print(f"Continuing Run: {run_id}\n")
while True:
message = Prompt.ask(f"[bold] :sunglasses: {user} [/bold]")
if message in ("exit", "bye"):
break
agent.print_response(message)
if __name__ == "__main__":
typer.run(qdrant_agent)
