change new logics

app.py

@@ -1,88 +1,101 @@
-# Import các thư viện cần thiết
 import os
-import json  # Thêm thư viện json để xử lý dữ liệu JSON
+import time
+import requests
+import random
 from threading import Thread
-from typing import Iterator
-
-import gradio as gr
-import spaces
+from typing import List, Dict, Union
 import torch
-import requests  # Thêm thư viện requests để thực hiện các yêu cầu HTTP
-from bs4 import BeautifulSoup  # Thêm BeautifulSoup để phân tích HTML
-from functools import lru_cache  # Thêm lru_cache để cache kết quả hàm
+import gradio as gr
+from bs4 import BeautifulSoup
 from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer
+from functools import lru_cache
+import re
+import io 
+import json
 
-# Mô tả về mô hình Llama 3.2 3B Instruct
-DESCRIPTION = """\
-# Llama 3.2 3B Instruct
-
-Llama 3.2 3B là phiên bản mới nhất của LLM từ Meta.
-Đây là bản demo của [`meta-llama/Llama-3.2-3B-Instruct`](https://huggingface.co/meta-llama/Llama-3.2-3B-Instruct), được tinh chỉnh để theo dõi hướng dẫn.
-Để biết thêm chi tiết, hãy xem [bài đăng của chúng tôi](https://huggingface.co/blog/llama32).
-"""
-
-# Thiết lập các thông số tối đa
-MAX_MAX_NEW_TOKENS = 2048  # Số token tối đa cho đầu ra mới
-DEFAULT_MAX_NEW_TOKENS = 1024  # Số token mặc định cho đầu ra mới
-MAX_INPUT_TOKEN_LENGTH = int(os.getenv("MAX_INPUT_TOKEN_LENGTH", "4096"))  # Lấy giá trị chiều dài token đầu vào từ biến môi trường
+# Model Loading (Done once at startup)
+MODEL_ID = "meta-llama/Llama-3.2-3B-Instruct"
+tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
+model = AutoModelForCausalLM.from_pretrained(
+    MODEL_ID,
+    torch_dtype=torch.float16,
+    device_map="cuda",
+    low_cpu_mem_usage=True
+).eval()
 
-# Kiểm tra thiết bị có hỗ trợ GPU không, nếu không thì sử dụng CPU
-device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
+# Path to example texts (Updated to remove image and video examples)
+examples_path = os.path.dirname(__file__)
+EXAMPLES = [
+    [
+        {
+            "text": "What is Friction? Explain in Detail.",
+        }
+    ],
+    [
+        {
+            "text": "Write me a Python function to generate unique passwords.",
+        }
+    ],
+    [
+        {
+            "text": "What's the latest price of Bitcoin?",
+        }
+    ],
+    [
+        {
+            "text": "Search and give me list of spaces trending on HuggingFace.",
+        }
+    ],
+    [
+        {
+            "text": "Create a Beautiful Picture of Eiffel at Night.",
+        }
+    ],
+    [
+        {
+            "text": "What unusual happens in this video.",
+            "files": [f"{examples_path}/example_video/accident.gif"],
+        }
+    ],
+    # Removed other image and video related examples
+]
 
-# Định danh mô hình và tải mô hình cùng tokenizer
-model_id = "nltpt/Llama-3.2-3B-Instruct"
-tokenizer = AutoTokenizer.from_pretrained(model_id)
-model = AutoModelForCausalLM.from_pretrained(
-    model_id,
-    device_map="auto",  # Tự động ánh xạ thiết bị
-    torch_dtype=torch.bfloat16,  # Sử dụng kiểu dữ liệu bfloat16
-)
-model.eval()  # Đặt mô hình vào chế độ đánh giá (evaluation mode)
+# Set bot avatar image
+BOT_AVATAR = "OpenAI_logo.png"
 
-# Hàm để trích xuất văn bản từ nội dung HTML, sử dụng cache để tăng tốc độ
-@lru_cache(maxsize=128)
+# Perform a Google search and return the results
+@lru_cache(maxsize=128) 
 def extract_text_from_webpage(html_content):
