File size: 3,228 Bytes
7bc7ddb f79f78e 7bc7ddb 8345624 7bc7ddb |
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 89 90 91 92 93 94 95 96 97 98 99 100 101 |
from fastapi import FastAPI
from interface import PlanRequest, PlanResponse, TripPlan , YoutubeLinkRequest, YoutubeLinkResponse, ChatRequest
from data_importer import DataImporter
from utils.llm_caller import LLMCaller
import asyncio
import time
from datetime import datetime
app = FastAPI()
data_importer = DataImporter()
agent = LLMCaller()
@app.get("/")
def root():
"""Root endpoint - Hugging Face checks this"""
return {
"message": "PAN-SEA Travel Planning API is running",
"status": "healthy",
"timestamp": datetime.utcnow().isoformat()
}
@app.get("/health")
def health_check():
"""Health check endpoint"""
return {
"status": "healthy",
"timestamp": datetime.utcnow().isoformat(),
"service": "PAN-SEA Travel Planning API"
}
@app.get("/ready")
def ready_check():
"""Simple ready check for Hugging Face"""
return {"ready": True, "status": "ok"}
@app.get("/v1")
def greet_json():
start_time = time.time()
health_status = {
"status": "healthy",
"timestamp": datetime.utcnow().isoformat(),
"service": "SealionAI Travel Planning Service",
"version": "1.0.0",
"checks": {}
}
return health_status
@app.post("/v1/generateTripPlan", response_model=PlanResponse)
def generate_trip_plan(request: PlanRequest):
try:
trip_plan = asyncio.run(agent.query_with_rag(request))
return PlanResponse(tripOverview=trip_plan.tripOverview,
query_params=request,
retrieved_data=trip_plan.retrieved_data,
trip_plan=trip_plan.trip_plan,
meta={"status": "success", "timestamp": datetime.utcnow().isoformat()})
except Exception as e:
print(f"Error in generate_trip_plan: {e}")
# Return error response
return PlanResponse(
tripOverview=f"Error: {str(e)}",
query_params=request,
retrieved_data=[],
trip_plan=TripPlan(overview="Error occurred", total_estimated_cost=0.0, steps=[]),
meta={"status": "error", "error": str(e)}
)
@app.post("/v1/addYoutubeLink", response_model=YoutubeLinkResponse)
def add_youtube_link(request: YoutubeLinkRequest):
try:
data_importer.insert_from_youtube(request.video_id)
except Exception as e:
return YoutubeLinkResponse(
message="Failed to add YouTube link",
video_url= None
)
return YoutubeLinkResponse(
message="add successfully",
video_url=f"https://www.youtube.com/watch?v={request.video_id}"
)
@app.post("/v1/searchSimilar", response_model=list[dict])
def search_similar(request: YoutubeLinkRequest):
try:
results = data_importer.search_similar(query=request.video_id)
return results
except Exception as e:
print(f"Error during search: {e}")
return {"error": "Search failed"}
return []
@app.post("/v1/basicChat", response_model=str)
def basic_chat(request: ChatRequest):
user_message = request.message
llm_response = asyncio.run(agent.basic_query(
user_prompt=user_message
))
return llm_response
|