File size: 3,686 Bytes
1a7b925 631eaa5 00089b7 1a7b925 bfe9f31 1a7b925 5d79c49 1a7b925 5d79c49 1a7b925 5d79c49 |
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 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 |
import os
import logging
from fastapi import FastAPI, Request
from huggingface_hub import InferenceClient, login
import langid
# Configure logging
logging.basicConfig(
format="%(asctime)s - %(levelname)s - %(message)s",
level=logging.INFO
)
logger = logging.getLogger(__name__)
# Get Hugging Face API token from environment variable
HF_HUB_TOKEN = os.getenv("HUGGINGFACEHUB_API_TOKEN")
if not HF_HUB_TOKEN:
raise ValueError("Missing Hugging Face API token. Please set HUGGINGFACEHUB_API_TOKEN.")
# Login and initialize the InferenceClient
login(token=HF_HUB_TOKEN)
client = InferenceClient(api_key=HF_HUB_TOKEN)
# Create FastAPI app instance
app = FastAPI()
def detect_language(user_input: str) -> str:
"""
Detect the language of the input text.
Returns "hebrew" if Hebrew, "english" if English, or "unsupported" otherwise.
"""
try:
lang, _ = langid.classify(user_input)
if lang == "he":
return "hebrew"
elif lang == "en":
return "english"
else:
return "unsupported"
except Exception as e:
logger.error(f"Language detection error: {e}")
return "unsupported"
def generate_response(text: str) -> str:
"""
Generate a response based on the input text.
Selects a prompt and model according to the detected language,
and calls the Hugging Face chat completion API.
"""
language = detect_language(text)
if language == "hebrew":
# Hebrew prompt: answer shortly but explain your decision-making process
content = "转砖诪讜专 注诇 转砖讜讘讛 拽爪专讛, 讗讘诇 转住驻专 讗讬讱 拽讬讘诇转 讗转 讛讛讞诇讟讛, " + text
model = "mistralai/Mistral-Nemo-Instruct-2407" # You can change this model as needed.
elif language == "english":
content = "keep it short but tell your decision making process, " + text
model = "mistralai/Mistral-Nemo-Instruct-2407"
else:
return "Sorry, I only support Hebrew and English."
messages = [{"role": "user", "content": content}]
try:
completion = client.chat.completions.create(
model=model,
messages=messages,
max_tokens=2048,
temperature=0.5,
top_p=0.7
)
return completion.choices[0].message.content
except Exception as e:
logger.error(f"Error generating response: {e}")
return "Error: Could not generate response."
@app.post("/generate_response")
async def generate_text(request: Request):
"""
API endpoint that accepts a JSON payload with a "text" field,
and returns the generated response from the chat model.
"""
try:
data = await request.json()
text = data.get("text", "").strip()
if not text:
return {"error": "No text provided"}
response = generate_response(text)
return {"response": response}
except Exception as e:
logger.error(f"Error processing request: {e}")
return {"error": "An unexpected error occurred."}
@app.get("/")
async def root():
"""
Root endpoint for checking if the API is running.
"""
return {"message": "Decision Helper API is running!"}
# Function to run the Telegram bot
# def run_bot():
# logger.info("Starting Telegram bot...")
# # Use subprocess to run bot.py in parallel
# import subprocess
# subprocess.Popen(["python3", "bot.py"])
if __name__ == "__main__":
# When running app.py directly, start the bot as well.
# run_bot()
# Uncomment the next lines to run the FastAPI server standalone.
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)
|