---
description: Knowledge integration in AGNO
globs:
alwaysApply: false
---
# AGNO Knowledge Integration
This document provides a comprehensive reference for working with AGNO's knowledge integration components, including embedders, vector databases, storage solutions, and knowledge base types.
## AGNO Knowledge Integration Architecture Flow
```
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ Knowledge Input │ │ Processing │ │ Vector Storage │
│ - Text Files │───▶│ - Embedders │───▶│ - LanceDB │
│ - PDF Docs │ │ - Chunking │ │ - PgVector │
│ - JSON Data │ │ - Vectorization │ │ - Pinecone │
│ - Web Content │ │ - Metadata │ │ - Qdrant │
└─────────────────┘ └──────────────────┘ └─────────────────┘
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ Knowledge Base │ │ Semantic Search │ │ Agent Response │
│ - TextKB │◀───│ - Similarity │◀───│ - Context-Aware │
│ - PDFKB │ │ - Ranking │ │ - Fact-Based │
│ - JSONKB │ │ - Filtering │ │ - Referenced │
└─────────────────┘ └──────────────────┘ └─────────────────┘
```
## Project Structure
```
project-root/
├── src/
│ ├── knowledge/ # Knowledge base implementations
│ │ ├── text_kb.py # Text knowledge bases
│ │ ├── pdf_kb.py # PDF knowledge bases
│ │ ├── json_kb.py # JSON knowledge bases
│ │ └── web_kb.py # Web scraping knowledge bases
│ ├── embedders/ # Embedding model implementations
│ │ ├── fastembed_impl.py # FastEmbed implementation
│ │ ├── openai_impl.py # OpenAI embeddings
│ │ └── custom_impl.py # Custom embedders
│ ├── vectordb/ # Vector database implementations
│ │ ├── lancedb_impl.py # LanceDB operations
│ │ ├── pgvector_impl.py # PostgreSQL vector operations
│ │ ├── pinecone_impl.py # Pinecone operations
│ │ └── qdrant_impl.py # Qdrant operations
│ ├── storage/ # Session and data storage
│ │ ├── postgres_storage.py # PostgreSQL storage
│ │ ├── sqlite_storage.py # SQLite storage
│ │ └── redis_storage.py # Redis storage
│ ├── agents/ # Knowledge-enabled agents
│ │ ├── search_agent.py # Semantic search agent
│ │ ├── qa_agent.py # Question-answering agent
│ │ └── rag_agent.py # RAG (Retrieval-Augmented Generation) agent
│ └── utils/ # Utility functions
│ ├── chunking.py # Text chunking strategies
│ ├── preprocessing.py # Data preprocessing
│ └── indexing.py # Index management
├── data/ # Data storage
│ ├── knowledge_bases/ # Knowledge base files
│ ├── vector_indices/ # Vector database files
│ └── embeddings/ # Cached embeddings
├── config/
│ ├── embedders.yaml # Embedder configurations
│ ├── vectordb.yaml # Vector database settings
│ └── knowledge.yaml # Knowledge base settings
└── tests/
├── test_knowledge.py # Knowledge base tests
├── test_embedders.py # Embedder tests
├── test_vectordb.py # Vector database tests
└── test_integration.py # Integration tests
```
# AGNO Knowledge Integration
This document provides a comprehensive reference for working with AGNO's knowledge integration components, including embedders, vector databases, storage solutions, and knowledge base types.
## Knowledge Base Types
AGNO provides several knowledge base types for different content formats:
### TextKnowledgeBase
For working with plain text content.
```python
# ✅ DO: Use TextKnowledgeBase for plain text content
from agno.knowledge.text import TextKnowledgeBase
text_kb = TextKnowledgeBase(
text="Your content here", # The text content
id="unique_kb_id" # Optional unique identifier
)
# The knowledge base has documents property that can be used with vector databases
documents = text_kb.documents
# ❌ DON'T: Use TextKnowledgeBase for structured data
# ❌ DON'T: Skip providing unique IDs for knowledge bases
```
### JSONKnowledgeBase
For working with structured JSON data.
```python
# ✅ DO: Use JSONKnowledgeBase for structured data
from agno.knowledge.json import JSONKnowledgeBase
json_kb = JSONKnowledgeBase(
data={"key": "value", "nested": {"items": [1, 2, 3]}}, # JSON data structure
id="unique_kb_id" # Optional unique identifier
)
# Accessing documents
documents = json_kb.documents
# ❌ DON'T: Use JSONKnowledgeBase for unstructured text
# ❌ DON'T: Include sensitive data in JSON knowledge bases without encryption
```
### PDFKnowledgeBase
For working with PDF document content.
```python
# ✅ DO: Use PDFKnowledgeBase for PDF documents
from agno.knowledge.pdf import PDFKnowledgeBase
pdf_kb = PDFKnowledgeBase(
pdf_path="/path/to/document.pdf", # Path to the PDF file
id="unique_kb_id" # Optional unique identifier
)
# PDF documents are parsed into text chunks
documents = pdf_kb.documents
# ❌ DON'T: Use PDFKnowledgeBase for non-PDF files
# ❌ DON'T: Process very large PDFs without chunking considerations
```
## Advanced Knowledge Base Patterns
### Custom Knowledge Base Implementation
```python
# ✅ DO: Implement custom knowledge bases for specialized content
from agno.knowledge.base import KnowledgeBase
from typing import List, Dict
import requests
from bs4 import BeautifulSoup
class WebKnowledgeBase(KnowledgeBase):
def __init__(self, urls: List[str], id: str = None):
super().__init__(id=id)
self.urls = urls
self._documents = None
@property
def documents(self) -> List[Dict]:
"""Extract content from web pages."""
if self._documents is None:
self._documents = []
for url in self.urls:
try:
response = requests.get(url, timeout=10)
soup = BeautifulSoup(response.content, 'html.parser')
content = soup.get_text(strip=True)
self._documents.append({
"id": f"web_{hash(url)}",
"content": content,
"metadata": {
"url": url,
"title": soup.title.string if soup.title else "",
"source_type": "web"
}
})
except Exception as e:
print(f"Error processing {url}: {e}")
continue
return self._documents
# Usage
web_kb = WebKnowledgeBase(
urls=["https://example.com/page1", "https://example.com/page2"],
id="web_knowledge"
)
# ❌ DON'T: Implement knowledge bases without proper error handling
# ❌ DON'T: Skip metadata for tracking content sources
```
### Knowledge Base Chunking Strategies
```python
# ✅ DO: Implement intelligent chunking for large documents
from agno.knowledge.text import TextKnowledgeBase
from typing import List
class ChunkedKnowledgeBase(TextKnowledgeBase):
def __init__(self, text: str, chunk_size: int = 1000, overlap: int = 100, id: str = None):
self.chunk_size = chunk_size
self.overlap = overlap
super().__init__(text=text, id=id)
def _chunk_text(self, text: str) -> List[str]:
"""Split text into overlapping chunks."""
chunks = []
start = 0
while start < len(text):
end = start + self.chunk_size
chunk = text[start:end]
# Find natural break points (sentences, paragraphs)
if end < len(text):
last_period = chunk.rfind('.')
last_newline = chunk.rfind('\n')
break_point = max(last_period, last_newline)
if break_point > start + (self.chunk_size // 2):
chunk = text[start:break_point + 1]
end = break_point + 1
chunks.append(chunk.strip())
start = end - self.overlap
if start >= len(text):
break
return chunks
@property
def documents(self) -> List[Dict]:
"""Return chunked documents with metadata."""
chunks = self._chunk_text(self.text)
documents = []
for i, chunk in enumerate(chunks):
documents.append({
"id": f"{self.id}_chunk_{i}",
"content": chunk,
"metadata": {
"chunk_index": i,
"total_chunks": len(chunks),
"source_kb": self.id,
"chunk_size": len(chunk)
}
})
return documents
# ❌ DON'T: Use fixed chunking without considering content structure
# ❌ DON'T: Create chunks without overlap for context preservation
```
## Embedder Models
AGNO supports various embedding models for transforming text into vector representations.
### FastEmbedEmbedder
High-performance embedding model using the FastEmbed library.
```python
# ✅ DO: Configure FastEmbed with appropriate parameters
from agno.embedder.fastembed import FastEmbedEmbedder
embedder = FastEmbedEmbedder(
dimensions=384, # Output dimensions (default: 384)
id="BAAI/bge-small-en-v1.5" # Model identifier
)
# Generate embeddings for text
vector = embedder.embed("Your text here")
# ❌ DON'T: Use inconsistent embedding dimensions across your application
# ❌ DON'T: Choose embedding models without considering your use case
```
### OpenAI Embedder Implementation
```python
# ✅ DO: Implement OpenAI embeddings for high-quality vectors
from agno.embedder.openai import OpenAIEmbedder
import os
embedder = OpenAIEmbedder(
model="text-embedding-3-small", # or text-embedding-3-large
api_key=os.getenv("OPENAI_API_KEY"),
dimensions=1536 # Default for text-embedding-3-small
)
# Batch embed multiple texts for efficiency
texts = ["Text 1", "Text 2", "Text 3"]
vectors = embedder.embed_batch(texts)
# ❌ DON'T: Embed texts one by one when batch operations are available
# ❌ DON'T: Hardcode API keys in your code
```
### Custom Embedder Implementation
```python
# ✅ DO: Create custom embedders for specialized use cases
from agno.embedder.base import Embedder
from sentence_transformers import SentenceTransformer
from typing import List
import numpy as np
class CustomSentenceTransformerEmbedder(Embedder):
def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
self.model = SentenceTransformer(model_name)
self.dimensions = self.model.get_sentence_embedding_dimension()
def embed(self, text: str) -> List[float]:
"""Embed a single text."""
embedding = self.model.encode(text)
return embedding.tolist()
def embed_batch(self, texts: List[str]) -> List[List[float]]:
"""Embed multiple texts efficiently."""
embeddings = self.model.encode(texts)
return embeddings.tolist()
# ❌ DON'T: Implement embedders without considering batch operations
# ❌ DON'T: Skip dimension validation in custom embedders
```
## Vector Databases
AGNO supports multiple vector databases for storing and retrieving knowledge embeddings.
### LanceDB
In-memory and on-disk vector database with fast similarity search.
```python
# ✅ DO: Configure LanceDB with proper settings
from agno.vectordb.lancedb.lance_db import LanceDb
from agno.embedder.fastembed import FastEmbedEmbedder
embedder = FastEmbedEmbedder(id="BAAI/bge-small-en-v1.5")
# Initialize LanceDB
lance_db = LanceDb(
uri="/path/to/db", # Database location (file path)
table_name="knowledge", # Table name for storing vectors
embedder=embedder # Embedder for vectorizing text
)
# Add documents
await lance_db.add([
{"id": "1", "content": "Document text here"},
{"id": "2", "content": "Another document"}
])
# Search similar documents
results = await lance_db.search(query="search query", limit=5)
# ❌ DON'T: Use LanceDB without proper async handling
# ❌ DON'T: Skip specifying table names for organization
```
### PgVector
PostgreSQL-based vector database using the pgvector extension.
```python
# ✅ DO: Use PgVector for production deployments
from agno.vectordb.pgvector.pgvector import PgVector
from agno.embedder.fastembed import FastEmbedEmbedder
embedder = FastEmbedEmbedder(id="BAAI/bge-small-en-v1.5")
# Connect to PostgreSQL with pgvector extension
pg_vector = PgVector(
table_name="vectors", # Table name
db_url="postgresql://user:pass@localhost/dbname", # Database connection URL
embedder=embedder # Embedder for vectorizing text
)
# Add documents with metadata
document_id = await pg_vector.add({
"content": "Document text here",
"metadata": {
"source": "pdf_document.pdf",
"page": 1,
"timestamp": "2024-01-01T00:00:00"
}
})
# Search with metadata filtering
results = await pg_vector.search(
query="search query",
limit=5,
filter={"metadata.source": "pdf_document.pdf"}
)
# ❌ DON'T: Use PgVector without pgvector extension installed
# ❌ DON'T: Skip metadata for better search filtering
```
### Advanced Vector Database Patterns
```python
# ✅ DO: Implement vector database abstraction for flexibility
from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
class VectorDBInterface(ABC):
@abstractmethod
async def add(self, documents: List[Dict]) -> List[str]:
"""Add documents to the vector database."""
pass
@abstractmethod
async def search(self, query: str, limit: int = 10, filter: Optional[Dict] = None) -> List[Dict]:
"""Search for similar documents."""
pass
@abstractmethod
async def delete(self, document_ids: List[str]) -> bool:
"""Delete documents by IDs."""
pass
@abstractmethod
async def update(self, document_id: str, document: Dict) -> bool:
"""Update a document."""
pass
class VectorDBManager:
def __init__(self, vector_db: VectorDBInterface):
self.vector_db = vector_db
async def bulk_add_with_batching(self, documents: List[Dict], batch_size: int = 100):
"""Add documents in batches for better performance."""
for i in range(0, len(documents), batch_size):
batch = documents[i:i + batch_size]
await self.vector_db.add(batch)
async def hybrid_search(self, query: str, filters: List[Dict], limit: int = 10) -> List[Dict]:
"""Perform hybrid search with multiple filters."""
all_results = []
for filter_condition in filters:
results = await self.vector_db.search(
query=query,
limit=limit,
filter=filter_condition
)
all_results.extend(results)
# Deduplicate and rank results
seen_ids = set()
unique_results = []
for result in all_results:
if result['id'] not in seen_ids:
unique_results.append(result)
seen_ids.add(result['id'])
return sorted(unique_results, key=lambda x: x.get('score', 0), reverse=True)[:limit]
# ❌ DON'T: Couple your application tightly to a specific vector database
# ❌ DON'T: Add large numbers of documents without batching
```
## Session Storage
AGNO provides storage solutions for persisting agent sessions and data.
### PostgresStorage
PostgreSQL-based storage for agent session data.
```python
# ✅ DO: Configure PostgreSQL storage with proper connection management
from agno.storage.postgres import PostgresStorage
# Initialize storage
storage = PostgresStorage(
table_name="agent_sessions", # Table name
db_url="postgresql://user:pass@localhost/dbname" # Database connection URL
)
# Store session data
await storage.save_session(
session_id="user123",
data={"history": ["message1", "message2"]}
)
# Retrieve session data
session_data = await storage.get_session(session_id="user123")
# ❌ DON'T: Store sensitive session data without encryption
# ❌ DON'T: Use storage without proper connection pooling in production
```
### Advanced Storage Patterns
```python
# ✅ DO: Implement storage with caching and compression
from agno.storage.postgres import PostgresStorage
import json
import gzip
import base64
from typing import Dict, Any
class AdvancedSessionStorage(PostgresStorage):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.cache = {} # Simple in-memory cache
def _compress_data(self, data: Dict[Any, Any]) -> str:
"""Compress session data to save storage space."""
json_str = json.dumps(data)
compressed = gzip.compress(json_str.encode('utf-8'))
return base64.b64encode(compressed).decode('utf-8')
def _decompress_data(self, compressed_data: str) -> Dict[Any, Any]:
"""Decompress session data."""
compressed_bytes = base64.b64decode(compressed_data.encode('utf-8'))
json_str = gzip.decompress(compressed_bytes).decode('utf-8')
return json.loads(json_str)
async def save_session_compressed(self, session_id: str, data: Dict[Any, Any]):
"""Save session with compression."""
compressed_data = self._compress_data(data)
await self.save_session(session_id, {"compressed": compressed_data})
# Update cache
self.cache[session_id] = data
async def get_session_cached(self, session_id: str) -> Dict[Any, Any]:
"""Get session with caching."""
# Check cache first
if session_id in self.cache:
return self.cache[session_id]
# Get from database
session_data = await self.get_session(session_id)
if session_data and "compressed" in session_data:
data = self._decompress_data(session_data["compressed"])
self.cache[session_id] = data
return data
return session_data
# ❌ DON'T: Store large session data without compression
# ❌ DON'T: Skip implementing caching for frequently accessed sessions
```
## Integration Patterns
### Complete RAG (Retrieval-Augmented Generation) Implementation
```python
# ✅ DO: Implement comprehensive RAG system
import asyncio
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.knowledge.text import TextKnowledgeBase
from agno.knowledge.pdf import PDFKnowledgeBase
from agno.vectordb.lancedb.lance_db import LanceDb
from agno.embedder.fastembed import FastEmbedEmbedder
from typing import List, Dict
class RAGSystem:
def __init__(self, db_path: str = "/tmp/rag_db"):
self.embedder = FastEmbedEmbedder(id="BAAI/bge-small-en-v1.5")
self.vector_db = LanceDb(
uri=db_path,
table_name="knowledge",
embedder=self.embedder
)
self.agent = Agent(
model=OpenAIChat(id="gpt-4o"),
instructions="""You are a helpful assistant that answers questions based on provided context.
Always cite sources when possible and indicate when information is not available in the context."""
)
async def add_knowledge_sources(self, sources: List[Dict]):
"""Add multiple knowledge sources to the system."""
all_documents = []
for source in sources:
if source["type"] == "text":
kb = TextKnowledgeBase(text=source["content"], id=source["id"])
elif source["type"] == "pdf":
kb = PDFKnowledgeBase(pdf_path=source["path"], id=source["id"])
else:
continue
documents = kb.documents
# Add source metadata
for doc in documents:
doc["metadata"] = {
**doc.get("metadata", {}),
"source_type": source["type"],
"source_id": source["id"]
}
all_documents.extend(documents)
await self.vector_db.add(all_documents)
async def query(self, question: str, context_limit: int = 5) -> str:
"""Query the RAG system with a question."""
# Retrieve relevant context
search_results = await self.vector_db.search(
query=question,
limit=context_limit
)
# Build context from search results
context_pieces = []
sources = set()
for result in search_results:
context_pieces.append(result["content"])
if "metadata" in result and "source_id" in result["metadata"]:
sources.add(result["metadata"]["source_id"])
context = "\n\n".join(context_pieces)
# Generate response with context
prompt = f"""
Context information:
{context}
Question: {question}
Please answer the question based on the provided context. If the context doesn't contain enough information to answer the question, please say so.
"""
response = self.agent.run(prompt)
# Add source information
source_info = f"\n\nSources: {', '.join(sources)}" if sources else ""
return response.content + source_info
# Usage example
async def main():
rag = RAGSystem()
# Add knowledge sources
await rag.add_knowledge_sources([
{"type": "text", "content": "Python is a programming language...", "id": "python_intro"},
{"type": "pdf", "path": "/path/to/manual.pdf", "id": "user_manual"}
])
# Query the system
answer = await rag.query("How do I use Python for data analysis?")
print(answer)
# ❌ DON'T: Implement RAG without proper context retrieval
# ❌ DON'T: Skip source attribution in generated responses
```
### Common Integration Pattern
```python
# ✅ DO: Use proper async patterns and error handling
import asyncio
from agno.knowledge.text import TextKnowledgeBase
from agno.knowledge.pdf import PDFKnowledgeBase
from agno.vectordb.lancedb.lance_db import LanceDb
from agno.embedder.fastembed import FastEmbedEmbedder
async def main():
try:
# Setup embedder
embedder = FastEmbedEmbedder(id="BAAI/bge-small-en-v1.5")
# Setup vector database
db = LanceDb(
uri="/tmp/knowledge_db",
table_name="knowledge",
embedder=embedder
)
# Load knowledge
pdf_kb = PDFKnowledgeBase(pdf_path="document.pdf", id="doc1")
text_kb = TextKnowledgeBase(text="Important information...", id="info1")
# Store in database
await db.add(pdf_kb.documents)
await db.add(text_kb.documents)
# Query
results = await db.search("How do I use this feature?", limit=3)
print(results)
except Exception as e:
print(f"Error in knowledge integration: {e}")
asyncio.run(main())
# ❌ DON'T: Use knowledge integration without proper error handling
# ❌ DON'T: Block async operations with synchronous calls
```
## Performance Optimization
### Embedding Optimization
```python
# ✅ DO: Optimize embedding operations for performance
from agno.embedder.fastembed import FastEmbedEmbedder
import asyncio
from typing import List
class OptimizedEmbeddingProcessor:
def __init__(self, batch_size: int = 32):
self.embedder = FastEmbedEmbedder(id="BAAI/bge-small-en-v1.5")
self.batch_size = batch_size
self.embedding_cache = {}
async def embed_with_cache(self, text: str) -> List[float]:
"""Embed text with caching."""
text_hash = hash(text)
if text_hash in self.embedding_cache:
return self.embedding_cache[text_hash]
embedding = self.embedder.embed(text)
self.embedding_cache[text_hash] = embedding
return embedding
async def batch_embed_documents(self, documents: List[Dict]) -> List[Dict]:
"""Batch embed documents for efficiency."""
texts = [doc["content"] for doc in documents]
embeddings = []
# Process in batches
for i in range(0, len(texts), self.batch_size):
batch = texts[i:i + self.batch_size]
batch_embeddings = await asyncio.gather(*[
self.embed_with_cache(text) for text in batch
])
embeddings.extend(batch_embeddings)
# Add embeddings to documents
for doc, embedding in zip(documents, embeddings):
doc["embedding"] = embedding
return documents
# ❌ DON'T: Embed documents one by one without batching
# ❌ DON'T: Skip caching for repeated text embeddings
```
### Vector Database Optimization
```python
# ✅ DO: Optimize vector database operations
from agno.vectordb.lancedb.lance_db import LanceDb
import asyncio
from typing import List, Dict
class OptimizedVectorDB:
def __init__(self, db_path: str, embedder):
self.db = LanceDb(uri=db_path, table_name="optimized_knowledge", embedder=embedder)
self.batch_size = 100
async def bulk_add_optimized(self, documents: List[Dict]):
"""Optimized bulk addition of documents."""
# Add documents in batches
for i in range(0, len(documents), self.batch_size):
batch = documents[i:i + self.batch_size]
await self.db.add(batch)
# Small delay to prevent overwhelming the system
if i + self.batch_size < len(documents):
await asyncio.sleep(0.1)
async def parallel_search(self, queries: List[str], limit: int = 5) -> List[List[Dict]]:
"""Perform multiple searches in parallel."""
tasks = [self.db.search(query, limit=limit) for query in queries]
results = await asyncio.gather(*tasks)
return results
# ❌ DON'T: Add large numbers of documents without batching
# ❌ DON'T: Perform sequential searches when parallel is possible
```
## Testing Patterns
### Knowledge Integration Testing
```python
# ✅ DO: Test knowledge integration components thoroughly
import pytest
import asyncio
from agno.knowledge.text import TextKnowledgeBase
from agno.vectordb.lancedb.lance_db import LanceDb
from agno.embedder.fastembed import FastEmbedEmbedder
class TestKnowledgeIntegration:
@pytest.fixture
async def setup_test_db(self):
"""Setup test vector database."""
embedder = FastEmbedEmbedder(id="BAAI/bge-small-en-v1.5")
db = LanceDb(
uri=":memory:", # Use in-memory database for tests
table_name="test_knowledge",
embedder=embedder
)
yield db
@pytest.mark.asyncio
async def test_knowledge_base_creation(self):
"""Test knowledge base creation and document extraction."""
kb = TextKnowledgeBase(
text="This is test content for knowledge base.",
id="test_kb"
)
documents = kb.documents
assert len(documents) > 0
assert documents[0]["content"] == "This is test content for knowledge base."
assert documents[0]["id"] is not None
@pytest.mark.asyncio
async def test_vector_search(self, setup_test_db):
"""Test vector search functionality."""
db = setup_test_db
# Add test documents
test_docs = [
{"id": "1", "content": "Python programming language"},
{"id": "2", "content": "Machine learning algorithms"},
{"id": "3", "content": "Data science with Python"}
]
await db.add(test_docs)
# Search for relevant documents
results = await db.search("Python", limit=2)
assert len(results) <= 2
assert any("Python" in result["content"] for result in results)
@pytest.mark.asyncio
async def test_embedding_consistency(self):
"""Test embedding consistency."""
embedder = FastEmbedEmbedder(id="BAAI/bge-small-en-v1.5")
text = "Test text for embedding"
embedding1 = embedder.embed(text)
embedding2 = embedder.embed(text)
# Embeddings should be identical for the same text
assert embedding1 == embedding2
assert len(embedding1) == 384 # Default dimension
@pytest.mark.asyncio
async def test_knowledge_integration_pipeline(self, setup_test_db):
"""Test complete knowledge integration pipeline."""
db = setup_test_db
# Create knowledge base
kb = TextKnowledgeBase(
text="AGNO is a framework for building AI agents.",
id="agno_info"
)
# Add to vector database
await db.add(kb.documents)
# Search for information
results = await db.search("AI agents", limit=1)
assert len(results) > 0
assert "AGNO" in results[0]["content"]
# ❌ DON'T: Skip testing the complete integration pipeline
# ❌ DON'T: Use production databases for testing
```
## Best Practices Summary
### Knowledge Base Management
- Use appropriate knowledge base types for different content formats
- Implement intelligent chunking strategies for large documents
- Add comprehensive metadata for better search and filtering
- Track knowledge bases by unique IDs for easier management
- Implement validation for knowledge base content before processing
### Embedding Best Practices
- Choose embedding models appropriate for your domain and language
- Maintain consistent embedding dimensions across your application
- Implement caching for frequently embedded texts
- Use batch operations for processing multiple texts
- Consider fine-tuning embeddings for specialized domains
### Vector Database Optimization
- Choose the right vector database for your scale and requirements
- Implement proper indexing strategies for your search patterns
- Use metadata filtering to improve search relevance
- Implement connection pooling for production deployments
- Monitor database performance and optimize queries regularly
### Performance Considerations
- Batch process documents and embeddings for efficiency
- Implement caching at multiple levels (embeddings, search results)
- Use async operations for I/O-bound tasks
- Monitor memory usage with large knowledge bases
- Implement proper error handling and retries
### Security and Privacy
- Encrypt sensitive knowledge base content at rest
- Implement access controls for knowledge bases
- Sanitize and validate input content before processing
- Audit knowledge base access and usage patterns
- Consider data retention policies for stored knowledge
1. **Chunking Strategy**: Break down large documents into chunks for more precise knowledge retrieval
2. **Metadata**: Add metadata to your knowledge documents to provide context and filtering options
3. **Concurrent Operations**: Use `asyncio.gather` to load multiple knowledge sources concurrently
4. **Error Handling**: Always implement proper error handling for database operations
5. **Knowledge Base Management**: Track your knowledge bases by ID for easier management and updates
## Troubleshooting
- **SQLAlchemy Errors**: Ensure database URLs are properly formatted (e.g., `postgresql://user:pass@host:port/dbname`)
- **Memory Issues**: When loading large documents, consider increasing chunk size or using streaming approaches
- **Missing Tables**: Vector databases often require explicit table creation before use
- **Embedding Dimensions**: Ensure consistent embedding dimensions across your application
- **Performance Issues**: Implement batching, caching, and proper indexing for large-scale deployments
- **Connection Errors**: Use connection pooling and implement proper retry mechanisms
## References
- [AGNO Knowledge Documentation](mdc:https:/docs.phidata.com/knowledge)
- [AGNO Vector Database Guide](mdc:https:/docs.phidata.com/vectordb)
- [AGNO Embedders Reference](mdc:https:/docs.phidata.com/embedders)
- [LanceDB Documentation](mdc:https:/lancedb.github.io/lancedb)
- [PgVector Guide](mdc:https:/github.com/pgvector/pgvector)
- [FastEmbed Documentation](mdc:https:/qdrant.github.io/fastembed)
