Upload 3 files

app.py

@@ -1,53 +1,157 @@
+from langchain.agents import create_tool_calling_agent
+from langchain.agents import AgentExecutor
+import os
+from langchain_openai import ChatOpenAI
+from langchain.agents import  Tool
+from serp import GoogleSerperAPIWrapper, get_youtube_url
+from langchain_core.prompts import ChatPromptTemplate
+from langchain_core.messages import HumanMessage, AIMessage
+import base64
+from PIL import Image
+import io
+
+def encode_image(image_path):
+    with open(image_path, "rb") as image_file:
+        return base64.b64encode(image_file.read()).decode('utf-8')
+
+os.environ["SERPER_API_KEY"] = '2a'
+os.environ['OPENAI_API_KEY'] = "sk-"
+
+llm = ChatOpenAI(temperature=0, model_name='gpt-4o', openai_api_key=os.environ['OPENAI_API_KEY'])
+search_web = GoogleSerperAPIWrapper()
+search_images = GoogleSerperAPIWrapper(type="images")
+tools = [
+    Tool(
+        name="web_search",
+        func=search_web.run,
+        description="useful when you need to extract from the internet a list of websites snippet and a **valid URL**"
+    ),
+    Tool(
+        name="image_search",
+        func=search_images.run,
+        description="useful when you need to extract from the internet a list of images with the their titles and a **valid URL**"
+    ),
+    Tool(
+        name="video_search",
+        func=get_youtube_url,
+        description="useful when you need to extract from the internet a list of videos. The output is a list with a **valid URL**"
+    ),
+]
+
+# prompt = ChatPromptTemplate.from_messages([
+#     self.system_prompt,
+#     self.source_prompt,
+#     self.generate_eval_message(url)])
+
+agent_prompt = ChatPromptTemplate.from_messages(
+    [
+        (
+            "system",
+            "You are a helpful assistant that can provide informative urls from the web for any request.Review the attached image and collect from the internet resources which are related and helpful to continue writing the document. This includes website, images and videos Group the resources you collect by type and subject. You should collect at least 2 images and 2 websites and 1 video."
+        ),
+        ("human", "{input}"),
+        ("placeholder", "{agent_scratchpad}"),
+    ]
+)
+
+agent = create_tool_calling_agent(llm, tools, agent_prompt)
+
+agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
 
 import gradio as gr
 import os
 from openai import OpenAI
 
-
 with gr.Blocks() as demo:
     with gr.Row():
         image = gr.Image(label="image", height=600)
         chatbot = gr.Chatbot()
 
-    prompt = gr.Textbox(label="prompt")
-    url = gr.Textbox(label="url")
+    #prompt = gr.Textbox(label="prompt")
+    button = gr.Button()
+    serper_api = gr.Textbox(label="Serper API key")
     openai_key = gr.Textbox(label="OpenAI API key")
-    gr.Examples(
-        examples=[
-            ["https://huggingface.co/Adapter/t2iadapter/resolve/main/figs_SDXLV1.0/org_sketch.png", "Describe what is in the image","https://huggingface.co/Adapter/t2iadapter/resolve/main/figs_SDXLV1.0/org_sketch.png"]
-        ],
-        inputs=[image, prompt,url],
-    )
-
-    def respond(message,openai_key, url ,chat_history):
-        os.environ["OPENAI_API_KEY"] = openai_key 
-        client = OpenAI()
-
-        response = client.chat.completions.create(
-            model="gpt-4-vision-preview",
-            messages=[
-                {
-                    "role": "user",
-                    "content": [
-                        {"type": "text", "text": message},
-                        {
-                            "type": "image_url",
-                            "image_url":url,
-                        },
-                    ],
-                },
-            ],
-            max_tokens=1000,
+
+
+    def respond(chat_history, image):
+        out = agent_executor.invoke({'input': ''})
+
+        chat_history.append(('', out['output']))
+        return chat_history
+
+
+    def update_serper_api(serper_api):
+        print(os.environ['OPENAI_API_KEY'])
+        print(serper_api)
+        os.environ["SERPER_API_KEY"] = serper_api
+        search_web = GoogleSerperAPIWrapper()
+        search_images = GoogleSerperAPIWrapper(type="images")
+        global tools
+        tools = [
+            Tool(
+                name="web_search",
+                func=search_web.run,
+                description="useful when you need to extract from the internet a list of websites snippet and a **valid URL**"
+            ),
+            Tool(
+                name="image_search",
+                func=search_images.run,
+                description="useful when you need to extract from the internet a list of images with the their titles and a **valid URL**"
+            ),
+            Tool(
+                name="video_search",
+                func=get_youtube_url,
+                description="useful when you need to extract from the internet a list of videos. The output is a list with a **valid URL**"
+            ),
+        ]
+        agent = create_tool_calling_agent(llm, tools, agent_prompt)
+        global agent_executor
+        agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
+
+
+    def update_agent(openai_key):
+        os.environ['OPENAI_API_KEY'] = openai_key
+        print(os.environ['OPENAI_API_KEY'])
+        global llm
+        llm = ChatOpenAI(temperature=0, model_name='gpt-4o', openai_api_key=os.environ['OPENAI_API_KEY'])
+        agent = create_tool_calling_agent(llm, tools, agent_prompt)
+        global agent_executor
+        agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
+
+    def change_image(image):
+        image_pil = Image.fromarray(image)
+
+        # Save the image to a bytes buffer
+        buffer = io.BytesIO()
+        image_pil.save(buffer, format="PNG")  # You can also use "JPEG" if needed
+
+        # Get the byte data from the buffer and encode it to base64
+        image_bytes = buffer.getvalue()
+        image_base64 = base64.b64encode(image_bytes).decode('utf-8')
+        message_content = [{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,"
+                                                                                      f"{image_base64}"}}]
+        image_message = HumanMessage(content=message_content)
+        global agent_prompt
+        agent_prompt = ChatPromptTemplate.from_messages(
+            [
+                (
+                    "system",
+                    "You are a helpful assistant that can provide informative urls from the web for any request.Review the attached image and collect from the internet resources which are related and helpful to continue writing the document. This includes website, images Group the resources you collect by type and subject. You should collect at least 3 images and 3 websites."
+                ),
+                image_message,
+                ("human", "{input}"),
+                ("placeholder", "{agent_scratchpad}"),
+            ]
         )
 
