first tools implementation with langgraph

geminiAgent.py

@@ -0,0 +1,22 @@
+from typing import TypedDict, Literal, List, Dict, Optional
+import time
+import requests
+from bs4 import BeautifulSoup
+import operator
+from langchain_core.prompts import ChatPromptTemplate
+from langchain_core.runnables import RunnablePassthrough
+from langchain_core.messages import HumanMessage, AIMessage
+from langchain.tools import tool
+from langchain_core.output_parsers import StrOutputParser
+from langchain_core.pydantic_v1 import BaseModel, Field
+from langgraph.graph import StateGraph, START, END
+import pytesseract
+from PIL import Image
+
+@tool
+def ocr_tool(image: str) -> str:
+    """
+    A tool that performs OCR processing on an image.
+    """
+    image = Image.open(image)
+    return pytesseract.image_to_string(image)

requirements.txt

@@ -1,2 +1,4 @@
 gradio
-requests
+requests
+pytesseract
+PIL

tools/WebSearchTool.py

@@ -0,0 +1,75 @@
+import time
+from typing import Optional
+import requests
+from bs4 import BeautifulSoup
+from langchain.tools import tool
+
+class WebSearchTool:
+    def __init__(self):
+        self.last_request_time = 0
+        self.min_request_interval = 2.0  # Minimum time between requests in seconds
+        self.max_retries = 10
+
+    def search(self, query: str, domain: Optional[str] = None) -> str:
+        """Perform web search with rate limiting and retries."""
+        for attempt in range(self.max_retries):
+            # Implement rate limiting
+            current_time = time.time()
+            time_since_last = current_time - self.last_request_time
+            if time_since_last < self.min_request_interval:
+                time.sleep(self.min_request_interval - time_since_last)
+            
+            try:
+                # Make the search request
+                results = self._do_search(query, domain)
+                self.last_request_time = time.time()
+                return results
+            except Exception as e:
+                if "202 Ratelimit" in str(e):
+                    if attempt < self.max_retries - 1:
+                        # Exponential backoff
+                        wait_time = (2 ** attempt) * self.min_request_interval
+                        time.sleep(wait_time)
+                        continue
+                return f"Search failed after {self.max_retries} attempts: {str(e)}"
+        
+        return "Search failed due to rate limiting"
+
+    def _do_search(self, query: str, domain: Optional[str] = None) -> str:
+        """Perform the actual search request."""
+        try:
+            # Construct search URL
+            base_url = "https://html.duckduckgo.com/html"
+            params = {"q": query}
+            if domain:
+                params["q"] += f" site:{domain}"
+
+            # Make request with increased timeout
+            response = requests.get(base_url, params=params, timeout=10)
+            response.raise_for_status()
+
+            if response.status_code == 202:
+                raise Exception("202 Ratelimit")
+
+            # Extract search results
+            results = []
+            soup = BeautifulSoup(response.text, 'html.parser')
+            for result in soup.find_all('div', {'class': 'result'}):
+                title = result.find('a', {'class': 'result__a'})
+                snippet = result.find('a', {'class': 'result__snippet'})
+                if title and snippet:
+                    results.append({
+                        'title': title.get_text(),
+                        'snippet': snippet.get_text(),
+                        'url': title.get('href')
+                    })
+
+            # Format results
+            formatted_results = []
+            for r in results[:10]:  # Limit to top 5 results
+                formatted_results.append(f"[{r['title']}]({r['url']})\n{r['snippet']}\n")
+
+            return "## Search Results\n\n" + "\n".join(formatted_results)
+
+        except requests.RequestException as e:
+            raise Exception(f"Search request failed: {str(e)}")