Update app.py

app.py

@@ -1,28 +1,50 @@
 import os
-import json
-import re
-import gradio as gr
-import pandas as pd
-import requests
-import random
-import urllib.parse
-from tempfile import NamedTemporaryFile
-from typing import List, Dict, Optional
-from bs4 import BeautifulSoup
 import logging
-from duckduckgo_search import DDGS
+import gradio as gr
+from huggingface_hub import hf_hub_download
+from transformers import HuggingFaceHub
+
+from llama_cpp_agent.providers import LlamaCppPythonProvider
+from llama_cpp_agent import LlamaCppAgent, MessagesFormatterType
+from llama_cpp_agent.chat_history import BasicChatHistory
+from llama_cpp_agent.chat_history.messages import Roles
+from llama_cpp_agent.llm_output_settings import (
+    LlmStructuredOutputSettings,
+    LlmStructuredOutputType,
+)
+from llama_cpp_agent.tools import WebSearchTool
+from llama_cpp_agent.prompt_templates import web_search_system_prompt, research_system_prompt
+from pydantic import BaseModel, Field
+from trafilatura import fetch_url, extract
+import json
+from datetime import datetime, timezone
+from typing import List
 
-from langchain_community.llms import HuggingFaceHub
-from langchain_community.vectorstores import FAISS
-from langchain_community.document_loaders import PyPDFLoader
-from langchain_community.embeddings import HuggingFaceEmbeddings
-from langchain_core.documents import Document
-from langchain.chains import LLMChain
-from langchain.prompts import PromptTemplate
+llm = None
+llm_model = None
 
-# Global variables
 huggingface_token = os.environ.get("HUGGINGFACE_TOKEN")
 
+examples = [
+    ["latest news about Yann LeCun"],
+    ["Latest news site:github.blog"],
+    ["Where I can find best hotel in Galapagos, Ecuador intitle:hotel"],
+    ["filetype:pdf intitle:python"]
+]
+
+def get_context_by_model(model_name):
+    model_context_limits = {
+        "Mistral-7B-Instruct-v0.3": 32768,
+    }
+    return model_context_limits.get(model_name, None)
+
+def get_messages_formatter_type(model_name):
+    model_name = model_name.lower()
+    if "mistral" in model_name:
+        return MessagesFormatterType.MISTRAL
+    else:
+        return MessagesFormatterType.CHATML
+
 def get_model(temperature, top_p, repetition_penalty):
     return HuggingFaceHub(
         repo_id="mistralai/Mistral-7B-Instruct-v0.3",
@@ -35,248 +57,190 @@ def get_model(temperature, top_p, repetition_penalty):
         huggingfacehub_api_token=huggingface_token
     )
 
