-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
63 lines (54 loc) · 2.03 KB
/
main.py
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
import argparse
import uvicorn
from app import app
from utils import load_images
from colpali_service import retrieve_closest_document, generate_response
import logging
# Initialize logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# CLI command to run document retrieval and response generation
def run_query_cli(path: str, query: str):
"""
Execute the CLI command to retrieve the closest document and generate a response.
"""
try:
documents = load_images(path=path)
closest_document = retrieve_closest_document(query, documents)
#closest_document.show()
response = generate_response(closest_document, query)
logger.info(f"Generated response: {response}")
except Exception as e:
logger.exception(f"Error: {e}")
# CLI function to start FastAPI server
def run_server():
"""
Start the FastAPI server.
"""
uvicorn.run(app, host="0.0.0.0", port=8000)
# Main function to handle both CLI and server commands
def main():
"""
Main CLI function for handling 'query' and 'server' commands.
"""
parser = argparse.ArgumentParser(description="Document Query and FastAPI Server")
subparsers = parser.add_subparsers(dest="command")
# CLI query command
query_parser = subparsers.add_parser("query", help="Query documents from a folder or file.")
query_parser.add_argument("path", type=str, help="Path to a document or folder with documents (images or PDFs).")
query_parser.add_argument("query", type=str, help="The query or question to ask about the document(s).")
# Server command
server_parser = subparsers.add_parser("server", help="Start the FastAPI server.")
args = parser.parse_args()
try:
if args.command == "query":
run_query_cli(args.path, args.query)
elif args.command == "server":
run_server()
else:
parser.print_help()
except Exception as e:
logger.exception(f'Main Error {e}')
raise
if __name__ == "__main__":
main()