Upload 2 files

examples/debate/agent/search/search_engine.py

@@ -0,0 +1,48 @@
+import asyncio
+from enum import Enum
+from typing import List
+
+
+class SearchAPI(Enum):
+    PERPLEXITY = "perplexity"
+    TAVILY = "tavily"
+    EXA = "exa"
+    ARXIV = "arxiv"
+    PUBMED = "pubmed"
+    LINKUP = "linkup"
+
+
+class SearchException(Exception):
+
+    def __init__(self, message, error_code=None):
+        super().__init__()
+        self.message = message
+        self.error_code = error_code
+
+    def __str__(self):
+        return f'{self.message}'
+
+
+class SearchEngine:
+
+    def batch_search(self, queries: List[str], max_results=5, include_raw_content=False, **kwargs):
+        try:
+            return asyncio.run(
+                self.async_batch_search(queries, max_results=max_results, include_raw_content=include_raw_content, **kwargs))
+        except Exception as err:
+            raise SearchException(f"search queries = {queries} failed.")
+
+    async def async_batch_search(self, queries: List[str], max_results=5, include_raw_content=False, **kwargs):
+        search_tasks = []
+        for query in queries:
+            search_tasks.append(
+                self.async_search(query, max_results, include_raw_content, **kwargs)
+            )
+
+        # Execute all searches concurrently
+        search_docs = await asyncio.gather(*search_tasks)
+
+        return search_docs
+
+    async def async_search(self, query: str, max_results=5, include_raw_content=False, **kwargs) -> dict:
+        pass

examples/debate/agent/search/tavily_search_engine.py

@@ -0,0 +1,46 @@
+
+from examples.debate.agent.search.search_engine import SearchEngine
+
+
+class TavilySearchEngine(SearchEngine):
+    """Tavily"""
+    async def async_search(self, query: str, max_results=5, include_raw_content=False, **kwargs) -> dict:
+        """
+            Performs concurrent web searches using the Tavily API.
+
+            Args:
+                query (Str): str
+
+            Returns:
+                    dict:  search responses from Tavily API, one per query. Each response has format:
+                        {
+                            'query': str, # The original search query
+                            'follow_up_questions': None,
+                            'answer': None,
+                            'images': list,
+                            'results': [                     # List of search results
+                                {
+                                    'title': str,            # Title of the webpage
+                                    'url': str,              # URL of the result
+                                    'content': str,          # Summary/snippet of content
+                                    'score': float,          # Relevance score
+                                    'raw_content': str|None  # Full page content if available
+                                },
+                                ...
+                            ]
+                        }
+            """
+        try:
+            from tavily import AsyncTavilyClient
+        except ImportError:
+            # install mistune
+            import subprocess
+            subprocess.run(["pip", "install", "tavily-python>=0.5.1"], check=True)
+            from tavily import AsyncTavilyClient
+        tavily_async_client = AsyncTavilyClient()
+        return await tavily_async_client.search(
+            query,
+            max_results=5,
+            include_raw_content=True,
+            topic="general"
+        )