Merge branch 'feat-qa-with-search' into 'mgx_ops'

QA with web search

See merge request pub/MetaGPT!269

examples/search_enhanced_qa.py

@@ -0,0 +1,27 @@
+"""
+This script demonstrates how to use the SearchEnhancedQA action to answer questions
+by leveraging web search results. It showcases a simple example of querying about
+the current weather in Beijing.
+
+The SearchEnhancedQA action combines web search capabilities with natural language
+processing to provide informative answers to user queries.
+"""
+
+import asyncio
+
+from metagpt.actions.search_enhanced_qa import SearchEnhancedQA
+
+
+async def main():
+    """Runs a sample query through SearchEnhancedQA and prints the result."""
+
+    action = SearchEnhancedQA()
+
+    query = "What is the weather like in Beijing today?"
+    answer = await action.run(query)
+
+    print(f"The answer to '{query}' is:\n\n{answer}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

metagpt/actions/research.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import asyncio
-from typing import Any, Callable, Optional, Union
+from typing import Any, Callable, Coroutine, Optional, Union
 
 from pydantic import TypeAdapter, model_validator
 
@@ -13,6 +13,7 @@ from metagpt.logs import logger
 from metagpt.tools.search_engine import SearchEngine
 from metagpt.tools.web_browser_engine import WebBrowserEngine
 from metagpt.utils.common import OutputParser
+from metagpt.utils.parse_html import WebPage
 from metagpt.utils.text import generate_prompt_chunk, reduce_message_length
 
 LANG_PROMPT = "Please respond in {language}."
@@ -160,7 +161,7 @@ class CollectLinks(Action):
             A list of ranked URLs.
         """
         max_results = max(num_results * 2, 6)
-        results = await self.search_engine.run(query, max_results=max_results, as_string=False)
+        results = await self._search_urls(query, max_results=max_results)
         _results = "\n".join(f"{i}: {j}" for i, j in zip(range(max_results), results))
         prompt = COLLECT_AND_RANKURLS_PROMPT.format(topic=topic, query=query, results=_results)
         logger.debug(prompt)
@@ -176,6 +177,9 @@ class CollectLinks(Action):
             results = self.rank_func(results)
         return [i["link"] for i in results[:num_results]]
 
+    async def _search_urls(self, query: str, max_results: int) -> list[str]:
+        return await self.search_engine.run(query, max_results=max_results, as_string=False)
+
 
 class WebBrowseAndSummarize(Action):
     """Action class to explore the web and provide summaries of articles and webpages."""
@@ -202,6 +206,8 @@ class WebBrowseAndSummarize(Action):
         *urls: str,
         query: str,
         system_text: str = RESEARCH_BASE_SYSTEM,
+        use_concurrent_summarization: bool = False,
+        per_page_timeout: Optional[float] = None,
     ) -> dict[str, str]:
         """Run the action to browse the web and provide summaries.
 
@@ -210,18 +216,41 @@ class WebBrowseAndSummarize(Action):
             urls: Additional URLs to browse.
             query: The research question.
             system_text: The system text.
+            use_concurrent_summarization: Whether to concurrently summarize the content of the webpage by LLM.
+            per_page_timeout: The maximum time for fetching a single page in seconds.
 
         Returns:
             A dictionary containing the URLs as keys and their summaries as values.
         """
-        contents = await self.web_browser_engine.run(url, *urls)
-        if not urls:
-            contents = [contents]
+        contents = await self._fetch_web_contents(url, *urls, per_page_timeout=per_page_timeout)
+
+        all_urls = [url] + list(urls)
+        summarize_tasks = [self._summarize_content(content, query, system_text) for content in contents]
+        summaries = await self._execute_summarize_tasks(summarize_tasks, use_concurrent_summarization)
+        result = {url: summary for url, summary in zip(all_urls, summaries) if summary}
+
+        return result
+
+    async def _fetch_web_contents(
+        self, url: str, *urls: str, per_page_timeout: Optional[float] = None
+    ) -> list[WebPage]:
+        """Fetch web contents from given URLs."""
+
+        contents = await self.web_browser_engine.run(url, *urls, per_page_timeout=per_page_timeout)
+
+        return [contents] if not urls else contents
+
+    async def _summarize_content(self, page: WebPage, query: str, system_text: str) -> str:
+        """Summarize web content."""
+        try:
+            prompt_template = WEB_BROWSE_AND_SUMMARIZE_PROMPT.format(query=query, content="{}")
+
+            content = page.inner_text
+
+            if self._is_content_invalid(content):
+                logger.warning(f"Invalid content detected for URL {page.url}: {content[:10]}...")
+                return None
 
-        summaries = {}
-        prompt_template = WEB_BROWSE_AND_SUMMARIZE_PROMPT.format(query=query, content="{}")
-        for u, content in zip([url, *urls], contents):
-            content = content.inner_text
             chunk_summaries = []
             for prompt in generate_prompt_chunk(content, prompt_template, self.llm.model, system_text, 4096):
                 logger.debug(prompt)
@@ -231,18 +260,33 @@ class WebBrowseAndSummarize(Action):
                 chunk_summaries.append(summary)
 
             if not chunk_summaries:
-                summaries[u] = None
-                continue
+                return None
 
             if len(chunk_summaries) == 1:
-                summaries[u] = chunk_summaries[0]
-                continue
+                return chunk_summaries[0]
 
             content = "\n".join(chunk_summaries)
             prompt = WEB_BROWSE_AND_SUMMARIZE_PROMPT.format(query=query, content=content)
             summary = await self._aask(prompt, [system_text])
-            summaries[u] = summary
-        return summaries
+            return summary
+        except Exception as e:
+            logger.error(f"Error summarizing content: {e}")
+            return None
+
+    def _is_content_invalid(self, content: str) -> bool:
+        """Check if the content is invalid based on specific starting phrases."""
+
+        invalid_starts = ["Fail to load page", "Access Denied"]
+
+        return any(content.strip().startswith(phrase) for phrase in invalid_starts)
+
+    async def _execute_summarize_tasks(self, tasks: list[Coroutine[Any, Any, str]], use_concurrent: bool) -> list[str]:
+        """Execute summarize tasks either concurrently or sequentially."""
+
+        if use_concurrent:
+            return await asyncio.gather(*tasks)
+
+        return [await task for task in tasks]
 
 
 class ConductResearch(Action):

metagpt/actions/search_enhanced_qa.py

@@ -0,0 +1,240 @@
+"""Enhancing question-answering capabilities through search engine augmentation."""
+
+from __future__ import annotations
+
+import json
+
+from pydantic import Field, model_validator
+
+from metagpt.actions import Action
+from metagpt.actions.research import CollectLinks, WebBrowseAndSummarize
+from metagpt.logs import logger
+from metagpt.tools.web_browser_engine import WebBrowserEngine
+from metagpt.utils.common import CodeParser
+
+REWRITE_QUERY_PROMPT = """
+Role: You are a highly efficient assistant that provide a better search query for web search engine to answer the given question.
+
+I will provide you with a question. Your task is to provide a better search query for web search engine.
+
+## Context
+### Question
+{q}
+
+## Format Example
+```json
+{{
+    "query": "the better search query for web search engine.",
+}}
+```
+
+## Instructions
+- Understand the question given by the user.
+- Provide a better search query for web search engine to answer the given question, your answer must be written in the same language as the question.
+- When rewriting, if you are unsure of the specific time, do not include the time.
+
+## Constraint
+Format: Just print the result in json format like **Format Example**.
+
+## Action
+Follow **Instructions**, generate output and make sure it follows the **Constraint**.
+"""
+
+SEARCH_ENHANCED_QA_SYSTEM_PROMPT = """
+You are a large language AI assistant built by MGX. You are given a user question, and please write clean, concise and accurate answer to the question. You will be given a set of related contexts to the question, each starting with a reference number like [[citation:x]], where x is a number. Please use the context.
+
+Your answer must be correct, accurate and written by an expert using an unbiased and professional tone. Please limit to 1024 tokens. Do not give any information that is not related to the question, and do not repeat. Say "information is missing on" followed by the related topic, if the given context do not provide sufficient information. Do not include [citation] in your anwser.
+
+Here are the set of contexts:
+
+{context}
+
+Remember, don't blindly repeat the contexts verbatim. And here is the user question:
+"""
+
+
+class SearchEnhancedQA(Action):
+    """Enhancing question-answering capabilities through search engine augmentation."""
+
+    name: str = "SearchEnhancedQA"
+    desc: str = "Integrating search engine results to anwser the question."
+
+    collect_links_action: CollectLinks = Field(
+        default=CollectLinks(), description="Action to collect relevant links from a search engine."
+    )
+    web_browse_and_summarize_action: WebBrowseAndSummarize = Field(
+        default=None,
+        description="Action to explore the web and provide summaries of articles and webpages.",
+    )
+    per_page_timeout: float = Field(
+        default=10, description="The maximum time for fetching a single page is in seconds. Defaults to 10s."
+    )
+    java_script_enabled: bool = Field(
+        default=False, description="Whether or not to enable JavaScript in the web browser context. Defaults to False."
+    )
+
+    @model_validator(mode="after")
+    def initialize(self):
+        if self.web_browse_and_summarize_action is None:
+            self.web_browser_engine = WebBrowserEngine.from_browser_config(
+                self.config.browser, proxy=self.config.proxy, java_script_enabled=self.java_script_enabled
+            )
+
+            self.web_browse_and_summarize_action = WebBrowseAndSummarize(web_browser_engine=self.web_browser_engine)
+
+        return self
+
+    async def run(self, query: str, rewrite_query: bool = True) -> str:
+        """Answer a query by leveraging web search results.
+
+        Args:
+            query (str): The original user query.
+            rewrite_query (bool): Whether to rewrite the query for better web search results. Defaults to True.
+
+        Returns:
+            str: A detailed answer based on web search results.
+
+        Raises:
+            ValueError: If the query is invalid.
+        """
+
+        self._validate_query(query)
+
+        processed_query = await self._process_query(query, rewrite_query)
+        context = await self._build_context(processed_query)
+
+        return await self._generate_answer(processed_query, context)
+
+    def _validate_query(self, query: str) -> None:
+        """Validate the input query.
+
+        Args:
+            query (str): The query to validate.
+
+        Raises:
+            ValueError: If the query is invalid.
+        """
+
+        if not query.strip():
+            raise ValueError("Query cannot be empty or contain only whitespace.")
+
+    async def _process_query(self, query: str, should_rewrite: bool) -> str:
+        """Process the query, optionally rewriting it."""
+
+        if should_rewrite:
+            return await self._rewrite_query(query)
+
+        return query
+
+    async def _rewrite_query(self, query: str) -> str:
+        """Write a better search query for web search engine.
+
+        If the rewrite process fails, the original query is returned.
+
+        Args:
+            query (str): The original search query.
+
+        Returns:
+            str: The rewritten query if successful, otherwise the original query.
+        """
+
+        prompt = REWRITE_QUERY_PROMPT.format(q=query)
+
+        try:
+            resp = await self._aask(prompt)
+            rewritten_query = self._extract_rewritten_query(resp)
+
+            logger.info(f"Query rewritten: '{query}' -> '{rewritten_query}'")
+            return rewritten_query
+        except Exception as e:
+            logger.warning(f"Query rewrite failed. Returning original query. Error: {e}")
+            return query
+
+    def _extract_rewritten_query(self, response: str) -> str:
+        """Extract the rewritten query from the LLM's JSON response."""
+
+        resp_json = json.loads(CodeParser.parse_code(response, lang="json"))
+        return resp_json["query"]
+
+    async def _build_context(self, query: str) -> str:
+        """Construct a context string from web search citations.
+
+        Args:
+            query (str): The search query.
+
+        Returns:
+            str: Formatted context with numbered citations.
+        """
+
+        citations = await self._search_citations(query)
+        context = "\n\n".join([f"[[citation:{i+1}]] {c}" for i, c in enumerate(citations)])
+
+        return context
+
+    async def _search_citations(self, query: str) -> list[str]:
+        """Perform web search and summarize relevant content.
+
+        Args:
+            query (str): The search query.
+
+        Returns:
+            list[str]: Summaries of relevant web content.
+        """
+
+        relevant_urls = await self._collect_relevant_links(query)
+        if not relevant_urls:
+            logger.warning(f"No relevant URLs found for query: {query}")
+            return []
+
+        logger.info(f"The Relevant links are: {relevant_urls}")
+
+        web_summaries = await self._summarize_web_content(relevant_urls, query)
+        if not web_summaries:
+            logger.warning(f"No summaries generated for query: {query}")
+            return []
+
+        citations = list(web_summaries.values())
+
+        return citations
+
+    async def _collect_relevant_links(self, query: str) -> list[str]:
+        """Search and rank URLs relevant to the query.
+
+        Args:
+            query (str): The search query.
+
+        Returns:
+            list[str]: Ranked list of relevant URLs.
+        """
+
+        return await self.collect_links_action._search_and_rank_urls(topic=query, query=query)
+
+    async def _summarize_web_content(self, urls: list[str], query: str) -> dict[str, str]:
+        """Fetch and summarize content from given URLs.
+
+        Args:
+            urls (list[str]): List of URLs to summarize.
+            query (str): The original query for context.
+
+        Returns:
+            dict[str, str]: Mapping of URLs to their summaries.
+        """
+
+        return await self.web_browse_and_summarize_action.run(
+            *urls, query=query, use_concurrent_summarization=True, per_page_timeout=self.per_page_timeout
+        )
+
+    async def _generate_answer(self, query: str, context: str) -> str:
+        """Generate an answer using the query and context.
+
+        Args:
+            query (str): The user's question.
+            context (str): Relevant information from web search.
+
+        Returns:
+            str: Generated answer based on the context.
+        """
+
+        system_prompt = SEARCH_ENHANCED_QA_SYSTEM_PROMPT.format(context=context)
+
+        return await self._aask(query, [system_prompt])

metagpt/tools/web_browser_engine.py

@@ -92,14 +92,14 @@ class WebBrowserEngine(BaseModel):
         return cls(**data, **kwargs)
 
     @overload
-    async def run(self, url: str) -> WebPage:
+    async def run(self, url: str, per_page_timeout: float = None) -> WebPage:
         ...
 
     @overload
-    async def run(self, url: str, *urls: str) -> list[WebPage]:
+    async def run(self, url: str, *urls: str, per_page_timeout: float = None) -> list[WebPage]:
         ...
 
-    async def run(self, url: str, *urls: str) -> WebPage | list[WebPage]:
+    async def run(self, url: str, *urls: str, per_page_timeout: float = None) -> WebPage | list[WebPage]:
         """Runs the browser engine to load one or more web pages.
 
         This method is the implementation of the overloaded run signatures. It delegates the task
@@ -108,8 +108,9 @@ class WebBrowserEngine(BaseModel):
         Args:
             url: The URL of the first web page to load.
             *urls: Additional URLs of web pages to load, if any.
+            per_page_timeout: The maximum time for fetching a single page in seconds.
 
         Returns:
             A WebPage object if a single URL is provided, or a list of WebPage objects if multiple URLs are provided.
         """
-        return await self.run_func(url, *urls)
+        return await self.run_func(url, *urls, per_page_timeout=per_page_timeout)

metagpt/tools/web_browser_engine_playwright.py

@@ -42,7 +42,10 @@ class PlaywrightWrapper(BaseModel):
         if "ignore_https_errors" in kwargs:
             self.context_kwargs["ignore_https_errors"] = kwargs["ignore_https_errors"]
 
-    async def run(self, url: str, *urls: str) -> WebPage | list[WebPage]:
+        if "java_script_enabled" in kwargs:
+            self.context_kwargs["java_script_enabled"] = kwargs["java_script_enabled"]
+
+    async def run(self, url: str, *urls: str, per_page_timeout: float = None) -> WebPage | list[WebPage]:
         async with async_playwright() as ap:
             browser_type = getattr(ap, self.browser_type)
             await self._run_precheck(browser_type)
@@ -50,11 +53,17 @@ class PlaywrightWrapper(BaseModel):
             _scrape = self._scrape
 
             if urls:
-                return await asyncio.gather(_scrape(browser, url), *(_scrape(browser, i) for i in urls))
-            return await _scrape(browser, url)
+                return await asyncio.gather(
+                    _scrape(browser, url, per_page_timeout), *(_scrape(browser, i, per_page_timeout) for i in urls)
+                )
+            return await _scrape(browser, url, per_page_timeout)
 
-    async def _scrape(self, browser, url):
+    async def _scrape(self, browser, url, timeout: float = None):
         context = await browser.new_context(**self.context_kwargs)
+
+        if timeout is not None:
+            context.set_default_timeout(timeout * 1000)  # playwright uses milliseconds.
+
         page = await context.new_page()
         async with page:
             try:

metagpt/tools/web_browser_engine_selenium.py

@@ -54,14 +54,16 @@ class SeleniumWrapper(BaseModel):
     def executable_path(self):
         return self.launch_kwargs.get("executable_path")
 
-    async def run(self, url: str, *urls: str) -> WebPage | list[WebPage]:
+    async def run(self, url: str, *urls: str, per_page_timeout: float = None) -> WebPage | list[WebPage]:
         await self._run_precheck()
 
-        _scrape = lambda url: self.loop.run_in_executor(self.executor, self._scrape_website, url)
+        _scrape = lambda url, per_page_timeout: self.loop.run_in_executor(
+            self.executor, self._scrape_website, url, per_page_timeout
+        )
 
         if urls:
-            return await asyncio.gather(_scrape(url), *(_scrape(i) for i in urls))
-        return await _scrape(url)
+            return await asyncio.gather(_scrape(url, per_page_timeout), *(_scrape(i, per_page_timeout) for i in urls))
+        return await _scrape(url, per_page_timeout)
 
     async def _run_precheck(self):
         if self._has_run_precheck:
@@ -75,11 +77,11 @@ class SeleniumWrapper(BaseModel):
         )
         self._has_run_precheck = True
 