-def load_document(file: NamedTemporaryFile) -> List[Document]:
-    loader = PyPDFLoader(file.name)
-    return loader.load_and_split()
-
-def update_vectors(files):
-    if not files:
-        return "Please upload at least one PDF file."
-    
-    embed = get_embeddings()
-    total_chunks = 0
-    
-    all_data = []
-    for file in files:
-        data = load_document(file)
-        all_data.extend(data)
-        total_chunks += len(data)
-    
-    if os.path.exists("faiss_database"):
-        database = FAISS.load_local("faiss_database", embed, allow_dangerous_deserialization=True)
-        database.add_documents(all_data)
-    else:
-        database = FAISS.from_documents(all_data, embed)
-    
-    database.save_local("faiss_database")
-    
-    return f"Vector store updated successfully. Processed {total_chunks} chunks from {len(files)} files."
-
-def get_embeddings():
-    return HuggingFaceEmbeddings(model_name="sentence-transformers/all-mpnet-base-v2")
-
-def clear_cache():
-    if os.path.exists("faiss_database"):
-        os.remove("faiss_database")
-        return "Cache cleared successfully."
-    else:
-        return "No cache to clear."
-
-def extract_text_from_webpage(html):
-    soup = BeautifulSoup(html, 'html.parser')
-    for script in soup(["script", "style"]):
-        script.extract()
-    text = soup.get_text()
-    lines = (line.strip() for line in text.splitlines())
-    chunks = (phrase.strip() for line in lines for phrase in line.split("  "))
-    text = '\n'.join(chunk for chunk in chunks if chunk)
-    return text
-
-_useragent_list = [
-    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
-    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
-    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Edge/91.0.864.59 Safari/537.36",
-    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Edge/91.0.864.59 Safari/537.36",
-    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Safari/537.36",
-    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Safari/537.36",
-]
-
-def google_search(term, num_results=5, lang="en", timeout=5, safe="active", ssl_verify=None):
-    escaped_term = urllib.parse.quote_plus(term)
-    start = 0
-    all_results = []
-    max_chars_per_page = 8000
-
-    with requests.Session() as session:
-        while start < num_results:
-            try:
-                user_agent = random.choice(_useragent_list)
-                headers = {
-                    'User-Agent': user_agent
-                }
-                resp = session.get(
-                    url="https://www.google.com/search",
-                    headers=headers,
-                    params={
-                        "q": term,
-                        "num": num_results - start,
-                        "hl": lang,
-                        "start": start,
-                        "safe": safe,
-                    },
-                    timeout=timeout,
-                    verify=ssl_verify,
-                )
-                resp.raise_for_status()
-            except requests.exceptions.RequestException as e:
-                print(f"Error retrieving search results: {e}")
-                break
-
-            soup = BeautifulSoup(resp.text, "html.parser")
-            result_block = soup.find_all("div", attrs={"class": "g"})
-            if not result_block:
-                break
-            
-            for result in result_block:
-                link = result.find("a", href=True)
-                if link:
-                    link = link["href"]
-                    try:
-                        webpage = session.get(link, headers=headers, timeout=timeout)
-                        webpage.raise_for_status()
-                        visible_text = extract_text_from_webpage(webpage.text)
-                        if len(visible_text) > max_chars_per_page:
-                            visible_text = visible_text[:max_chars_per_page] + "..."
-                        all_results.append({"link": link, "text": visible_text})
-                    except requests.exceptions.RequestException as e:
-                        print(f"Error retrieving webpage content: {e}")
-                        all_results.append({"link": link, "text": None})
-                else:
-                    all_results.append({"link": None, "text": None})
-            start += len(result_block)
+def get_server_time():
+    utc_time = datetime.now(timezone.utc)
+    return utc_time.strftime("%Y-%m-%d %H:%M:%S")
 
-    if not all_results:
-        return [{"link": None, "text": "No information found in the web search results."}]
-
-    return all_results
-
-def duckduckgo_search(query, max_results=5):
+def get_website_content_from_url(url: str) -> str:
     try:
-        search = DDGSearch()
-        results = search.text(query, max_results=max_results)
-        formatted_results = []
-        for result in results:
-            formatted_results.append({
-                "link": result.get('href', ''),
-                "text": result.get('title', '') + '. ' + result.get('body', '')
-            })
-        return formatted_results
+        downloaded = fetch_url(url)
+        result = extract(downloaded, include_formatting=True, include_links=True, output_format='json', url=url)
+        if result:
+            result = json.loads(result)
+            return f'=========== Website Title: {result["title"]} ===========\n\n=========== Website URL: {url} ===========\n\n=========== Website Content ===========\n\n{result["raw_text"]}\n\n=========== Website Content End ===========\n\n'
+        else:
+            return ""
     except Exception as e:
-        print(f"Error in DuckDuckGo search: {e}")
-        return [{"link": None, "text": "No information found in the web search results."}]
+        return f"An error occurred: {str(e)}"
+
+class CitingSources(BaseModel):
+    sources: List[str] = Field(
+        ...,
+        description="List of sources to cite. Should be an URL of the source. E.g. GitHub URL, Blogpost URL or Newsletter URL."
+    )
+
+def write_message_to_user():
+    return "Please write the message to the user."
 
+@spaces.GPU(duration=120)
 def respond(
     message,
     history: list[tuple[str, str]],
+    model,
+    system_message,
+    max_tokens,
     temperature,
     top_p,
-    repetition_penalty,
-    max_tokens,
-    search_engine
+    top_k,
+    repeat_penalty,
 ):