-    """Trích xuất văn bản hiển thị từ nội dung HTML sử dụng BeautifulSoup."""
+    """Extracts visible text from HTML content using BeautifulSoup."""
     soup = BeautifulSoup(html_content, "html.parser")
-    # Loại bỏ các thẻ không hiển thị như script, style, header, footer, nav, form, svg
     for tag in soup(["script", "style", "header", "footer", "nav", "form", "svg"]):
         tag.extract()
-    # Lấy văn bản hiển thị và loại bỏ khoảng trắng thừa
-    visible_text = soup.get_text(strip=True)
+    visible_text = soup.get_text(separator=' ', strip=True)
     return visible_text
 
-# Hàm thực hiện tìm kiếm trên Google và trả về kết quả
 def search(query):
     term = query
     all_results = []
-    max_chars_per_page = 8000  # Số ký tự tối đa cho mỗi trang
+    max_chars_per_page = 8000
     with requests.Session() as session:
-        # Thực hiện yêu cầu tìm kiếm trên Google
         resp = session.get(
             url="https://www.google.com/search",
             headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/111.0"},
-            params={"q": term, "num": 4, "udm": 14},
+            params={"q": term, "num": 4},
             timeout=5,
-            verify=False,  # Không xác thực SSL
+            verify=False,
         )
-        resp.raise_for_status()  # Kiểm tra phản hồi HTTP
+        resp.raise_for_status()
         soup = BeautifulSoup(resp.text, "html.parser")
-        result_block = soup.find_all("div", attrs={"class": "g"})  # Tìm các kết quả tìm kiếm
+        result_block = soup.find_all("div", attrs={"class": "g"})
         for result in result_block:
             link = result.find("a", href=True)
-            if link:
+            if link and 'href' in link.attrs:
                 link = link["href"]
                 try:
-                    # Lấy nội dung trang web từ liên kết
-                    webpage = session.get(
-                        link,
-                        headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/111.0"},
-                        timeout=5,
-                        verify=False
-                    )
+                    webpage = session.get(link, headers={"User-Agent": "Mozilla/5.0"}, timeout=5, verify=False)
                     webpage.raise_for_status()
                     visible_text = extract_text_from_webpage(webpage.text)
-                    # Cắt bớt văn bản nếu vượt quá giới hạn
                     if len(visible_text) > max_chars_per_page:
                         visible_text = visible_text[:max_chars_per_page]
                     all_results.append({"link": link, "text": visible_text})
@@ -90,243 +103,155 @@ def search(query):
                     all_results.append({"link": link, "text": None})
     return all_results
 
