Merge pull request #50 from LeonZh0u/leon_dev

[6.i Support SERPER api] Integrate Serper API intop searchandsummarize Action

config/config.yaml

@@ -26,6 +26,8 @@ RPM: 10
 #GOOGLE_API_KEY: "YOUR_API_KEY"
 ## Visit https://programmablesearchengine.google.com/controlpanel/create to get id.
 #GOOGLE_CSE_ID: "YOUR_CSE_ID"
+## Visit https://serper.dev/ to get key.
+#SERPER_API_KEY: "YOUR_API_KEY"
 
 #### for TTS
 

examples/search_with_specific_engine.py

@@ -0,0 +1,15 @@
+import asyncio
+from metagpt.config import Config
+from metagpt.roles import Searcher
+from metagpt.tools import SearchEngineType
+
+async def main():
+    # Serper API
+    await Searcher(engine = SearchEngineType.SERPER_GOOGLE).run("What are some good sun protection products?")
+    # Serper API
+    #await Searcher(engine = SearchEngineType.SERPAPI_GOOGLE).run("What are the best ski brands for skiers?")
+    # Google API
+    #await Searcher(engine = SearchEngineType.DIRECT_GOOGLE).run("What are the most interesting human facts?")
+
+if __name__ == '__main__':
+    asyncio.run(main())

metagpt/actions/search_and_summarize.py

@@ -110,10 +110,14 @@ class SearchAndSummarize(Action):
         super().__init__(name, context, llm)
 
     async def run(self, context: list[Message], system_text=SEARCH_AND_SUMMARIZE_SYSTEM) -> str:
-        if not self.config.serpapi_api_key or 'YOUR_API_KEY' == self.config.serpapi_api_key:
-            logger.warning('Configure SERPAPI_API_KEY to unlock full feature')
+        no_serpapi = not self.config.serpapi_api_key or 'YOUR_API_KEY' == self.config.serpapi_api_key
+        no_serper = not self.config.serper_api_key or 'YOUR_API_KEY' == self.config.serper_api_key
+        no_google= not self.config.google_api_key or 'YOUR_API_KEY' == self.config.google_api_key
+        
+        if no_serpapi and no_google and no_serper:
+            logger.warning('Configure one of SERPAPI_API_KEY, SERPER_API_KEY, GOOGLE_API_KEY to unlock full feature')
             return ""
-
+        
         query = context[-1].content
         # logger.debug(query)
         rsp = await self.search_engine.run(query)

metagpt/config.py

@@ -55,6 +55,7 @@ class Config(metaclass=Singleton):
         self.deployment_id = self._get('DEPLOYMENT_ID')
 
         self.serpapi_api_key = self._get('SERPAPI_API_KEY')
+        self.serper_api_key = self._get('SERPER_API_KEY')
         self.google_api_key = self._get('GOOGLE_API_KEY')
         self.google_cse_id = self._get('GOOGLE_CSE_ID')
         self.search_engine = self._get('SEARCH_ENGINE', SearchEngineType.SERPAPI_GOOGLE)

metagpt/roles/seacher.py