-        out = response.choices[0].message.content
 
-        chat_history.append((message, out))
-        return "", chat_history
 
-    def update_image(url, image):
-        return url
+        agent = create_tool_calling_agent(llm, tools, agent_prompt)
+        global agent_executor
+        agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
 
-    prompt.submit(respond, [prompt,openai_key, url, chatbot], [prompt, chatbot])
-    url.submit(update_image, [ url, image], [image])
-demo.queue().launch(share=True)
+    button.click(respond, [chatbot, image], [chatbot])
+    openai_key.submit(update_agent, [openai_key], [])
+    serper_api.submit(update_serper_api, [serper_api], [])
+    image.change(change_image,[image],[])
+demo.queue().launch(share=True)

requirements.txt

@@ -1 +1,12 @@
-openai
+tqdm==4.66.1
+langchain==0.2.7
+openai==1.35.10
+tiktoken==0.7.0
+easydict==1.11
+sentence-transformers==2.2.2
+langchain-google-genai==1.0.8
+pillow==10.2.0
+langchain_openai==0.1.20
+langchain_community
+gradio
+youtube_search

serp.py

@@ -0,0 +1,238 @@
+"""Util that calls Google Search using the Serper.dev API."""
+
+from typing import Any, Dict, List, Optional
+
+import aiohttp
+import requests
+from langchain_core.pydantic_v1 import BaseModel, root_validator
+from langchain_core.utils import get_from_dict_or_env
+from typing_extensions import Literal
+
+import requests
+import json
+
+def check_link_no_redirect(url):
+    try:
+        # Make a HEAD request without allowing redirects, with a 0.5-second timeout
+        response = requests.head(url, allow_redirects=False, timeout=0.3)
+
+        # Check for successful status code (200 OK) and no redirection (3xx codes)
+        if response.status_code == 200:
+            return True
+        elif 300 <= response.status_code < 400:
+            return False
+        else:
+            return False
+    except requests.exceptions.Timeout:
+        return False
+    except requests.exceptions.RequestException as e:
+        return False
+
+
+
+
+class GoogleSerperAPIWrapper(BaseModel):
+    """Wrapper around the Serper.dev Google Search API.
+
+    You can create a free API key at https://serper.dev.
+
+    To use, you should have the environment variable ``SERPER_API_KEY``
+    set with your API key, or pass `serper_api_key` as a named parameter
+    to the constructor.
+
+    Example:
+        .. code-block:: python
+
+            from langchain_community.utilities import GoogleSerperAPIWrapper
+            google_serper = GoogleSerperAPIWrapper()
+    """
+
+    k: int = 10
+    gl: str = "us"
+    hl: str = "en"
+    # "places" and "images" is available from Serper but not implemented in the
+    # parser of run(). They can be used in results()
+    type: Literal["news", "search", "places", "images"] = "search"
+    result_key_for_type = {
+        "news": "news",
+        "places": "places",
+        "images": "images",
+        "search": "organic",
+    }
+
+    tbs: Optional[str] = None
+    serper_api_key: Optional[str] = None
+    aiosession: Optional[aiohttp.ClientSession] = None
+
+    class Config:
+        """Configuration for this pydantic object."""
+
+        arbitrary_types_allowed = True
+
+    @root_validator(pre=True)
+    def validate_environment(cls, values: Dict) -> Dict:
+        """Validate that api key exists in environment."""
+        serper_api_key = get_from_dict_or_env(
+            values, "serper_api_key", "SERPER_API_KEY"
+        )
+        values["serper_api_key"] = serper_api_key
+
+        return values
+
+    def results(self, query: str, **kwargs: Any) -> Dict:
+        """Run query through GoogleSearch."""
+        return self._google_serper_api_results(
+            query,
+            gl=self.gl,
+            hl=self.hl,
+            num=self.k,
+            tbs=self.tbs,
+            search_type=self.type,
+            **kwargs,
+        )
+
+    def run(self, query: str, **kwargs: Any) -> str:
+        """Run query through GoogleSearch and parse result."""
+        results = self._google_serper_api_results(
+            query,
+            gl=self.gl,
+            hl=self.hl,
+            num=self.k,
+            tbs=self.tbs,
+            search_type=self.type,
+            **kwargs,
+        )
+
+        return self._parse_results(results)
+
+    async def aresults(self, query: str, **kwargs: Any) -> Dict:
+        """Run query through GoogleSearch."""
+        results = await self._async_google_serper_search_results(
+            query,
+            gl=self.gl,
+            hl=self.hl,
+            num=self.k,
+            search_type=self.type,
+            tbs=self.tbs,
+            **kwargs,
+        )
+        return results
+
+    async def arun(self, query: str, **kwargs: Any) -> str:
+        """Run query through GoogleSearch and parse result async."""
+        results = await self._async_google_serper_search_results(
+            query,
+            gl=self.gl,
+            hl=self.hl,
+            num=self.k,
+            search_type=self.type,
+            tbs=self.tbs,
+            **kwargs,
+        )
+
+        return self._parse_results(results)
+
+    def _parse_snippets(self, results: dict) -> List[str]:
+        snippets = []
+
+        # if results.get("answerBox"):
+        #     answer_box = results.get("answerBox", {})
+        #     if answer_box.get("answer"):
+        #         return [answer_box.get("answer")]
+        #     elif answer_box.get("snippet"):
+        #         return [answer_box.get("snippet").replace("\n", " ")]
+        #     elif answer_box.get("snippetHighlighted"):
+        #         return answer_box.get("snippetHighlighted")
+        #
+        # if results.get("knowledgeGraph"):
+        #     kg = results.get("knowledgeGraph", {})
+        #     title = kg.get("title")
+        #     entity_type = kg.get("type")
+        #     if entity_type:
+        #         snippets.append(f"{title}: {entity_type}.")
+        #     description = kg.get("description")
+        #     if description:
+        #         snippets.append(description)
+        #     for attribute, value in kg.get("attributes", {}).items():
+        #         snippets.append(f"{title} {attribute}: {value}.")
+
+        for result in results[self.result_key_for_type[self.type]][: self.k]:
+            if "snippet" in result:
+                if not check_link_no_redirect(result['link']):
+                    continue
+                snippets.append('Snippet: {}\nUrl: {}'.format(result['snippet'],result['link']))
+
+        if len(snippets) == 0:
+            return ["No good Google Search Result was found"]
+        return snippets
+
+    def _parse_results(self, results: dict) -> str:
+        all_res = []
+        if self.type == "images":
+            for image in results["images"][: self.k]:
+                if not check_link_no_redirect(image['imageUrl']):
+                    continue
+                all_res.append('Title: {}\nUrl: {}'.format(image['title'], image['imageUrl']))
+            return "\n-----\n".join(all_res)
+        return "\n-----\n".join(self._parse_snippets(results))
+
+    def _google_serper_api_results(
+        self, search_term: str, search_type: str = "search", **kwargs: Any
+    ) -> dict:
+        headers = {
+            "X-API-KEY": self.serper_api_key or "",
+            "Content-Type": "application/json",
+        }
+        params = {
+            "q": search_term,
+            **{key: value for key, value in kwargs.items() if value is not None},
+        }
+        response = requests.post(
+            f"https://google.serper.dev/{search_type}", headers=headers, params=params
+        )
+        response.raise_for_status()
+        search_results = response.json()
+        return search_results
+
+    async def _async_google_serper_search_results(
+        self, search_term: str, search_type: str = "search", **kwargs: Any
+    ) -> dict:
+        headers = {
+            "X-API-KEY": self.serper_api_key or "",
+            "Content-Type": "application/json",
+        }
+        url = f"https://google.serper.dev/{search_type}"
+        params = {
+            "q": search_term,
+            **{key: value for key, value in kwargs.items() if value is not None},
+        }
+
+        if not self.aiosession:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    url, params=params, headers=headers, raise_for_status=False
+                ) as response:
+                    search_results = await response.json()
+        else:
+            async with self.aiosession.post(
+                url, params=params, headers=headers, raise_for_status=True
+            ) as response:
+                search_results = await response.json()
+
+        return search_results
+
+
+
+
+
+
+
+from youtube_search import YoutubeSearch
+def get_youtube_url(query:str) -> str:
+    num_results = 4
+    results = YoutubeSearch(query, num_results).to_json()
+    data = json.loads(results)
+    all_data = []
+    for video in data["videos"][:num_results]:
+        all_data.append('Title: {}\nUrl: {}'.format(video['title'], 'https://www.youtube.com' + video['url_suffix']))
+    return  "\n-----\n".join(all_data)