-# Danh sách metadata các hàm có thể được gọi
-functions_metadata = [
-    {
-        "name": "web_search",
-        "description": "Tìm kiếm trên Google và tìm thông tin mới nhất, thông tin về bất kỳ người, đối tượng, địa điểm nào có sẵn trên Google.",
-        "parameters": {
-            "type": "object",
-            "properties": {
-                "query": {
-                    "type": "string",
-                    "description": "Truy vấn tìm kiếm web"
+def generate_response(prompt, chat_history):
+    # Construct the conversation history
+    conversation = ""
+    for user, assistant in chat_history:
+        conversation += f"User: {user}\nAssistant: {assistant}\n"
+    conversation += f"User: {prompt}\nAssistant:"
+
+    inputs = tokenizer(conversation, return_tensors="pt").to("cuda")
+
+    streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True)
+    thread = Thread(target=model.generate, args=(inputs.input_ids,), kwargs={
+        "max_new_tokens": 512,
+        "do_sample": True,
+        "top_p": 0.95,
+        "temperature": 0.8,
+        "streamer": streamer
+    })
+    thread.start()
+
+    response = ""
+    for new_text in streamer:
+        response += new_text
+        yield response
+
+@lru_cache(maxsize=128)
+def process_query(query, chat_history):
+    # Here you can implement logic to decide between web_search, general_query, or hard_query
+    # For simplicity, let's assume all queries go through general_query
+    # You can expand this with your own logic
+    functions_metadata = [
+        {
+            "type": "function",
+            "function": {
+                "name": "web_search",
+                "description": "Search query on google and find latest information.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "query": {"type": "string", "description": "Web search query"}
+                    },
+                    "required": ["query"]
                 }
-            },
-            "required": ["query"]
-        }
-    },
-    {
-        "name": "general_query",
-        "description": "Trả lời các truy vấn chung của người dùng bằng LLM như bạn. Nhưng không trả lời các câu hỏi khó và thông tin mới nhất.",
-        "parameters": {
-            "type": "object",
-            "properties": {
-                "prompt": {
-                    "type": "string",
-                    "description": "Một đề bài chi tiết"
+            }
+        },
+        {
+            "type": "function",
+            "function": {
+                "name": "general_query",
+                "description": "Reply general query with LLM.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "prompt": {"type": "string", "description": "A detailed prompt"}
+                    },
+                    "required": ["prompt"]
                 }
-            },
-            "required": ["prompt"]
-        }
-    },
-    {
-        "name": "hard_query",
-        "description": "Trả lời các truy vấn khó của người dùng, sử dụng LLM mạnh mẽ. Nhưng không trả lời các thông tin mới nhất.",
-        "parameters": {
-            "type": "object",
-            "properties": {
-                "prompt": {
-                    "type": "string",
-                    "description": "Một đề bài chi tiết"
+            }
+        },
+        {
+            "type": "function",
+            "function": {
+                "name": "hard_query",
+                "description": "Reply tough query using powerful LLM.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "prompt": {"type": "string", "description": "A detailed prompt"}
+                    },
+                    "required": ["prompt"]
                 }
-            },
-            "required": ["prompt"]
-        }
-    },
-]
-
-# Prompt role system với các hướng dẫn về khả năng và công cụ
-SYSTEM_PROMPT = f"""\
-Bạn là một trợ lý thông minh và hữu ích với khả năng sử dụng các công cụ sau đây để hỗ trợ người dùng:
-
-{json.dumps(functions_metadata, indent=4, ensure_ascii=False)}
+            }
+        },
+    ]
 
-Bạn có thể thực hiện các công việc như tìm kiếm trên web, tạo hình ảnh, tạo video, và trả lời các câu hỏi liên quan đến hình ảnh. Khi cần sử dụng một trong các công cụ này, bạn sẽ trả về một lời gọi hàm theo định dạng Python như sau:
+    # Example logic to choose function (you can customize this)
+    if "search" in query.lower():
+        function_name = "web_search"
+    elif "explain" in query.lower() or "detail" in query.lower():
+        function_name = "general_query"
+    else:
+        function_name = "hard_query"
 
-[func1(params_name=params_value, params_name2=params_value2...), func2(params)]
+    return {
+        "name": function_name,
+        "arguments": {
+            "query" if function_name == "web_search" else "prompt": query
+        }
+    }
 
-Bạn chỉ được gọi một hàm tại một thời điểm. Đảm bảo rằng bạn chọn các công cụ một cách khôn ngoan để cung cấp câu trả lời tốt nhất cho người dùng.
+def handle_functions(function_call, chat_history):
+    function_name = function_call["name"]
+    arguments = function_call["arguments"]
 
-"""
+    if function_name == "web_search":
+        query = arguments["query"]
+        web_results = search(query)
+        web_summary = ' '.join([f"Link: {res['link']}\nText: {res['text']}\n\n" for res in web_results if res["text"]])
+        # Append web results to chat history or pass to the model as context
+        # Here we directly return the summarized web results
+        return f"Here are the search results:\n{web_summary}"
 
-# Hàm chính để xử lý việc sinh phản hồi từ mô hình
-@spaces.GPU(duration=30)  # Chỉ định hàm này chạy trên GPU trong tối đa 90 giây
-def generate(
-    message: str,
-    chat_history: list[tuple[str, str]],
-    max_new_tokens: int = 1024,
-    temperature: float = 0.6,
-    top_p: float = 0.9,
-    top_k: int = 50,
-    repetition_penalty: float = 1.2,
-) -> Iterator[str]:
-    conversation = []
-    func_caller = []
+    elif function_name in ["general_query", "hard_query"]:
+        prompt = arguments["prompt"]
+        # Generate response using the local model
+        response_generator = generate_response(prompt, chat_history)
+        return response_generator
 
