From 191bce15bdf7ebed45680fa8b77a7b9cf0726d67 Mon Sep 17 00:00:00 2001 From: Kye Date: Sat, 5 Aug 2023 19:18:56 -0400 Subject: [PATCH] workernode documentation Former-commit-id: fbeefa140bb233f2d2fcd72e12487185ee59227a --- DOCS/agents/README.md | 5 +- DOCS/workers/WorkerNode.md | 275 ++++++++++++++++ swarms/utils/embeddings/openai.py | 512 ++++++++++++++++++++++++++++++ 3 files changed, 789 insertions(+), 3 deletions(-) create mode 100644 DOCS/workers/WorkerNode.md create mode 100644 swarms/utils/embeddings/openai.py diff --git a/DOCS/agents/README.md b/DOCS/agents/README.md index f771ca88..ddd3a29f 100644 --- a/DOCS/agents/README.md +++ b/DOCS/agents/README.md @@ -61,7 +61,7 @@ from swarms.agents import vectorstore, tool, Agent # Create an instance of the Agent class agent = Agent( - llm=HuggingFaceLLM, + llm=huggingface, memory=vectorstore, tools=tool, ) @@ -72,5 +72,4 @@ agent.run("Make me an instagram clone") In conclusion, the Agents in Swarms represent a new way of thinking about AI. They are simple, modular, and highly customizable, allowing you to create powerful AI systems that are more than the sum of their parts. And as always, we're just getting started. There's always room for improvement, for simplification, for making things even better. That's the spirit of open collaboration. That's the spirit of Swarms. -Thanks for becoming an alpha build user, email kye@apac.ai with all complaints - +Thanks for becoming an alpha build user, email kye@apac.ai with all complaints. \ No newline at end of file diff --git a/DOCS/workers/WorkerNode.md b/DOCS/workers/WorkerNode.md new file mode 100644 index 00000000..39283840 --- /dev/null +++ b/DOCS/workers/WorkerNode.md @@ -0,0 +1,275 @@ +Swarms Documentation + +==================== + +Worker Node + +----------- + +The `WorkerNode` class is a powerful component of the Swarms framework. It is designed to spawn an autonomous agent instance as a worker to accomplish complex tasks. It can search the internet, spawn child multi-modality models to process and generate images, text, audio, and so on. + +### WorkerNodeInitializer + +The `WorkerNodeInitializer` class is used to initialize a worker node. + +#### Initialization + +``` + +WorkerNodeInitializer(openai_api_key: str, + +llm: Optional[Union[InMemoryDocstore, ChatOpenAI]] = None, + +tools: Optional[List[Tool]] = None, + +worker_name: Optional[str] = "Swarm Worker AI Assistant", + +worker_role: Optional[str] = "Assistant", + +human_in_the_loop: Optional[bool] = False, + +search_kwargs: dict = {}, + +verbose: Optional[bool] = False, + +chat_history_file: str = "chat_history.txt") + +``` + +Copy code + +##### Parameters + +- `openai_api_key` (str): The OpenAI API key. + +- `llm` (Union[InMemoryDocstore, ChatOpenAI], optional): The language model to use. Default is `ChatOpenAI`. + +- `tools` (List[Tool], optional): The tools to use. + +- `worker_name` (str, optional): The name of the worker. Default is "Swarm Worker AI Assistant". + +- `worker_role` (str, optional): The role of the worker. Default is "Assistant". + +- `human_in_the_loop` (bool, optional): Whether to include a human in the loop. Default is False. + +- `search_kwargs` (dict, optional): The keyword arguments for the search. + +- `verbose` (bool, optional): Whether to print verbose output. Default is False. + +- `chat_history_file` (str, optional): The file to store the chat history. Default is "chat_history.txt". + +##### Example + +``` + +from swarms.agents.tools.autogpt import DuckDuckGoSearchRun + +worker_node_initializer = WorkerNodeInitializer(openai_api_key="your_openai_api_key", + +tools=[DuckDuckGoSearchRun()], + +worker_name="My Worker", + +worker_role="Assistant", + +human_in_the_loop=True) + +``` + +Copy code + +### WorkerNode + +The `WorkerNode` class is used to create a worker node. + +#### Initialization + +``` + +WorkerNode(openai_api_key: str, + +temperature: int, + +llm: Optional[Union[InMemoryDocstore, ChatOpenAI]] = None, + +tools: Optional[List[Tool]] = None, + +worker_name: Optional[str] = "Swarm Worker AI Assistant", + +worker_role: Optional[str] = "Assistant", + +human_in_the_loop: Optional[bool] = False, + +search_kwargs: dict = {}, + +verbose: Optional[bool] = False, + +chat_history_file: str = "chat_history.txt") + +``` + +Copy code + +##### Parameters + +- `openai_api_key` (str): The OpenAI API key. + +- `temperature` (int): The temperature for the language model. + +- `llm` (Union[InMemoryDocstore, ChatOpenAI], optional): The language model to use. Default is `ChatOpenAI`. + +- `tools` (List[Tool], optional): The tools to use. + +- `worker_name` (str, optional): The name of the worker. Default is "Swarm Worker AI Assistant". + +- `worker_role` (str, optional): The role of the worker. Default is "Assistant". + +- `human_in_the_loop` (bool, optional): Whether to include a human in the loop. Default is False. + +- `search_kwargs` (dict, optional): The keyword arguments for the search. + +- `verbose` (bool, optional): Whether to print verbose output. Default is False. + +- `chat_history_file` (str, optional): The file to store the chat history. Default is "chat_history.txt". + +##### Example + +``` + +worker_node = WorkerNode(openai_api_key="your_openai_api_key", + +temperature=0.8, + +tools=[DuckDuckGoSearchRun()], + +worker_name="My Worker", + +worker_role="As``` + +tools=[DuckDuckGoSearchRun()], + +worker_name="My Worker", + +worker_role="Assistant", + +human_in_the_loop=True) + +# Create a worker node + +worker_node = WorkerNode(openai_api_key="your_openai_api_key", + +temperature=0.8, + +tools=[DuckDuckGoSearchRun()], + +worker_name="My Worker", + +worker_role="Assistant", + +human_in_the_loop=True) + +# Add a tool to the worker node + +worker_node_initializer.add_tool(DuckDuckGoSearchRun()) + +# Initialize the language model and tools for the worker node + +worker_node.initialize_llm(ChatOpenAI, temperature=0.8) + +worker_node.initialize_tools(ChatOpenAI) + +# Create the worker node + +worker_node.create_worker_node(worker_name="My Worker Node", + +worker_role="Assistant", + +human_in_the_loop=True, + +llm_class=ChatOpenAI, + +search_kwargs={}) + +# Run the worker node + +`worker_node.run("Hello, world!")` + +In this example, we first initialize a `WorkerNodeInitializer` and a `WorkerNode`. We then add a tool to the `WorkerNodeInitializer` and initialize the language model and tools for the `WorkerNode`. Finally, we create the worker node and run it with a given prompt. + +This example shows how you can use the `WorkerNode` and `WorkerNodeInitializer` classes to create a worker node, add tools to it, initialize its language model and tools, and run it with a given prompt. The parameters of these classes can be customized to suit your specific needs. + +Thanks for becoming an alpha build user, email kye@apac.ai with all complaintssistant", + +human_in_the_loop=True) + +``` + +Copy code + +### Full Example + +Here is a full example of how to use the `WorkerNode` and `WorkerNodeInitializer` classes: + +```python + +from swarms.agents.tools.autogpt import DuckDuckGoSearchRun + +from swarms.worker_node import WorkerNode, WorkerNodeInitializer + +# Initialize a worker node + +worker_node_initializer = WorkerNodeInitializer(openai_api_key="your_openai_api_key", + +tools=[DuckDuckGoSearchRun()], + +worker_name="My Worker", + +worker_role="Assistant", + +human_in_the_loop=True) + +# Create a worker node + +worker_node = WorkerNode(openai_api_key="your_openai_api_key", + +temperature=0.8, + +tools=[DuckDuckGoSearchRun()], + +worker_name="My Worker", + +worker_role="Assistant", + +human_in_the_loop=True) + +# Add a tool to the worker node + +worker_node_initializer.add_tool(DuckDuckGoSearchRun()) + +# Initialize the language model and tools for the worker node + +worker_node.initialize_llm(ChatOpenAI, temperature=0.8) + +worker_node.initialize_tools(ChatOpenAI) + +# Create the worker node + +worker_node.create_worker_node(worker_name="My Worker Node", + +worker_role="Assistant", + +human_in_the_loop=True, + +llm_class=ChatOpenAI, + +search_kwargs={}) + +# Run the worker node + +worker_node.run("Hello, world!") + +``` + +In this example, we first initialize a `WorkerNodeInitializer` and a `WorkerNode`. We then add a tool to the `WorkerNodeInitializer` and initialize the language model and tools for the `WorkerNode`. Finally, we create the worker node and run it with a given prompt. + +This example shows how you can use the `WorkerNode` and `WorkerNodeInitializer` classes to create a worker node, add tools to it, initialize its language model and tools, and run it with a given prompt. The parameters of these classes can be customized to suit your specific needs. \ No newline at end of file diff --git a/swarms/utils/embeddings/openai.py b/swarms/utils/embeddings/openai.py new file mode 100644 index 00000000..5d232f1f --- /dev/null +++ b/swarms/utils/embeddings/openai.py @@ -0,0 +1,512 @@ +from __future__ import annotations + +import logging +import warnings +from typing import ( + Any, + Callable, + Dict, + List, + Literal, + Optional, + Sequence, + Set, + Tuple, + Union, +) + +import numpy as np +from pydantic import BaseModel, Extra, Field, root_validator +from tenacity import ( + AsyncRetrying, + before_sleep_log, + retry, + retry_if_exception_type, + stop_after_attempt, + wait_exponential, +) + +from langchain.embeddings.base import Embeddings +from langchain.utils import get_from_dict_or_env, get_pydantic_field_names + +logger = logging.getLogger(__name__) + + +def _create_retry_decorator(embeddings: OpenAIEmbeddings) -> Callable[[Any], Any]: + import openai + + min_seconds = 4 + max_seconds = 10 + # Wait 2^x * 1 second between each retry starting with + # 4 seconds, then up to 10 seconds, then 10 seconds afterwards + return retry( + reraise=True, + stop=stop_after_attempt(embeddings.max_retries), + wait=wait_exponential(multiplier=1, min=min_seconds, max=max_seconds), + retry=( + retry_if_exception_type(openai.error.Timeout) + | retry_if_exception_type(openai.error.APIError) + | retry_if_exception_type(openai.error.APIConnectionError) + | retry_if_exception_type(openai.error.RateLimitError) + | retry_if_exception_type(openai.error.ServiceUnavailableError) + ), + before_sleep=before_sleep_log(logger, logging.WARNING), + ) + + +def _async_retry_decorator(embeddings: OpenAIEmbeddings) -> Any: + import openai + + min_seconds = 4 + max_seconds = 10 + # Wait 2^x * 1 second between each retry starting with + # 4 seconds, then up to 10 seconds, then 10 seconds afterwards + async_retrying = AsyncRetrying( + reraise=True, + stop=stop_after_attempt(embeddings.max_retries), + wait=wait_exponential(multiplier=1, min=min_seconds, max=max_seconds), + retry=( + retry_if_exception_type(openai.error.Timeout) + | retry_if_exception_type(openai.error.APIError) + | retry_if_exception_type(openai.error.APIConnectionError) + | retry_if_exception_type(openai.error.RateLimitError) + | retry_if_exception_type(openai.error.ServiceUnavailableError) + ), + before_sleep=before_sleep_log(logger, logging.WARNING), + ) + + def wrap(func: Callable) -> Callable: + async def wrapped_f(*args: Any, **kwargs: Any) -> Callable: + async for _ in async_retrying: + return await func(*args, **kwargs) + raise AssertionError("this is unreachable") + + return wrapped_f + + return wrap + + +# https://stackoverflow.com/questions/76469415/getting-embeddings-of-length-1-from-langchain-openaiembeddings +def _check_response(response: dict) -> dict: + if any(len(d["embedding"]) == 1 for d in response["data"]): + import openai + + raise openai.error.APIError("OpenAI API returned an empty embedding") + return response + + +def embed_with_retry(embeddings: OpenAIEmbeddings, **kwargs: Any) -> Any: + """Use tenacity to retry the embedding call.""" + retry_decorator = _create_retry_decorator(embeddings) + + @retry_decorator + def _embed_with_retry(**kwargs: Any) -> Any: + response = embeddings.client.create(**kwargs) + return _check_response(response) + + return _embed_with_retry(**kwargs) + + +async def async_embed_with_retry(embeddings: OpenAIEmbeddings, **kwargs: Any) -> Any: + """Use tenacity to retry the embedding call.""" + + @_async_retry_decorator(embeddings) + async def _async_embed_with_retry(**kwargs: Any) -> Any: + response = await embeddings.client.acreate(**kwargs) + return _check_response(response) + + return await _async_embed_with_retry(**kwargs) + + +class OpenAIEmbeddings(BaseModel, Embeddings): + """OpenAI embedding models. + + To use, you should have the ``openai`` python package installed, and the + environment variable ``OPENAI_API_KEY`` set with your API key or pass it + as a named parameter to the constructor. + + Example: + .. code-block:: python + + from langchain.embeddings import OpenAIEmbeddings + openai = OpenAIEmbeddings(openai_api_key="my-api-key") + + In order to use the library with Microsoft Azure endpoints, you need to set + the OPENAI_API_TYPE, OPENAI_API_BASE, OPENAI_API_KEY and OPENAI_API_VERSION. + The OPENAI_API_TYPE must be set to 'azure' and the others correspond to + the properties of your endpoint. + In addition, the deployment name must be passed as the model parameter. + + Example: + .. code-block:: python + + import os + os.environ["OPENAI_API_TYPE"] = "azure" + os.environ["OPENAI_API_BASE"] = "https:// Dict[str, Any]: + """Build extra kwargs from additional params that were passed in.""" + all_required_field_names = get_pydantic_field_names(cls) + extra = values.get("model_kwargs", {}) + for field_name in list(values): + if field_name in extra: + raise ValueError(f"Found {field_name} supplied twice.") + if field_name not in all_required_field_names: + warnings.warn( + f"""WARNING! {field_name} is not default parameter. + {field_name} was transferred to model_kwargs. + Please confirm that {field_name} is what you intended.""" + ) + extra[field_name] = values.pop(field_name) + + invalid_model_kwargs = all_required_field_names.intersection(extra.keys()) + if invalid_model_kwargs: + raise ValueError( + f"Parameters {invalid_model_kwargs} should be specified explicitly. " + f"Instead they were passed in as part of `model_kwargs` parameter." + ) + + values["model_kwargs"] = extra + return values + + @root_validator() + def validate_environment(cls, values: Dict) -> Dict: + """Validate that api key and python package exists in environment.""" + values["openai_api_key"] = get_from_dict_or_env( + values, "openai_api_key", "OPENAI_API_KEY" + ) + values["openai_api_base"] = get_from_dict_or_env( + values, + "openai_api_base", + "OPENAI_API_BASE", + default="", + ) + values["openai_api_type"] = get_from_dict_or_env( + values, + "openai_api_type", + "OPENAI_API_TYPE", + default="", + ) + values["openai_proxy"] = get_from_dict_or_env( + values, + "openai_proxy", + "OPENAI_PROXY", + default="", + ) + if values["openai_api_type"] in ("azure", "azure_ad", "azuread"): + default_api_version = "2022-12-01" + else: + default_api_version = "" + values["openai_api_version"] = get_from_dict_or_env( + values, + "openai_api_version", + "OPENAI_API_VERSION", + default=default_api_version, + ) + values["openai_organization"] = get_from_dict_or_env( + values, + "openai_organization", + "OPENAI_ORGANIZATION", + default="", + ) + try: + import openai + + values["client"] = openai.Embedding + except ImportError: + raise ImportError( + "Could not import openai python package. " + "Please install it with `pip install openai`." + ) + return values + + @property + def _invocation_params(self) -> Dict: + openai_args = { + "model": self.model, + "request_timeout": self.request_timeout, + "headers": self.headers, + "api_key": self.openai_api_key, + "organization": self.openai_organization, + "api_base": self.openai_api_base, + "api_type": self.openai_api_type, + "api_version": self.openai_api_version, + **self.model_kwargs, + } + if self.openai_api_type in ("azure", "azure_ad", "azuread"): + openai_args["engine"] = self.deployment + if self.openai_proxy: + import openai + + openai.proxy = { + "http": self.openai_proxy, + "https": self.openai_proxy, + } # type: ignore[assignment] # noqa: E501 + return openai_args + + # please refer to + # https://github.com/openai/openai-cookbook/blob/main/examples/Embedding_long_inputs.ipynb + def _get_len_safe_embeddings( + self, texts: List[str], *, engine: str, chunk_size: Optional[int] = None + ) -> List[List[float]]: + embeddings: List[List[float]] = [[] for _ in range(len(texts))] + try: + import tiktoken + except ImportError: + raise ImportError( + "Could not import tiktoken python package. " + "This is needed in order to for OpenAIEmbeddings. " + "Please install it with `pip install tiktoken`." + ) + + tokens = [] + indices = [] + model_name = self.tiktoken_model_name or self.model + try: + encoding = tiktoken.encoding_for_model(model_name) + except KeyError: + logger.warning("Warning: model not found. Using cl100k_base encoding.") + model = "cl100k_base" + encoding = tiktoken.get_encoding(model) + for i, text in enumerate(texts): + if self.model.endswith("001"): + # See: https://github.com/openai/openai-python/issues/418#issuecomment-1525939500 + # replace newlines, which can negatively affect performance. + text = text.replace("\n", " ") + token = encoding.encode( + text, + allowed_special=self.allowed_special, + disallowed_special=self.disallowed_special, + ) + for j in range(0, len(token), self.embedding_ctx_length): + tokens.append(token[j : j + self.embedding_ctx_length]) + indices.append(i) + + batched_embeddings: List[List[float]] = [] + _chunk_size = chunk_size or self.chunk_size + + if self.show_progress_bar: + try: + import tqdm + + _iter = tqdm.tqdm(range(0, len(tokens), _chunk_size)) + except ImportError: + _iter = range(0, len(tokens), _chunk_size) + else: + _iter = range(0, len(tokens), _chunk_size) + + for i in _iter: + response = embed_with_retry( + self, + input=tokens[i : i + _chunk_size], + **self._invocation_params, + ) + batched_embeddings.extend(r["embedding"] for r in response["data"]) + + results: List[List[List[float]]] = [[] for _ in range(len(texts))] + num_tokens_in_batch: List[List[int]] = [[] for _ in range(len(texts))] + for i in range(len(indices)): + results[indices[i]].append(batched_embeddings[i]) + num_tokens_in_batch[indices[i]].append(len(tokens[i])) + + for i in range(len(texts)): + _result = results[i] + if len(_result) == 0: + average = embed_with_retry( + self, + input="", + **self._invocation_params, + )[ + "data" + ][0]["embedding"] + else: + average = np.average(_result, axis=0, weights=num_tokens_in_batch[i]) + embeddings[i] = (average / np.linalg.norm(average)).tolist() + + return embeddings + + # please refer to + # https://github.com/openai/openai-cookbook/blob/main/examples/Embedding_long_inputs.ipynb + async def _aget_len_safe_embeddings( + self, texts: List[str], *, engine: str, chunk_size: Optional[int] = None + ) -> List[List[float]]: + embeddings: List[List[float]] = [[] for _ in range(len(texts))] + try: + import tiktoken + except ImportError: + raise ImportError( + "Could not import tiktoken python package. " + "This is needed in order to for OpenAIEmbeddings. " + "Please install it with `pip install tiktoken`." + ) + + tokens = [] + indices = [] + model_name = self.tiktoken_model_name or self.model + try: + encoding = tiktoken.encoding_for_model(model_name) + except KeyError: + logger.warning("Warning: model not found. Using cl100k_base encoding.") + model = "cl100k_base" + encoding = tiktoken.get_encoding(model) + for i, text in enumerate(texts): + if self.model.endswith("001"): + # See: https://github.com/openai/openai-python/issues/418#issuecomment-1525939500 + # replace newlines, which can negatively affect performance. + text = text.replace("\n", " ") + token = encoding.encode( + text, + allowed_special=self.allowed_special, + disallowed_special=self.disallowed_special, + ) + for j in range(0, len(token), self.embedding_ctx_length): + tokens.append(token[j : j + self.embedding_ctx_length]) + indices.append(i) + + batched_embeddings: List[List[float]] = [] + _chunk_size = chunk_size or self.chunk_size + for i in range(0, len(tokens), _chunk_size): + response = await async_embed_with_retry( + self, + input=tokens[i : i + _chunk_size], + **self._invocation_params, + ) + batched_embeddings.extend(r["embedding"] for r in response["data"]) + + results: List[List[List[float]]] = [[] for _ in range(len(texts))] + num_tokens_in_batch: List[List[int]] = [[] for _ in range(len(texts))] + for i in range(len(indices)): + results[indices[i]].append(batched_embeddings[i]) + num_tokens_in_batch[indices[i]].append(len(tokens[i])) + + for i in range(len(texts)): + _result = results[i] + if len(_result) == 0: + average = ( + await async_embed_with_retry( + self, + input="", + **self._invocation_params, + ) + )["data"][0]["embedding"] + else: + average = np.average(_result, axis=0, weights=num_tokens_in_batch[i]) + embeddings[i] = (average / np.linalg.norm(average)).tolist() + + return embeddings + + def embed_documents( + self, texts: List[str], chunk_size: Optional[int] = 0 + ) -> List[List[float]]: + """Call out to OpenAI's embedding endpoint for embedding search docs. + + Args: + texts: The list of texts to embed. + chunk_size: The chunk size of embeddings. If None, will use the chunk size + specified by the class. + + Returns: + List of embeddings, one for each text. + """ + # NOTE: to keep things simple, we assume the list may contain texts longer + # than the maximum context and use length-safe embedding function. + return self._get_len_safe_embeddings(texts, engine=self.deployment) + + async def aembed_documents( + self, texts: List[str], chunk_size: Optional[int] = 0 + ) -> List[List[float]]: + """Call out to OpenAI's embedding endpoint async for embedding search docs. + + Args: + texts: The list of texts to embed. + chunk_size: The chunk size of embeddings. If None, will use the chunk size + specified by the class. + + Returns: + List of embeddings, one for each text. + """ + # NOTE: to keep things simple, we assume the list may contain texts longer + # than the maximum context and use length-safe embedding function. + return await self._aget_len_safe_embeddings(texts, engine=self.deployment) + + def embed_query(self, text: str) -> List[float]: + """Call out to OpenAI's embedding endpoint for embedding query text. + + Args: + text: The text to embed. + + Returns: + Embedding for the text. + """ + return self.embed_documents([text])[0] + + async def aembed_query(self, text: str) -> List[float]: + """Call out to OpenAI's embedding endpoint async for embedding query text. + + Args: + text: The text to embed. + + Returns: + Embedding for the text. + """ + embeddings = await self.aembed_documents([text]) + return embeddings[0] \ No newline at end of file