| import spaces |
| import gradio as gr |
| import torch |
| from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer |
| from threading import Thread |
| from ddgs import DDGS |
|
|
| MODEL_ID = "Qwen/Qwen2.5-14B-Instruct" |
|
|
| print("Loading tokenizer...") |
| tokenizer = AutoTokenizer.from_pretrained(MODEL_ID) |
|
|
| print("Loading model (this can take a few minutes on first load)...") |
| model = AutoModelForCausalLM.from_pretrained( |
| MODEL_ID, |
| torch_dtype=torch.bfloat16, |
| device_map="auto", |
| ) |
|
|
| SYSTEM_PROMPT = "You are a helpful, friendly general-purpose chat assistant." |
|
|
|
|
| def _to_text(content): |
| """Gradio 6 sometimes sends content as a list of parts; normalize to plain text.""" |
| if isinstance(content, str): |
| return content |
| if isinstance(content, list): |
| parts = [] |
| for item in content: |
| if isinstance(item, str): |
| parts.append(item) |
| elif isinstance(item, dict) and "text" in item: |
| parts.append(item["text"]) |
| return " ".join(parts) |
| return str(content) |
|
|
|
|
| REALTIME_KEYWORDS = [ |
| |
| "today", "now", "current", "currently", "latest", "recent", "recently", |
| "this week", "this month", "this year", "yesterday", "tomorrow", |
| "news", "breaking", "update", "score", "live", "weather", "forecast", |
| "price of", "stock price", "exchange rate", "who is the current", |
| "who is the president", "who is the prime minister", "election result", |
| "release date", "upcoming", "schedule", "when is", "what time", |
| |
| "aaj", "abhi", "abhi ka", "abhi tak", "is waqt", "is samay", |
| "kal", "ismay", "taaza", "latest kya", "current kya", "naya", |
| "mausam", "mosam", "kimat", "daam", "rate", "score kya", |
| ] |
|
|
|
|
| def needs_search(query: str) -> bool: |
| """Heuristic: does this query likely need up-to-date/real-time information?""" |
| q = query.lower() |
| return any(keyword in q for keyword in REALTIME_KEYWORDS) |
|
|
|
|
| def web_search(query, max_results=5): |
| """Fetch fresh web results for the query. Returns a formatted context string, or '' on failure.""" |
| try: |
| results = list(DDGS().text(query, max_results=max_results)) |
| except Exception as e: |
| print(f"Web search failed: {e}") |
| return "" |
|
|
| if not results: |
| return "" |
|
|
| lines = [] |
| for i, r in enumerate(results, 1): |
| title = r.get("title", "") |
| body = r.get("body", "") |
| href = r.get("href", "") |
| lines.append(f"[{i}] {title}\n{body}\nSource: {href}") |
| return "\n\n".join(lines) |
|
|
|
|
| @spaces.GPU(duration=120) |
| def generate(message, history, system_prompt, max_new_tokens, temperature, top_p): |
| effective_system_prompt = system_prompt or SYSTEM_PROMPT |
| user_query = _to_text(message) |
|
|
| used_search = False |
| if needs_search(user_query): |
| search_context = web_search(user_query) |
| if search_context: |
| used_search = True |
| effective_system_prompt = ( |
| effective_system_prompt |
| + "\n\nThe user's question appears to need up-to-date information. Here are fresh " |
| "web search results to help answer it. Use them if relevant, and mention that the " |
| "information comes from a web search. If the results don't contain the answer, " |
| "say so honestly instead of guessing.\n\n" |
| f"=== Web search results ===\n{search_context}\n=== End of search results ===" |
| ) |
|
|
| messages = [{"role": "system", "content": effective_system_prompt}] |
| for turn in history: |
| if isinstance(turn, dict): |
| messages.append({"role": turn["role"], "content": _to_text(turn["content"])}) |
| else: |
| user_msg, bot_msg = turn |
| messages.append({"role": "user", "content": _to_text(user_msg)}) |
| if bot_msg: |
| messages.append({"role": "assistant", "content": _to_text(bot_msg)}) |
| messages.append({"role": "user", "content": _to_text(message)}) |
|
|
| text = tokenizer.apply_chat_template( |
| messages, tokenize=False, add_generation_prompt=True |
| ) |
| model_inputs = tokenizer([text], return_tensors="pt").to(model.device) |
|
|
| streamer = TextIteratorStreamer( |
| tokenizer, skip_prompt=True, skip_special_tokens=True |
| ) |
|
|
| generation_kwargs = dict( |
| **model_inputs, |
| streamer=streamer, |
| max_new_tokens=max_new_tokens, |
| temperature=temperature, |
| top_p=top_p, |
| do_sample=True, |
| ) |
|
|
| thread = Thread(target=model.generate, kwargs=generation_kwargs) |
| thread.start() |
|
|
| partial_text = "" |
| for new_text in streamer: |
| partial_text += new_text |
| yield partial_text |
|
|
|
|
| with gr.Blocks(title="Qwen 2.5 14B Chat + Auto Web Search") as demo: |
| gr.Markdown("# 🤖 Qwen 2.5 14B Instruct — General Chat") |
| gr.Markdown( |
| "Powered by ZeroGPU. Pehla response thoda slow ho sakta hai (GPU allocation + model warm-up). " |
| "🌐 Agar sawal current/real-time info ke baare mein ho (jaise 'aaj ka weather', 'latest news'), " |
| "bot khud automatically web search karke jawab dega — warna normal chat karega." |
| ) |
|
|
| with gr.Accordion("⚙️ Advanced settings", open=False): |
| system_prompt = gr.Textbox( |
| label="System Prompt", |
| value=SYSTEM_PROMPT, |
| lines=2, |
| ) |
| max_new_tokens = gr.Slider(64, 2048, value=512, step=32, label="Max new tokens") |
| temperature = gr.Slider(0.1, 1.5, value=0.7, step=0.05, label="Temperature") |
| top_p = gr.Slider(0.1, 1.0, value=0.9, step=0.05, label="Top-p") |
|
|
| chat = gr.ChatInterface( |
| fn=generate, |
| additional_inputs=[system_prompt, max_new_tokens, temperature, top_p], |
| title=None, |
| examples=[ |
| ["Aaj ka weather kaisa hai Delhi mein?"], |
| ["Explain quantum computing in simple terms"], |
| ["Latest smartphone launches is month"], |
| ], |
| ) |
|
|
| demo.queue(max_size=20) |
|
|
| if __name__ == "__main__": |
| demo.launch() |
|
|