-    # Xây dựng cuộc hội thoại từ lịch sử trò chuyện
-    for user, assistant in chat_history:
-        conversation.extend(
-            [
-                {"role": "user", "content": user},
-                {"role": "assistant", "content": assistant},
-            ]
-        )
-    
-    # Thêm tin nhắn mới của người dùng vào cuộc hội thoại
-    conversation.append({"role": "user", "content": message})
+    else:
+        return "Function not recognized."
 
-    # Thêm prompt role system vào đầu cuộc hội thoại
-    conversation.insert(0, {"role": "system", "content": SYSTEM_PROMPT})
+def model_inference(user_prompt, chat_history):
+    prompt = user_prompt["text"]
 
-    # Áp dụng mẫu hội thoại và chuyển thành tensor
-    input_ids = tokenizer.apply_chat_template(conversation, add_generation_prompt=True, return_tensors="pt")
-    
-    # Kiểm tra và cắt bớt chuỗi đầu vào nếu vượt quá chiều dài tối đa
-    if input_ids.shape[1] > MAX_INPUT_TOKEN_LENGTH:
-        input_ids = input_ids[:, -MAX_INPUT_TOKEN_LENGTH:]
-        gr.Warning(f"Đã cắt bớt đầu vào từ cuộc hội thoại vì vượt quá {MAX_INPUT_TOKEN_LENGTH} tokens.")
-    
-    # Chuyển tensor đến thiết bị của mô hình
-    input_ids = input_ids.to(model.device)
+    # Determine which function to call
+    function_call = process_query(prompt, chat_history)
 
-    # Khởi tạo Streamer để lấy đầu ra theo từng phần (real-time)
-    streamer = TextIteratorStreamer(tokenizer, timeout=20.0, skip_prompt=True, skip_special_tokens=True)
-    
-    # Thiết lập các tham số cho quá trình sinh đầu ra
-    generate_kwargs = dict(
-        {"input_ids": input_ids},
-        streamer=streamer,
-        max_new_tokens=max_new_tokens,
-        do_sample=True,
-        top_p=top_p,
-        top_k=top_k,
-        temperature=temperature,
-        num_beams=1,
-        repetition_penalty=repetition_penalty,
-    )
-    
-    # Tạo một luồng để chạy quá trình sinh đầu ra
-    t = Thread(target=model.generate, kwargs=generate_kwargs)
-    t.start()
+    if function_call["name"] == "web_search":
+        yield "Performing web search..."
+        result = handle_functions(function_call, chat_history)
+        yield result
 
-    outputs = []
-    for text in streamer:
-        outputs.append(text)
-        response = "".join(outputs)
-        yield response
+    elif function_call["name"] in ["general_query", "hard_query"]:
+        yield "Generating response..."
+        response_generator = handle_functions(function_call, chat_history)
+        for response in response_generator:
+            yield response
 
-        # Kiểm tra xem phản hồi có phải là lời gọi hàm không
-        if response.startswith("[") and response.endswith("]"):
-            try:
-                # Phân tích phản hồi JSON
-                json_data = json.loads(response)
-                
-                # Xử lý các hàm được gọi
-                for func in json_data:
-                    func_name = func.get("name")
-                    arguments = func.get("arguments", {})
-                    
-                    if func_name == "web_search":
-                        query = arguments.get("query")
-                        if query:
-                            gr.Info("Đang tìm kiếm trên web...")
-                            yield "Đang tìm kiếm trên web..."
-                            web_results = search(query)
-                            
-                            gr.Info("Đang trích xuất thông tin liên quan...")
-                            yield "Đang trích xuất thông tin liên quan..."
-                            web_content = ' '.join([f"Link: {res['link']}\nText: {res['text']}\n\n" for res in web_results])
-                            
-                            # Xây dựng lại cuộc hội thoại với kết quả tìm kiếm
-                            conversation_updated = conversation.copy()
-                            conversation_updated.append({"role": "user", "content": message})
-                            conversation_updated.append({"role": "system", "content": f"[WEB RESULTS] {web_content}"})
-                            
-                            # Gửi lại cuộc hội thoại để mô hình trả lời dựa trên kết quả tìm kiếm
-                            input_ids_updated = tokenizer.apply_chat_template(conversation_updated, add_generation_prompt=True, return_tensors="pt")
-                            input_ids_updated = input_ids_updated.to(model.device)
-                            
-                            streamer_updated = TextIteratorStreamer(tokenizer, timeout=20.0, skip_prompt=True, skip_special_tokens=True)
-                            generate_kwargs_updated = dict(
-                                {"input_ids": input_ids_updated},
-                                streamer=streamer_updated,
-                                max_new_tokens=max_new_tokens,
-                                do_sample=True,
-                                top_p=top_p,
-                                top_k=top_k,
-                                temperature=temperature,
-                                num_beams=1,
-                                repetition_penalty=repetition_penalty,
-                            )
-                            
-                            t_updated = Thread(target=model.generate, kwargs=generate_kwargs_updated)
-                            t_updated.start()
-                            
-                            for text_updated in streamer_updated:
-                                yield text_updated
-                    
-            except json.JSONDecodeError:
-                # Nếu phản hồi không phải là JSON hợp lệ, tiếp tục trả lời bình thường
-                continue
+    else:
+        yield "Invalid function call."
 
