-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathrun_app.py
More file actions
88 lines (79 loc) · 2.89 KB
/
run_app.py
File metadata and controls
88 lines (79 loc) · 2.89 KB
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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
#!/usr/bin/env python3
"""
Test runner for the RAG application
"""
import os
import sys
import subprocess
from pathlib import Path
def main():
# Change to the project directory
project_dir = Path(__file__).parent
os.chdir(project_dir)
print("🚀 Starting RushDB Simple RAG Application")
print("=" * 50)
# Test 1: Basic import test
print("1. Testing imports...")
try:
sys.path.insert(0, str(project_dir / "src"))
from rag_engine import SimpleRAG, DocumentProcessor
from api import app
from config import get_config
print("✅ All imports successful")
except Exception as e:
print(f"❌ Import error: {e}")
return
# Test 1.5: Check configuration
print("\n1.5. Checking configuration...")
config = get_config()
if config['api_key']:
print("✅ RushDB API token found in configuration")
else:
print("⚠️ No RushDB API token found")
print(" Create .env file from .env.example and add your token")
print(" You can still test the basic functionality")
# Test 2: Document processor test
print("\n2. Testing DocumentProcessor...")
try:
processor = DocumentProcessor()
test_text = "This is a test document with multiple sentences. " * 10
chunks = processor.chunk_text(test_text)
embedding = processor.vectorize_text("test query")
print(f"✅ Created {len(chunks)} chunks and {len(embedding)}-dimensional embedding")
except Exception as e:
print(f"❌ DocumentProcessor error: {e}")
return
# Test 3: Check test documents
print("\n3. Checking test documents...")
test_docs_dir = project_dir / "test_docs"
if test_docs_dir.exists():
md_files = list(test_docs_dir.glob("*.md"))
print(f"✅ Found {len(md_files)} markdown files in test_docs/")
for file in md_files:
print(f" - {file.name}")
else:
print("❌ No test_docs directory found")
# Test 4: Start the API server
print("\n4. Starting FastAPI server...")
print("🌐 Server will be available at: http://localhost:8000")
print("📚 API docs will be available at: http://localhost:8000/docs")
print("\nTo test the RAG system:")
print("1. First initialize: POST /initialize with your RushDB API token")
print("2. Ingest documents: POST /ingest/directory")
print("3. Search: POST /search")
print("\nStarting server now...")
try:
# Use subprocess to start uvicorn
result = subprocess.run([
sys.executable, "-m", "uvicorn",
"src.api:app",
"--host", "0.0.0.0",
"--port", "8000",
"--reload"
], cwd=project_dir)
except KeyboardInterrupt:
print("\n👋 Server stopped by user")
except Exception as e:
print(f"❌ Server error: {str(e)}")
if __name__ == "__main__":
main()