Upload 4 files

planning/autogen_planner.py

@@ -0,0 +1,150 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from typing import Optional
+import semantic_kernel, autogen
+
+
+class AutoGenPlanner:
+    """(Demo) Semantic Kernel planner using Conversational Programming via AutoGen.
+
+    AutoGenPlanner leverages OpenAI Function Calling and AutoGen agents to solve
+    a task using only the Plugins loaded into Semantic Kernel. SK Plugins are
+    automatically shared with AutoGen, so you only need to load the Plugins in SK
+    with the usual `kernel.import_skill(...)` syntax. You can use native and
+    semantic functions without any additional configuration. Currently the integration
+    is limited to functions with a single string parameter. The planner has been
+    tested with GPT 3.5 Turbo and GPT 4. It always used 3.5 Turbo with OpenAI,
+    just for performance and cost reasons.
+    """
+
+    import datetime
+    from typing import List, Dict
+
+    ASSISTANT_PERSONA = f"""Only use the functions you have been provided with.
+Do not ask the user to perform other actions than executing the functions.
+Use the functions you have to find information not available.
+Today's date is: {datetime.date.today().strftime("%B %d, %Y")}.
+Reply TERMINATE when the task is done.
+"""
+
+    def __init__(self, kernel: semantic_kernel.Kernel, llm_config: Dict = None):
+        """
+        Args:
+            kernel: an instance of Semantic Kernel, with plugins loaded.
+            llm_config: a dictionary with the following keys:
+                - type: "openai" or "azure"
+                - openai_api_key: OpenAI API key
+                - azure_api_key: Azure API key
+                - azure_deployment: Azure deployment name
+                - azure_endpoint: Azure endpoint
+        """
+        super().__init__()
+        self.kernel = kernel
+        self.llm_config = llm_config
+
+    def create_assistant_agent(self, name: str, persona: str = ASSISTANT_PERSONA) -> autogen.AssistantAgent:
+        """
+        Create a new AutoGen Assistant Agent.
+        Args:
+            name (str): the name of the agent
+            persona (str): the LLM system message defining the agent persona,
+                in case you want to customize it.
+        """
+        return autogen.AssistantAgent(name=name, system_message=persona, llm_config=self.__get_autogen_config())
+
+    def create_user_agent(
+        self, name: str, max_auto_reply: Optional[int] = None, human_input: Optional[str] = "ALWAYS"
+    ) -> autogen.UserProxyAgent:
+        """
+        Create a new AutoGen User Proxy Agent.
+        Args:
+            name (str): the name of the agent
+            max_auto_reply (int): the maximum number of consecutive auto replies.
+                default to None (no limit provided).
+            human_input (str): the human input mode. default to "ALWAYS".
+                Possible values are "ALWAYS", "TERMINATE", "NEVER".
+                (1) When "ALWAYS", the agent prompts for human input every time a message is received.
+                    Under this mode, the conversation stops when the human input is "exit",
+                    or when is_termination_msg is True and there is no human input.
+                (2) When "TERMINATE", the agent only prompts for human input only when a termination message is received or
+                    the number of auto reply reaches the max_consecutive_auto_reply.
+                (3) When "NEVER", the agent will never prompt for human input. Under this mode, the conversation stops
+                    when the number of auto reply reaches the max_consecutive_auto_reply or when is_termination_msg is True.
+        """
+        return autogen.UserProxyAgent(
+            name=name,
+            human_input_mode=human_input,
+            max_consecutive_auto_reply=max_auto_reply,
+            function_map=self.__get_function_map(),
+        )
+
+    def __get_autogen_config(self):
+        """
+        Get the AutoGen LLM and Function Calling configuration.
+        """
+        if self.llm_config:
+            if self.llm_config["type"] == "openai":
+                if not self.llm_config["openai_api_key"] or self.llm_config["openai_api_key"] == "sk-...":
+                    raise Exception("OpenAI API key is not set")
+                return {
+                    "functions": self.__get_function_definitions(),
+                    "config_list": [{"model": "gpt-3.5-turbo", "api_key": self.llm_config["openai_api_key"]}],
+                }
+            if self.llm_config["type"] == "azure":
+                if (
+                    not self.llm_config["azure_api_key"]
+                    or not self.llm_config["azure_deployment"]
+                    or not self.llm_config["azure_endpoint"]
+                ):
+                    raise Exception("Azure OpenAI API configuration is incomplete")
+                return {
+                    "functions": self.__get_function_definitions(),
+                    "config_list": [
+                        {
+                            "model": self.llm_config["azure_deployment"],
+                            "api_type": "azure",
+                            "api_key": self.llm_config["azure_api_key"],
+                            "api_base": self.llm_config["azure_endpoint"],
+                            "api_version": "2023-08-01-preview",
+                        }
+                    ],
+                }
+
+        raise Exception("LLM type not provided, must be 'openai' or 'azure'")
+
+    def __get_function_definitions(self) -> List:
+        """
+        Get the list of function definitions for OpenAI Function Calling.
+        """
+        functions = []
+        sk_functions = self.kernel.skills.get_functions_view()
+        for ns in {**sk_functions.native_functions, **sk_functions.semantic_functions}:
+            for f in sk_functions.native_functions[ns]:
+                functions.append(
+                    {
+                        "name": f.name,
+                        "description": f.description,
+                        "parameters": {
+                            "type": "object",
+                            "properties": {
+                                f.parameters[0].name: {
+                                    "description": f.parameters[0].description,
+                                    "type": f.parameters[0].type_,
+                                }
+                            },
+                            "required": [f.parameters[0].name],
+                        },
+                    }
+                )
+        return functions
+
+    def __get_function_map(self) -> Dict:
+        """
+        Get the function map for AutoGen Function Calling.
+        """
+        function_map = {}
+        sk_functions = self.kernel.skills.get_functions_view()
+        for ns in {**sk_functions.native_functions, **sk_functions.semantic_functions}:
+            for f in sk_functions.native_functions[ns]:
+                function_map[f.name] = self.kernel.skills.get_function(f.skill_name, f.name)
+        return function_map

