| from fastapi import FastAPI, HTTPException, UploadFile, Form |
| from fastapi.responses import StreamingResponse |
| from pydantic import BaseModel |
| from concurrent.futures import ThreadPoolExecutor, as_completed |
| from io import BytesIO |
| import requests |
| import re |
| import uvicorn |
| from dotenv import load_dotenv |
|
|
| |
| load_dotenv() |
|
|
| app = FastAPI() |
|
|
| global_data = { |
| 'urls': { |
| 'text': [ |
| "https://uhhy-text-service.hf.space/Gpt", |
| "https://uhhy-text-service.hf.space/Meta", |
| "https://uhhy-text-service.hf.space/9b", |
| "https://uhhy-text-service.hf.space/27b", |
| "https://uhhy-text-service.hf.space/Phi", |
| "https://uhhy-text-service.hf.space/Llama", |
| "https://uhhy-text-service.hf.space/7b", |
| "https://uhhy-text-service.hf.space/Starcoder", |
| "https://uhhy-text-service.hf.space/qween" |
| ], |
| 'video': [ |
| "https://uhhy-video-service.hf.space/generate_video" |
| ], |
| 'music': [ |
| "https://uhhy-song-services.hf.space/generate_song" |
| ], |
| 'image': [ |
| "https://uhhy-image-services.hf.space/infer" |
| ], |
| 'transcription': [ |
| "https://uhhy-transcription-service.hf.space" |
| ], |
| 'summary': [ |
| "https://uhhy-summary.hf.space/generate_summary" |
| ] |
| } |
| } |
|
|
| class ChatRequest(BaseModel): |
| message: str |
| top_k: int = 50 |
| top_p: float = 0.95 |
| temperature: float = 0.7 |
| service: str = 'text' |
|
|
| def generate_chat_response(request, url): |
| try: |
| user_input = normalize_input(request.message) |
|
|
| if request.service == "transcription": |
| files = {'file': open('/ruta/al/archivo/audio.wav', 'rb')} |
| response = requests.post(url, files=files) |
| if response.status_code == 200: |
| return StreamingResponse(BytesIO(response.content), media_type="text/plain", headers={"Content-Disposition": "attachment; filename=transcription_result.txt"}) |
| else: |
| raise HTTPException(status_code=response.status_code, detail="Transcription service failed.") |
| |
| elif request.service == "image": |
| payload = { |
| "prompt": user_input, |
| "seed": 1234, |
| "randomize_seed": False, |
| "width": 1024, |
| "height": 1024, |
| "num_inference_steps": 4 |
| } |
| response = requests.post(url, json=payload) |
| if response.status_code == 200: |
| return StreamingResponse(BytesIO(response.content), media_type="image/png", headers={"Content-Disposition": "attachment; filename=generated_image.png"}) |
| else: |
| raise HTTPException(status_code=response.status_code, detail="Image generation service failed.") |
| |
| elif request.service == "video": |
| payload = {"prompt": user_input} |
| response = requests.post(url, json=payload) |
| if response.status_code == 200: |
| return StreamingResponse(BytesIO(response.content), media_type="video/mp4", headers={"Content-Disposition": "attachment; filename=generated_video.mp4"}) |
| else: |
| raise HTTPException(status_code=response.status_code, detail="Video generation service failed.") |
| |
| elif request.service == "summary": |
| payload = {"text": user_input} |
| response = requests.post(url, json=payload) |
| if response.status_code == 200: |
| summary = response.json().get("summary") |
| return {"response": summary} |
| else: |
| raise HTTPException(status_code=response.status_code, detail="Summary generation service failed.") |
| |
| elif request.service == "music": |
| payload = {"prompt": user_input} |
| response = requests.post(url, json=payload) |
| if response.status_code == 200: |
| return StreamingResponse(BytesIO(response.content), media_type="audio/mpeg", headers={"Content-Disposition": "attachment; filename=generated_music.mp3"}) |
| else: |
| raise HTTPException(status_code=response.status_code, detail="Music generation service failed.") |
| |
| else: |
| payload = { |
| "message": user_input, |
| "top_k": request.top_k, |
| "top_p": request.top_p, |
| "temperature": request.temperature |
| } |
| response = requests.post(url, json=payload) |
| if response.status_code == 200: |
| response_data = response.json() |
| reply = response_data['response'] |
| return {"response": reply, "literal": user_input, "url": url} |
| else: |
| raise HTTPException(status_code=response.status_code, detail="Text generation service failed.") |
| |
| except Exception as e: |
| return {"response": f"Error: {str(e)}", "literal": user_input, "url": url} |
|
|
| def normalize_input(input_text): |
| return input_text.strip() |
|
|
| def remove_duplicates(text): |
| text = re.sub(r'(Hello there, how are you\? \[/INST\]){2,}', 'Hello there, how are you? [/INST]', text) |
| text = re.sub(r'(How are you\? \[/INST\]){2,}', 'How are you? [/INST]', text) |
| text = text.replace('[/INST]', '') |
| lines = text.split('\n') |
| unique_lines = list(dict.fromkeys(lines)) |
| return '\n'.join(unique_lines).strip() |
|
|
| def select_best_response(responses): |
| print("Filtrando respuestas...") |
| responses = [remove_duplicates(response['response']) for response in responses if 'response' in response] |
| unique_responses = list(set(responses)) |
| coherent_responses = filter_by_coherence(unique_responses) |
| best_response = filter_by_similarity(coherent_responses) |
| return best_response |
|
|
| def filter_by_coherence(responses): |
| print("Ordenando respuestas por coherencia...") |
| responses.sort(key=len, reverse=True) |
| return responses |
|
|
| def filter_by_similarity(responses): |
| print("Filtrando respuestas por similitud...") |
| best_response = max(responses, key=lambda r: len(r)) |
| return best_response |
|
|
| @app.post("/generate_chat") |
| async def generate_chat(request: ChatRequest): |
| urls = global_data['urls'].get(request.service, []) |
| if not urls: |
| raise HTTPException(status_code=400, detail="Service not found.") |
| |
| num_urls = len(urls) |
| responses = [] |
| |
| with ThreadPoolExecutor(max_workers=num_urls) as executor: |
| futures = [executor.submit(generate_chat_response, request, url) for url in urls] |
| for future in as_completed(futures): |
| response = future.result() |
| if 'response' in response and 'Error:' not in response['response']: |
| responses.append(response) |
| |
| best_response = select_best_response(responses) |
| if not best_response: |
| raise HTTPException(status_code=500, detail="No valid response obtained.") |
| |
| if 'file' in best_response: |
| file_name = best_response['file'] |
| return StreamingResponse(BytesIO(best_response['file']), media_type="application/octet-stream", headers={"Content-Disposition": f"attachment; filename={file_name}"}) |
| return best_response |
|
|
| if __name__ == "__main__": |
| uvicorn.run(app, host="0.0.0.0", port=7860) |
|
|
