add web search tool with serper api

Author: denglianghao.dlh

examples/assistant_qwq.py

@@ -19,6 +19,7 @@ def init_agent_service():
 
     tools = [
         'image_gen',
+        # 'web_search',  # Apply for an apikey here (https://serper.dev) and set it as an environment variable by `export SERPER_API_KEY=xxxxxx`
     ]
     bot = Assistant(
         llm=llm_cfg,

qwen_agent/agents/fncall_agent.py

@@ -40,7 +40,6 @@ def __init__(self,
                          description=description)
 
         if not hasattr(self, 'mem'):
-            # Default to use Memory to manage files
             # Default to use Memory to manage files
             if 'qwq' in self.llm.model or 'qvq' in self.llm.model:
                 mem_llm = {

qwen_agent/tools/__init__.py

@@ -10,6 +10,7 @@
 from .simple_doc_parser import SimpleDocParser
 from .storage import Storage
 from .web_extractor import WebExtractor
+from .web_search import WebSearch
 
 __all__ = [
     'BaseTool',
@@ -28,4 +29,5 @@
     'FrontPageSearch',
     'ExtractDocVocabulary',
     'PythonExecutor',
+    'WebSearch',
 ]

qwen_agent/tools/base.py

@@ -150,11 +150,11 @@ def _verify_json_format_args(self, params: Union[str, dict], strict_json: bool =
     @property
     def function(self) -> dict:  # Bad naming. It should be `function_info`.
         return {
-            'name_for_human': self.name_for_human,
+            # 'name_for_human': self.name_for_human,
             'name': self.name,
             'description': self.description,
             'parameters': self.parameters,
-            'args_format': self.args_format
+            # 'args_format': self.args_format
         }
 
     @property

qwen_agent/tools/web_search.py

@@ -0,0 +1,53 @@
+import os
+from typing import Any, List, Union
+
+import requests
+
+from qwen_agent.tools.base import BaseTool, register_tool
+
+SERPER_API_KEY = os.getenv('SERPER_API_KEY', '')
+SERPER_URL = os.getenv('SERPER_URL', 'https://google.serper.dev/search')
+
+
+@register_tool('web_search', allow_overwrite=True)
+class WebSearch(BaseTool):
+    name = 'web_search'
+    description = 'Search for information from the internet.'
+    parameters = {
+        'type': 'object',
+        'properties': {
+            'query': {
+                'type': 'string',
+            }
+        },
+        'required': ['query'],
+    }
+
+    def call(self, params: Union[str, dict], **kwargs) -> str:
+        params = self._verify_json_format_args(params)
+        query = params['query']
+
+        search_results = self.search(query)
+        formatted_results = self._format_results(search_results)
+        return formatted_results
+
+    @staticmethod
+    def search(query: str) -> List[Any]:
+        if not SERPER_API_KEY:
+            raise ValueError(
+                'SERPER_API_KEY is None! Please Apply for an apikey from https://serper.dev and set it as an environment variable by `export SERPER_API_KEY=xxxxxx`'
+            )
+        headers = {'Content-Type': 'application/json', 'X-API-KEY': SERPER_API_KEY}
+        payload = {'q': query}
+        response = requests.post(SERPER_URL, json=payload, headers=headers)
+        response.raise_for_status()
+
+        return response.json()['organic']
+
+    @staticmethod
+    def _format_results(search_results: List[Any]) -> str:
+        content = '```\n{}\n```'.format('\n\n'.join([
+            f"[{i}]\"{doc['title']}\n{doc.get('snippet', '')}\"{doc.get('date', '')}"
+            for i, doc in enumerate(search_results, 1)
+        ]))
+        return content