plugins/bing_connector.py

@@ -0,0 +1,112 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import urllib, aiohttp
+from logging import Logger
+from typing import Any, List, Optional
+from semantic_kernel.connectors.search_engine.connector import ConnectorBase
+from semantic_kernel.utils.null_logger import NullLogger
+
+
+class BingConnector(ConnectorBase):
+    """
+    A search engine connector that uses the Bing Search API to perform a web search.
+    The connector can be used to read "answers" from Bing, when "snippets" are available,
+    or simply to retrieve the URLs of the search results.
+    """
+
+    _api_key: str
+
+    def __init__(self, api_key: str, logger: Optional[Logger] = None) -> None:
+        self._api_key = api_key
+        self._logger = logger if logger else NullLogger()
+
+        if not self._api_key:
+            raise ValueError("Bing API key cannot be null. Please set environment variable BING_API_KEY.")
+
+    async def search_url_async(self, query: str, num_results: str, offset: str) -> List[str]:
+        """
+        Returns the search results URLs of the query provided by Bing web search API.
+        Returns `num_results` results and ignores the first `offset`.
+
+        :param query: search query
+        :param num_results: the number of search results to return
+        :param offset: the number of search results to ignore
+        :return: list of search results
+        """
+        data = await self.__search(query, num_results, offset)
+        if data:
+            pages = data["webPages"]["value"]
+            self._logger.info(pages)
+            result = list(map(lambda x: x["url"], pages))
+            self._logger.info(result)
+            return result
+        else:
+            return []
+
+    async def search_snippet_async(self, query: str, num_results: str, offset: str) -> List[str]:
+        """
+        Returns the search results Text Preview (aka snippet) of the query provided by Bing web search API.
+        Returns `num_results` results and ignores the first `offset`.
+
+        :param query: search query
+        :param num_results: the number of search results to return
+        :param offset: the number of search results to ignore
+        :return: list of search results
+        """
+        data = await self.__search(query, num_results, offset)
+        if data:
+            pages = data["webPages"]["value"]
+            self._logger.info(pages)
+            result = list(map(lambda x: x["snippet"], pages))
+            self._logger.info(result)
+            return result
+        else:
+            return []
+
+    async def __search(self, query: str, num_results: str, offset: str) -> Any:
+        """
+        Returns the search response of the query provided by pinging the Bing web search API.
+        Returns the response content
+
+        :param query: search query
+        :param num_results: the number of search results to return
+        :param offset: the number of search results to ignore
+        :return: response content or None
+        """
+        if not query:
+            raise ValueError("query cannot be 'None' or empty.")
+
+        if not num_results:
+            num_results = 1
+        if not offset:
+            offset = 0
+
+        num_results = int(num_results)
+        offset = int(offset)
+
+        if num_results <= 0:
+            raise ValueError("num_results value must be greater than 0.")
+        if num_results >= 50:
+            raise ValueError("num_results value must be less than 50.")
+
+        if offset < 0:
+            raise ValueError("offset must be greater than 0.")
+
+        self._logger.info(
+            f"Received request for bing web search with \
+                params:\nquery: {query}\nnum_results: {num_results}\noffset: {offset}"
+        )
+
+        _base_url = "https://api.bing.microsoft.com/v7.0/search"
+        _request_url = f"{_base_url}?q={urllib.parse.quote_plus(query)}&count={num_results}&offset={offset}"
+
+        self._logger.info(f"Sending GET request to {_request_url}")
+
+        headers = {"Ocp-Apim-Subscription-Key": self._api_key}
+
+        async with aiohttp.ClientSession() as session:
+            async with session.get(_request_url, headers=headers, raise_for_status=True) as response:
+                if response.status == 200:
+                    return await response.json()
+                else:
+                    return None