@@ -5,17 +5,33 @@
 @Author  : alexanderwu
 @File    : seacher.py
 """
-from metagpt.roles import Role
-from metagpt.actions import SearchAndSummarize
-from metagpt.tools import SearchEngineType
+from metagpt.logs import logger
 
+from metagpt.roles import Role
+from metagpt.actions import SearchAndSummarize, ActionOutput
+from metagpt.tools import SearchEngineType
+from metagpt.schema import Message
 
 class Searcher(Role):
     def __init__(self, name='Alice', profile='Smart Assistant', goal='Provide search services for users',
-                 constraints='Answer is rich and complete', **kwargs):
+                 constraints='Answer is rich and complete', engine=SearchEngineType.SERPAPI_GOOGLE, **kwargs):
         super().__init__(name, profile, goal, constraints, **kwargs)
-        self._init_actions([SearchAndSummarize])
+        self._init_actions([SearchAndSummarize(engine = engine)])
 
     def set_search_func(self, search_func):
         action = SearchAndSummarize("", engine=SearchEngineType.CUSTOM_ENGINE, search_func=search_func)
         self._init_actions([action])
+
+    async def _act_sp(self) -> Message:
+        logger.info(f"{self._setting}: ready to {self._rc.todo}")
+        response = await self._rc.todo.run(self._rc.memory.get(k=0))
+        # logger.info(response)
+        if isinstance(response, ActionOutput):
+            msg = Message(content=response.content, instruct_content=response.instruct_content,
+                            role=self.profile, cause_by=type(self._rc.todo))
+        else:
+            msg = Message(content=response, role=self.profile, cause_by=type(self._rc.todo))
+        self._rc.memory.add(msg)
+
+    async def _act(self) -> Message:
+        return await self._act_sp()

metagpt/tools/__init__.py

@@ -13,4 +13,5 @@ from enum import Enum, auto
 class SearchEngineType(Enum):
     SERPAPI_GOOGLE = auto()
     DIRECT_GOOGLE = auto()
+    SERPER_GOOGLE = auto()
     CUSTOM_ENGINE = auto()

metagpt/tools/search_engine.py

@@ -14,6 +14,7 @@ from duckduckgo_search import ddg
 
 from metagpt.config import Config
 from metagpt.tools.search_engine_serpapi import SerpAPIWrapper
+from metagpt.tools.search_engine_serper import SerperWrapper
 
 config = Config()
 from metagpt.tools import SearchEngineType
@@ -44,6 +45,12 @@ class SearchEngine:
             rsp = await api.run(query)
         elif self.engine == SearchEngineType.DIRECT_GOOGLE:
             rsp = SearchEngine.run_google(query, max_results)
+        elif self.engine == SearchEngineType.SERPER_GOOGLE:
+            api = SerperWrapper()
+            if isinstance(query, list):
+                rsp = await api.run(query)
+            elif isinstance(query, str):
+                rsp = await api.run([query])
         elif self.engine == SearchEngineType.CUSTOM_ENGINE:
             rsp = self.run_func(query)
         else:

metagpt/tools/search_engine_serper.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+@Time    : 2023/5/23 18:27
+@Author  : alexanderwu
+@File    : search_engine_serpapi.py
+"""
+from typing import Any, Dict, Optional, Tuple
+from metagpt.logs import logger
+import aiohttp
+import json
+from pydantic import BaseModel, Field
+
+from metagpt.config import Config
+
+
+class SerperWrapper(BaseModel):
+    """Wrapper around SerpAPI.
+
+    To use, you should have the ``google-search-results`` python package installed,
+    and the environment variable ``SERPAPI_API_KEY`` set with your API key, or pass
+    `serpapi_api_key` as a named parameter to the constructor.
+    """
+
+    search_engine: Any  #: :meta private:
+    payload: dict = Field(
+        default={
+            "page": 1,
+            "num": 10
+        }
+    )
+    config = Config()
+    serper_api_key: Optional[str] = config.serper_api_key
+    aiosession: Optional[aiohttp.ClientSession] = None
+
+    class Config:
+        arbitrary_types_allowed = True
+
+    async def run(self, query: str, **kwargs: Any) -> str:
+        """Run query through Serper and parse result async."""
+        return ";".join([self._process_response(res) for res in await self.results(query)])
+
+    async def results(self, queries: list[str]) -> dict:
+        """Use aiohttp to run query through Serper and return the results async."""
+
+        def construct_url_and_payload_and_headers() -> Tuple[str, Dict[str, str]]:
+            payloads = self.get_payloads(queries)
+            url = "https://google.serper.dev/search"
+            headers = self.get_headers()
+            return url, payloads, headers
+
+        url, payloads, headers = construct_url_and_payload_and_headers()
+        if not self.aiosession:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(url, data=payloads, headers=headers) as response:
+                    res = await response.json()
+                    
+        else:
+            async with self.aiosession.get.post(url, data=payloads, headers=headers) as response:
+                res = await response.json()
+
+        return res
+
+    def get_payloads(self, queries: list[str]) -> Dict[str, str]:
+        """Get payloads for Serper."""
+        payloads = []
+        for query in queries:
+            _payload = {
+                "q": query,
+            }
+            payloads.append({**self.payload, **_payload})
+        return json.dumps(payloads, sort_keys=True)
+
+    def get_headers(self) -> Dict[str, str]:
+        headers = {
+            'X-API-KEY':  self.serper_api_key,
+            'Content-Type': 'application/json'
+        }
+        return headers
+
+    @staticmethod
+    def _process_response(res: dict) -> str:
+        """Process response from SerpAPI."""
+        # logger.debug(res)
+        focus = ['title', 'snippet', 'link']
+        def get_focused(x): return {i: j for i, j in x.items() if i in focus}
+
+        if "error" in res.keys():
+            raise ValueError(f"Got error from SerpAPI: {res['error']}")
+        if "answer_box" in res.keys() and "answer" in res["answer_box"].keys():
+            toret = res["answer_box"]["answer"]
+        elif "answer_box" in res.keys() and "snippet" in res["answer_box"].keys():
+            toret = res["answer_box"]["snippet"]
+        elif (
+            "answer_box" in res.keys()
+            and "snippet_highlighted_words" in res["answer_box"].keys()
+        ):
+            toret = res["answer_box"]["snippet_highlighted_words"][0]
+        elif (
+            "sports_results" in res.keys()
+            and "game_spotlight" in res["sports_results"].keys()
+        ):
+            toret = res["sports_results"]["game_spotlight"]
+        elif (
+            "knowledge_graph" in res.keys()
+            and "description" in res["knowledge_graph"].keys()
+        ):
+            toret = res["knowledge_graph"]["description"]
+        elif "snippet" in res["organic"][0].keys():
+            toret = res["organic"][0]["snippet"]
+        else:
+            toret = "No good search result found"
+
+        toret_l = []
+        if "answer_box" in res.keys() and "snippet" in res["answer_box"].keys():
+            toret_l += [get_focused(res["answer_box"])]
+        if res.get("organic"):
+            toret_l += [get_focused(i) for i in res.get("organic")]
+
+        return str(toret) + '\n' + str(toret_l)