-# Tạo giao diện chat với Gradio
-chat_interface = gr.ChatInterface(
-    fn=generate,
-    additional_inputs=[
-        gr.Slider(
-            label="Số token mới tối đa",
-            minimum=1,
-            maximum=MAX_MAX_NEW_TOKENS,
-            step=1,
-            value=DEFAULT_MAX_NEW_TOKENS,
-        ),
-        gr.Slider(
-            label="Nhiệt độ (Temperature)",
-            minimum=0.1,
-            maximum=4.0,
-            step=0.1,
-            value=0.6,
-        ),
-        gr.Slider(
-            label="Top-p (nucleus sampling)",
-            minimum=0.05,
-            maximum=1.0,
-            step=0.05,
-            value=0.9,
-        ),
-        gr.Slider(
-            label="Top-k",
-            minimum=1,
-            maximum=1000,
-            step=1,
-            value=50,
-        ),
-        gr.Slider(
-            label="Hình phạt lặp lại (Repetition penalty)",
-            minimum=1.0,
-            maximum=2.0,
-            step=0.05,
-            value=1.2,
-        ),
-    ],
-    stop_btn=None,  # Không có nút dừng
-    examples=[
-        ["Xin chào! Bạn có khỏe không?"],
-        ["Bạn có thể giải thích ngắn gọn về ngôn ngữ lập trình Python không?"],
-        ["Giải thích cốt truyện của Cô bé Lọ Lem trong một câu."],
-        ["Mất bao nhiêu giờ để một người ăn một chiếc trực thăng?"],
-        ["Viết một bài báo 100 từ về 'Lợi ích của mã nguồn mở trong nghiên cứu AI'"],
-    ],
-    cache_examples=False,  # Không lưu trữ các ví dụ
+# Create a chatbot interface
+chatbot = gr.Chatbot(
+    label="OpenGPT-4o",
+    avatar_images=[None, BOT_AVATAR],
+    show_copy_button=True,
+    layout="panel",
+    height=400,
 )
+input_box = gr.Textbox(label="Prompt")
 
-# Tạo bố cục giao diện với Gradio
-with gr.Blocks(css="style.css", fill_height=True) as demo:
-    gr.Markdown(DESCRIPTION)  # Hiển thị phần mô tả
-    gr.DuplicateButton(value="Tạo bản sao cho sử dụng cá nhân", elem_id="duplicate-button")
-    chat_interface.render()  # Hiển thị giao diện chat
+iface = gr.Interface(
+    fn=model_inference,
+    inputs=[input_box, chatbot],
+    outputs=chatbot,
+    live=True,
+    examples=EXAMPLES,
+    title="OpenGPT-4o Chatbot",
+    description="A powerful AI assistant using local Llama-3.2 model.",
+)
 
-# Hàm chính để khởi chạy ứng dụng
 if __name__ == "__main__":
-    demo.queue(max_size=20).launch()  # Khởi chạy ứng dụng với hàng đợi tối đa 20 yêu cầu
+    iface.launch()