app.py

from huggingface_hub import login, InferenceClient
import os, gc, time, random, datetime, json, re
HF_TOKEN=os.getenv('HF_TOKEN')
SERP_API_KEY=os.getenv('SERP_KEY')
login(token=HF_TOKEN)
import gradio as gr
from transformers import CodeAgent, Tool, ToolCollection, load_tool, ReactCodeAgent, ReactJsonAgent
from transformers.agents import PythonInterpreterTool
from langchain.memory import ConversationBufferMemory
import bs4
import requests
from llm_engine import HfEngine
import datasets
import spaces
import tqdm
from langchain_huggingface.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.docstore.document import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.vectorstores import VectorStore
from transformers.agents.prompts import DEFAULT_REACT_CODE_SYSTEM_PROMPT, DEFAULT_REACT_JSON_SYSTEM_PROMPT
from transformers.agents.default_tools import Tool, PythonInterpreterTool
from duckduckgo_search import DDGS
from web_surfer import (SearchInformationTool, NavigationalSearchTool, VisitTool, DownloadTool, PageUpTool, PageDownTool, FinderTool, FindNextTool, ArchiveSearchTool,)
from mdconvert import MarkdownConverter
from visual_qa import VisualQATool, VisualQAGPT4Tool
HF_HUB_DISABLE_TELEMETRY=1
DO_NOT_TRACK=1
HF_HUB_ENABLE_HF_TRANSFER=0
def search_ducky(query):
    with DDGS() as ddgs:
        results = list(ddgs.text(query, max_results=10))
        content = ''
        if results:
            for result in results:
                content += result['body']
                return content
knowledge_base = datasets.load_dataset("m-ric/huggingface_doc", split="train")
source_docs = [Document(page_content=doc["text"], metadata={"source": doc["source"].split("/")[1]}) for doc in knowledge_base]
docs_processed = RecursiveCharacterTextSplitter(chunk_size=500).split_documents(source_docs)[:1000]
embedding_model = HuggingFaceEmbeddings(model_name="thenlper/gte-small")
vectordb = FAISS.from_documents(documents=docs_processed, embedding=embedding_model)
all_sources = list(set([doc.metadata["source"] for doc in docs_processed]))
print(all_sources)
class RetrieverTool(Tool):
    name = "retriever"
    description = "Retrieves some documents from the knowledge base that have the closest embeddings to the input query."
    inputs = {
        "query": {
            "type": "text",
            "description": "The query to perform. This should be semantically close to your target documents. Use the affirmative form rather than a question.",
        },
        "source": {
            "type": "text", 
            "description": ""
        },
    }
    output_type = "text"
    
    def __init__(self, vectordb: VectorStore, all_sources: str, **kwargs):
        super().__init__(**kwargs)
        self.vectordb = vectordb
        self.inputs["source"]["description"] = (f"The source of the documents to search, as a str representation of a list. Possible values in the list are: {all_sources}. If this argument is not provided, all sources will be searched.")

    def forward(self, query: str, source: str = None) -> str:
        assert isinstance(query, str), "Your search query must be a string"

        if source:
            if isinstance(source, str) and "[" not in str(source): # if the source is not representing a list
                source = [source]
            source = json.loads(str(source).replace("'", '"'))

        docs = self.vectordb.similarity_search(query, filter=({"source": source} if source else None), k=3)

        if len(docs) == 0:
            return "No documents found with this filtering. Try removing the source filter."
        return "Retrieved documents:\n\n" + "\n===Document===\n".join([doc.page_content for doc in docs])
memory = ConversationBufferMemory(memory_key="chat_history")
llm_engine = HfEngine(model="Jopmt/JoPmt")
##gradio_prompt_generator_tool = StableDiffusionPromptGeneratorTool()
##prompt_generator_tool = Tool.from_gradio(gradio_prompt_generator_tool)
##tools = [StableDiffusionTool().langchain, ImageCaptioningTool().langchain, StableDiffusionPromptGeneratorTool().langchain, TextToVideoTool().langchain]
##tools=[prompt_generator_tool(), image_generation_tool(), PythonInterpreterTool()]
class SearchTool(Tool):
    name = "ask_search_agent"
    description = "A search agent that will browse the internet to answer a question. Use it to gather informations, not for problem-solving."

    inputs = {
        "question": {
            "description": "Your question, as a natural language sentence. You are talking to an agent, so provide them with as much context as possible.",
            "type": "text",
        }
    }
    output_type = "text"

    def forward(self, question: str) -> str:
        return websurfer_agent.run(question)
tools=[PythonInterpreterTool(),SearchTool(),RetrieverTool(vectordb, all_sources)]
additional_authorized_imports=['requests', 'bs4', 'os', 'time', 'datetime', 'json', 're']
WEB_TOOLS = [SearchInformationTool(), NavigationalSearchTool(), VisitTool(), DownloadTool(), PageUpTool(), PageDownTool(), FinderTool(), FindNextTool(), ArchiveSearchTool(),]
websurfer_agent = ReactJsonAgent(tools=WEB_TOOLS,llm_engine=llm_engine, add_base_tools=True,max_iterations=1)
reagent = ReactCodeAgent(tools=tools, llm_engine=llm_engine, add_base_tools=True,max_iterations=1,additional_authorized_imports=additional_authorized_imports)
def plix(inut, progress=gr.Progress(track_tqdm=True)):
    goose=reagent.run(inut)
    return goose
with gr.Blocks(theme=random.choice([gr.themes.Monochrome(),gr.themes.Base.from_hub("gradio/seafoam"),gr.themes.Base.from_hub("freddyaboulton/dracula_revamped"),gr.themes.Glass(),gr.themes.Base(),]),analytics_enabled=False) as iface:
    out=gr.Textbox(label="🤗Output",lines=5,interactive=False)
    inut=gr.Textbox(label="Prompt")
    btn=gr.Button("GENERATE")
    btn.click(fn=plix,inputs=inut,outputs=out)
iface.queue(max_size=1,api_open=False)
iface.launch(max_threads=20,inline=False,show_api=False)