-    model = get_model(temperature, top_p, repetition_penalty)
-
-    # Perform web search
-    if search_engine == "Google":
-        search_results = google_search(message)
-    else:
-        search_results = duckduckgo_search(message, max_results=5)
-
-    # Check if we have a FAISS database
-    if os.path.exists("faiss_database"):
-        embed = get_embeddings()
-        database = FAISS.load_local("faiss_database", embed, allow_dangerous_deserialization=True)
-        retriever = database.as_retriever()
-        relevant_docs = retriever.get_relevant_documents(message)
-        context_str = "\n".join([doc.page_content for doc in relevant_docs])
-        
-        # Use the context in the prompt
-        prompt_template = f"""
-        Answer the question based on the following context and web search results:
-        Context from documents:
-        {context_str}
-        
-        Web Search Results:
-        {{search_results}}
-        
-        Question: {{message}}
-        
-        If the context and web search results don't contain relevant information, state that the information is not available.
-        Provide a concise and direct answer to the question.
-        """
-    else:
-        prompt_template = """
-        Answer the question based on the following web search results:
-        Web Search Results:
-        {search_results}
-        
-        Question: {message}
-        
-        If the web search results don't contain relevant information, state that the information is not available.
-        Provide a concise and direct answer to the question.
-        """
+    global llm
+    global llm_model
+    chat_template = get_messages_formatter_type(model)
+    if llm is None or llm_model != model:
+        llm = get_model(temperature, top_p, repeat_penalty)
+        llm_model = model
+    provider = LlamaCppPythonProvider(llm)
+    logging.info(f"Loaded chat examples: {chat_template}")
+    search_tool = WebSearchTool(
+        llm_provider=provider,
+        message_formatter_type=chat_template,
+        max_tokens_search_results=12000,
+        max_tokens_per_summary=2048,
+    )
 
