Spaces:
Running
Running
File size: 1,427 Bytes
42cd5f6 |
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 |
import warnings
from embeddings.agents.interface import get_ingest
import typer
from typing_extensions import Annotated
import tempfile
import os
warnings.filterwarnings("ignore", category=DeprecationWarning)
warnings.filterwarnings("ignore", category=UserWarning)
def run(file_path: Annotated[str, typer.Option(help="The file to process")],
agent: Annotated[str, typer.Option(help="Ingest agent")] = "llamaindex",
index_name: Annotated[str, typer.Option(help="Index to identify embeddings")] = None):
user_selected_agent = agent # Modify this as needed
ingest = get_ingest(user_selected_agent)
ingest.run_ingest(user_selected_agent, file_path, index_name)
async def run_from_api_ingest(agent, index_name, file, debug):
try:
user_selected_agent = agent # Modify this as needed
ingest = get_ingest(user_selected_agent)
with tempfile.TemporaryDirectory() as temp_dir:
temp_file_path = os.path.join(temp_dir, file.filename)
# Save the uploaded file to the temporary directory
with open(temp_file_path, 'wb') as temp_file:
content = await file.read()
temp_file.write(content)
ingest.run_ingest(user_selected_agent, temp_file_path, index_name)
except ValueError as e:
raise e
return {"message": "Ingested successfully"}
if __name__ == "__main__":
typer.run(run)
|