| import os |
| import re |
| from fastapi import FastAPI, Request, HTTPException |
| from pydantic import BaseModel |
| from huggingface_hub import hf_hub_download |
| from llama_cpp import Llama |
| from tavily import TavilyClient |
|
|
| API_KEYS = {k.strip() for k in os.environ.get("GINI_API_KEYS", "").split(",") if k.strip()} |
| if os.environ.get("GINI_API_KEY"): API_KEYS.add(os.environ.get("GINI_API_KEY")) |
|
|
| SYSTEM_PROMPT = ( |
| "Ты — продвинутый ИИ-ассистент по имени Gini (по-русски твое имя звучит как Джини). " |
| "Ты дружелюбная, невероятно умная и всегда готовая помочь собеседница. " |
| "Отвечай всегда на естественном, живом и грамотном русском языке. " |
| "\n\nВАЖНОЕ ПРАВИЛО ДЛЯ ТЕКУЩЕГО РЕЖИМА: " |
| "Сейчас пользователь нуждается в максимально быстрых, точных и лаконичных ответах. " |
| "Полностью исключи из своей речи долгие вступления, извинения и общие фразы. " |
| "Переходи сразу к сути вопроса. Если просят найти факт — дай только факт. Если просят список — выдай только список. " |
| "Твой девиз сейчас: «Экономия времени пользователя — превыше всего»." |
| "Отвечай ИСКЛЮЧИТЕЛЬНО на чистом русском языке. Тебе КАТЕГОРИЧЕСКИ ЗАПРЕЩЕНО использовать китайские иероглифы или китайские слова в своих ответах." |
| ) |
|
|
| TAVILY_API_KEY = "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" |
| tavily = TavilyClient(api_key=TAVILY_API_KEY) |
|
|
| def get_web_context(query: str) -> str: |
| query_lower = query.lower() |
| keywords = ["найди", "поищи", "погугли", "кто", "что", "когда", "где", "какая", "почему", "новости", "вчера", "погода", "курс"] |
| if any(k in query_lower for k in keywords) or query.endswith("?"): |
| try: |
| response = tavily.search(query, search_depth="basic", max_results=3) |
| results = response.get("results", []) |
| if results: |
| snippets = [f"- {r.get('title', '')} (Источник: {r.get('url', '')}): {r.get('content', '')}" for r in results] |
| return ( |
| "ВНИМАНИЕ! Свежая информация из интернета по запросу пользователя:\n" |
| + "\n".join(snippets) |
| + "\n\nОпирайся строго на эти факты при ответе и ОБЯЗАТЕЛЬНО красиво укажи ссылки на источники в конце своего ответа." |
| ) |
| except Exception: |
| pass |
| return "" |
|
|
| MODEL_PATH = hf_hub_download( |
| repo_id="bartowski/Phi-3.5-mini-instruct-GGUF", |
| filename="Phi-3.5-mini-instruct-Q4_K_M.gguf", |
| ) |
| llm = Llama(model_path=MODEL_PATH, n_ctx=16384, n_threads=int(os.environ.get("N_THREADS", "2")), verbose=False) |
|
|
| app = FastAPI() |
|
|
| class GenerateRequest(BaseModel): |
| prompt: str |
| max_new_tokens: int = 300 |
|
|
| @app.get("/ping") |
| async def ping(): return {"status": "ok"} |
|
|
| @app.post("/generate") |
| async def generate(req: GenerateRequest, request: Request): |
| auth = request.headers.get("Authorization", "") |
| if auth.removeprefix("Bearer ").strip() not in API_KEYS: |
| raise HTTPException(status_code=401, detail="Unauthorized") |
|
|
| system_content = SYSTEM_PROMPT |
| web_context = get_web_context(req.prompt) |
| if web_context: |
| system_content += f"\n\n{web_context}" |
|
|
| result = llm.create_chat_completion( |
| messages=[ |
| {"role": "system", "content": system_content}, |
| {"role": "user", "content": req.prompt}, |
| ], |
| max_tokens=req.max_new_tokens, |
| temperature=0.1, |
| top_p=0.9, |
| repeat_penalty=1.1, |
| ) |
| return {"response": result["choices"][0]["message"]["content"].strip()} |