-    prompt = PromptTemplate(
-        input_variables=["search_results", "message"],
-        template=prompt_template
+    web_search_agent = LlamaCppAgent(
+        provider,
+        system_prompt=web_search_system_prompt,
+        predefined_messages_formatter_type=chat_template,
+        debug_output=True,
     )
 
-    chain = LLMChain(llm=model, prompt=prompt)
+    answer_agent = LlamaCppAgent(
+        provider,
+        system_prompt=research_system_prompt,
+        predefined_messages_formatter_type=chat_template,
+        debug_output=True,
+    )
 
-    search_results_text = "\n".join([f"- {result['text']}" for result in search_results if result['text']])
-    response = chain.run(search_results=search_results_text, message=message)
+    settings = provider.get_provider_default_settings()
+    settings.stream = False
+    settings.temperature = temperature
+    settings.top_k = top_k
+    settings.top_p = top_p
+    settings.max_tokens = max_tokens
+    settings.repeat_penalty = repeat_penalty
 
-    # Add sources
-    sources = set(result["link"] for result in search_results if result["link"])
-    sources_section = "\n\nSources:\n" + "\n".join(f"- {source}" for source in sources)
-    response += sources_section
+    output_settings = LlmStructuredOutputSettings.from_functions(
+        [search_tool.get_tool()]
+    )
 
-    # Update history and return
-    history.append((message, response))
-    return history
+    messages = BasicChatHistory()
+
+    for msn in history:
+        user = {"role": Roles.user, "content": msn[0]}
+        assistant = {"role": Roles.assistant, "content": msn[1]}
+        messages.add_message(user)
+        messages.add_message(assistant)
+
+    result = web_search_agent.get_chat_response(
+        message,
+        llm_sampling_settings=settings,
+        structured_output_settings=output_settings,
+        add_message_to_chat_history=False,
+        add_response_to_chat_history=False,
+        print_output=False,
+    )
 
-# Gradio interface
-demo = gr.Blocks()
+    outputs = ""
+
+    settings.stream = True
+    response_text = answer_agent.get_chat_response(
+        f"Write a detailed and complete research document that fulfills the following user request: '{message}', based on the information from the web below.\n\n" +
+        result[0]["return_value"],
+        role=Roles.tool,
+        llm_sampling_settings=settings,
+        chat_history=messages,
+        returns_streaming_generator=True,
+        print_output=False,
+    )
 
-with demo:
-    gr.Markdown("# Chat with your PDF documents and Web Search")
-    
-    with gr.Row():
-        file_input = gr.Files(label="Upload your PDF documents", file_types=[".pdf"])
-        update_button = gr.Button("Upload PDF")
-    
-    update_output = gr.Textbox(label="Update Status")
-    update_button.click(update_vectors, inputs=[file_input], outputs=update_output)
-    
-    with gr.Row():
-        with gr.Column(scale=2):
-            chatbot = gr.Chatbot(label="Conversation")
-            message_input = gr.Textbox(label="Enter your message")
-            submit_button = gr.Button("Submit")
-        with gr.Column(scale=1):
-            temperature = gr.Slider(minimum=0.1, maximum=1.0, value=0.7, step=0.1, label="Temperature")
-            top_p = gr.Slider(minimum=0.1, maximum=1.0, value=0.95, step=0.05, label="Top-p")
-            repetition_penalty = gr.Slider(minimum=1.0, maximum=2.0, value=1.1, step=0.1, label="Repetition penalty")
-            max_tokens = gr.Slider(minimum=1, maximum=1000, value=500, step=1, label="Max tokens")
-            search_engine = gr.Dropdown(["DuckDuckGo", "Google"], value="DuckDuckGo", label="Search Engine")
+    for text in response_text:
+        outputs += text
+        yield outputs
 
-    submit_button.click(
-        respond,
-        inputs=[
-            message_input,
-            gr.State([]),  # Initialize empty history
-            temperature,
-            top_p,
-            repetition_penalty,
-            max_tokens,
-            search_engine
-        ],
-        outputs=[chatbot]
+    output_settings = LlmStructuredOutputSettings.from_pydantic_models(
+        [CitingSources], LlmStructuredOutputType.object_instance
     )
-    
-    clear_button = gr.Button("Clear Cache")
-    clear_output = gr.Textbox(label="Cache Status")
-    clear_button.click(clear_cache, inputs=[], outputs=clear_output)
 
-if __name__ == "__main__":
-    demo.launch()
+    citing_sources = answer_agent.get_chat_response(
+        "Cite the sources you used in your response.",
+        role=Roles.tool,
+        llm_sampling_settings=settings,
+        chat_history=messages,
+        returns_streaming_generator=False,
+        structured_output_settings=output_settings,
+        print_output=False,
+    )
+    outputs += "\n\nSources:\n"
+    outputs += "\n".join(citing_sources.sources)
+    yield outputs
+
+demo = gr.ChatInterface(
+    respond,
+    additional_inputs=[
+        gr.Dropdown([
+            'Mistral-7B-Instruct-v0.3'
+        ],
+            value="Mistral-7B-Instruct-v0.3",
+            label="Model"
+        ),
+        gr.Textbox(value=web_search_system_prompt, label="System message"),
+        gr.Slider(minimum=1, maximum=4096, value=2048, step=1, label="Max tokens"),
+        gr.Slider(minimum=0.1, maximum=1.0, value=0.45, step=0.1, label="Temperature"),
+        gr.Slider(
+            minimum=0.1,
+            maximum=1.0,
+            value=0.95,
+            step=0.05,
+            label="Top-p",
+        ),
+        gr.Slider(
+            minimum=0,
+            maximum=100,
+            value=40,
+            step=1,
+            label="Top-k",
+        ),
+        gr.Slider(
+            minimum=0.0,
+            maximum=2.0,
+            value=1.1,
+            step=0.1,
+            label="Repetition penalty",
+        ),
+    ],
+    theme=gr.themes.Soft(
+        primary_hue="orange",
+        secondary_hue="amber",
+        neutral_hue="gray",
+        font=[gr.themes.GoogleFont("Exo"), "ui-sans-serif", "system-ui", "sans-serif"]).set(
+            body_background_fill_dark="#0c0505",
+            block_background_fill_dark="#0c0505",
+            block_border_width="1px",
+            block_title_background_fill_dark="#1b0f0f",
+            input_background_fill_dark="#140b0b",
+            button_secondary_background_fill_dark="#140b0b",
+            border_color_accent_dark="#1b0f0f",
+            border_color_primary_dark="#1b0f0f",
+            slider_color="#ff911a",
+            button_primary_background_fill="#ff911a",
+            button_primary_background_fill_dark="#ff911a",
+            button_primary_text_color="#f9f9f9",
+            button_primary_text_color_dark="#f9f9f9"
+        ),
+    examples=examples,
+    title="llama.cpp agent",
+)
+
+demo.queue().launch()