-    def _scrape_website(self, url):
+    def _scrape_website(self, url, timeout: float = None):
         with self._get_driver() as driver:
             try:
                 driver.get(url)
-                WebDriverWait(driver, 30).until(EC.presence_of_element_located((By.TAG_NAME, "body")))
+                WebDriverWait(driver, timeout or 30).until(EC.presence_of_element_located((By.TAG_NAME, "body")))
                 inner_text = driver.execute_script("return document.body.innerText;")
                 html = driver.page_source
             except Exception as e:

metagpt/utils/common.py

@@ -15,6 +15,7 @@ import ast
 import base64
 import contextlib
 import csv
+import functools
 import importlib
 import inspect
 import json
@@ -23,7 +24,10 @@ import os
 import platform
 import re
 import sys
+import time
 import traceback
+from asyncio import iscoroutinefunction
+from datetime import datetime
 from io import BytesIO
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Literal, Optional, Tuple, Union
@@ -1044,3 +1048,35 @@ def tool2name(cls, methods: List[str], entry) -> Dict[str, Any]:
     if len(mappings) < 2:
         mappings[class_name] = entry
     return mappings
+
+
+def log_time(method):
+    """A time-consuming decorator for printing execution duration."""
+
+    def before_call():
+        start_time, cpu_start_time = time.perf_counter(), time.process_time()
+        logger.info(f"[{method.__name__}] started at: " f"{datetime.now().strftime('%Y-%m-%d %H:%m:%S')}")
+        return start_time, cpu_start_time
+
+    def after_call(start_time, cpu_start_time):
+        end_time, cpu_end_time = time.perf_counter(), time.process_time()
+        logger.info(
+            f"[{method.__name__}] ended. "
+            f"Time elapsed: {end_time - start_time:.4} sec, CPU elapsed: {cpu_end_time - cpu_start_time:.4} sec"
+        )
+
+    @functools.wraps(method)
+    def timeit_wrapper(*args, **kwargs):
+        start_time, cpu_start_time = before_call()
+        result = method(*args, **kwargs)
+        after_call(start_time, cpu_start_time)
+        return result
+
+    @functools.wraps(method)
+    async def timeit_wrapper_async(*args, **kwargs):
+        start_time, cpu_start_time = before_call()
+        result = await method(*args, **kwargs)
+        after_call(start_time, cpu_start_time)
+        return result
+
+    return timeit_wrapper_async if iscoroutinefunction(method) else timeit_wrapper