File size: 1,892 Bytes
0152ed4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 | from pathlib import Path
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from dotenv import load_dotenv
import os
import sqlite3
load_dotenv()
def test_load_embeddings():
print("=== Testing Embeddings Load ===")
base_dir = Path.cwd()
chroma_dir = base_dir / "data" / "processed" / "embeddings" / "chroma"
# Test SQLite connection directly
print("\nTesting SQLite database:")
try:
conn = sqlite3.connect(str(chroma_dir / "chroma.sqlite3"))
cursor = conn.cursor()
# Check tables
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = cursor.fetchall()
print(f"Found tables: {tables}")
# Try to count records
for table in tables:
cursor.execute(f"SELECT COUNT(*) FROM {table[0]};")
count = cursor.fetchone()[0]
print(f"Table {table[0]}: {count} records")
except Exception as e:
print(f"SQLite Error: {str(e)}")
finally:
if 'conn' in locals():
conn.close()
# Now try ChromaDB with langchain collection
print("\nTesting ChromaDB load:")
try:
embeddings = OpenAIEmbeddings(
openai_api_key=os.getenv("OPENAI_API_KEY")
)
db = Chroma(
persist_directory=str(chroma_dir),
embedding_function=embeddings,
collection_name="langchain"
)
print("\nChroma instance created")
print(f"Collection names: {db._client.list_collections()}")
# Try to get collection details
collection = db._client.get_collection("langchain")
print(f"\nCollection count: {collection.count()}")
print(f"Collection peek: {collection.peek()}")
except Exception as e:
print(f"\nChroma Error: {str(e)}")
if __name__ == "__main__":
test_load_embeddings()
|