plugins/sk_bing_plugin.py

@@ -0,0 +1,38 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from semantic_kernel.skill_definition import sk_function
+from plugins.bing_connector import BingConnector
+
+
+class BingPlugin:
+    """
+    A plugin to search Bing.
+    """
+
+    def __init__(self, bing_api_key: str):
+        self.bing = BingConnector(api_key=bing_api_key)
+        if not bing_api_key or bing_api_key == "...":
+            raise Exception("Bing API key is not set")
+
+    @sk_function(
+        description="Use Bing to find a page about a topic. The return is a URL of the page found.",
+        name="find_web_page_about",
+        input_description="Two comma separated values: #1 Offset from the first result (default zero), #2 The topic to search, e.g. '0,who won the F1 title in 2023?'.",
+    )
+    async def find_web_page_about(self, input: str) -> str:
+        """
+        A native function that uses Bing to find a page URL about a topic.
+        To simplify the integration with Autogen, the input parameter is a string with two comma separated
+        values, rather than the usual context dictionary.
+        """
+
+        # Input validation, the error message can help self-correct the input
+        if "," not in input:
+            raise ValueError("The input argument must contain a comma, e.g. '0,who won the F1 title in 2023?'")
+
+        parts = input.split(",", 1)
+        result = await self.bing.search_url_async(query=parts[1], num_results=1, offset=parts[0])
+        if result:
+            return result[0]
+        else:
+            return f"Nothing found, try again or try to adjust the topic."

plugins/sk_web_pages_plugin.py

@@ -0,0 +1,35 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from semantic_kernel.skill_definition import sk_function
+from bs4 import BeautifulSoup
+import re, aiohttp
+
+
+class WebPagesPlugin:
+    """
+    A plugin to interact with web pages, e.g. download the text content of a page.
+    """
+
+    @sk_function(
+        description="Fetch the text content of a webpage. The return is a string containing all the text.",
+        name="fetch_webpage",
+        input_description="URL of the page to fetch.",
+    )
+    async def fetch_webpage(self, input: str) -> str:
+        """
+        A native function that fetches the text content of a webpage.
+        HTML tags are removed, and empty lines are compacted.
+        """
+        if not input:
+            raise ValueError("url cannot be `None` or empty")
+        async with aiohttp.ClientSession() as session:
+            async with session.get(input, raise_for_status=True) as response:
+                html = await response.text()
+                soup = BeautifulSoup(html, features="html.parser")
+                # remove some elements
+                for el in soup(["script", "style", "iframe", "img", "video", "audio"]):
+                    el.extract()
+
+                # get text and compact empty lines
+                text = soup.get_text()
+                return re.sub(r"[\r\n][\r\n]{2,}", "\n\n", text)