anthropic

swarms/agents/base.py

@@ -1,24 +1,22 @@
 from __future__ import annotations
 
 from typing import List, Optional
+from pydantic import ValidationError
+
+
+from swarms.agents.utils.Agent import AgentOutputParser
+from swarms.agents.utils.human_input import HumanInputRun
+from swarms.agents.prompts.prompt_generator import FINISH_NAME
+
 
 from langchain.chains.llm import LLMChain
 from langchain.chat_models.base import BaseChatModel
 from langchain.memory import ChatMessageHistory
-from langchain.schema import (
-    BaseChatMessageHistory,
-    Document,
-)
+
+from langchain.schema import (BaseChatMessageHistory, Document,)
 from langchain.schema.messages import AIMessage, HumanMessage, SystemMessage
 from langchain.tools.base import BaseTool
-from langchain.tools.human.tool import HumanInputRun
 from langchain.vectorstores.base import VectorStoreRetriever
-from langchain_experimental.autonomous_agents.autogpt.prompt_generator import (
-    FINISH_NAME,
-)
-from pydantic import ValidationError
-
-from swarms.agents.utils.Agent import AgentOutputParser
 
 
 class Agent:

swarms/agents/models/anthropic.py

@@ -0,0 +1,43 @@
+import requests
+import os
+
+class Anthropic:
+    """Anthropic large language models."""
+
+    def __init__(self, model="claude-2", max_tokens_to_sample=256, temperature=None, top_k=None, top_p=None, streaming=False, default_request_timeout=None):
+        self.model = model
+        self.max_tokens_to_sample = max_tokens_to_sample
+        self.temperature = temperature
+        self.top_k = top_k
+        self.top_p = top_p
+        self.streaming = streaming
+        self.default_request_timeout = default_request_timeout or 600
+        self.anthropic_api_url = os.getenv("ANTHROPIC_API_URL", "https://api.anthropic.com")
+        self.anthropic_api_key = os.getenv("ANTHROPIC_API_KEY")
+
+    def _default_params(self):
+        """Get the default parameters for calling Anthropic API."""
+        d = {
+            "max_tokens_to_sample": self.max_tokens_to_sample,
+            "model": self.model,
+        }
+        if self.temperature is not None:
+            d["temperature"] = self.temperature
+        if self.top_k is not None:
+            d["top_k"] = self.top_k
+        if self.top_p is not None:
+            d["top_p"] = self.top_p
+        return d
+
+    def _call(self, prompt, stop=None):
+        """Call out to Anthropic's completion endpoint."""
+        stop = stop or []
+        params = self._default_params()
+        headers = {"Authorization": f"Bearer {self.anthropic_api_key}"}
+        data = {
+            "prompt": prompt,
+            "stop_sequences": stop,
+            **params
+        }
+        response = requests.post(f"{self.anthropic_api_url}/completions", headers=headers, json=data, timeout=self.default_request_timeout)
+        return response.json().get("completion")

swarms/agents/prompts/prompt_generator.py

@@ -0,0 +1,186 @@
+import json
+from typing import List
+
+from langchain.tools.base import BaseTool
+
+FINISH_NAME = "finish"
+
+
+class PromptGenerator:
+    """A class for generating custom prompt strings.
+
+    Does this based on constraints, commands, resources, and performance evaluations.
+    """
+
+    def __init__(self) -> None:
+        """Initialize the PromptGenerator object.
+
+        Starts with empty lists of constraints, commands, resources,
+        and performance evaluations.
+        """
+        self.constraints: List[str] = []
+        self.commands: List[BaseTool] = []
+        self.resources: List[str] = []
+        self.performance_evaluation: List[str] = []
+        self.response_format = {
+            "thoughts": {
+                "text": "thought",
+                "reasoning": "reasoning",
+                "plan": "- short bulleted\n- list that conveys\n- long-term plan",
+                "criticism": "constructive self-criticism",
+                "speak": "thoughts summary to say to user",
+            },
+            "command": {"name": "command name", "args": {"arg name": "value"}},
+        }
+
+    def add_constraint(self, constraint: str) -> None:
+        """
+        Add a constraint to the constraints list.
+
+        Args:
+            constraint (str): The constraint to be added.
+        """
+        self.constraints.append(constraint)
+
+    def add_tool(self, tool: BaseTool) -> None:
+        self.commands.append(tool)
+
+    def _generate_command_string(self, tool: BaseTool) -> str:
+        output = f"{tool.name}: {tool.description}"
+        output += f", args json schema: {json.dumps(tool.args)}"
+        return output
+
+    def add_resource(self, resource: str) -> None:
+        """
+        Add a resource to the resources list.
+
+        Args:
+            resource (str): The resource to be added.
+        """
+        self.resources.append(resource)
+
+    def add_performance_evaluation(self, evaluation: str) -> None:
+        """
+        Add a performance evaluation item to the performance_evaluation list.
+
+        Args:
+            evaluation (str): The evaluation item to be added.
+        """
+        self.performance_evaluation.append(evaluation)
+
+    def _generate_numbered_list(self, items: list, item_type: str = "list") -> str:
+        """
+        Generate a numbered list from given items based on the item_type.
+
+        Args:
+            items (list): A list of items to be numbered.
+            item_type (str, optional): The type of items in the list.
+                Defaults to 'list'.
+
+        Returns:
+            str: The formatted numbered list.
+        """
+        if item_type == "command":
+            command_strings = [
+                f"{i + 1}. {self._generate_command_string(item)}"
+                for i, item in enumerate(items)
+            ]
+            finish_description = (
+                "use this to signal that you have finished all your objectives"
+            )
+            finish_args = (
+                '"response": "final response to let '
+                'people know you have finished your objectives"'
+            )
+            finish_string = (
+                f"{len(items) + 1}. {FINISH_NAME}: "
+                f"{finish_description}, args: {finish_args}"
+            )
+            return "\n".join(command_strings + [finish_string])
+        else:
+            return "\n".join(f"{i+1}. {item}" for i, item in enumerate(items))
+
+    def generate_prompt_string(self) -> str:
+        """Generate a prompt string.
+
+        Returns:
+            str: The generated prompt string.
+        """
+        formatted_response_format = json.dumps(self.response_format, indent=4)
+        prompt_string = (
+            f"Constraints:\n{self._generate_numbered_list(self.constraints)}\n\n"
+            f"Commands:\n"
+            f"{self._generate_numbered_list(self.commands, item_type='command')}\n\n"
+            f"Resources:\n{self._generate_numbered_list(self.resources)}\n\n"
+            f"Performance Evaluation:\n"
+            f"{self._generate_numbered_list(self.performance_evaluation)}\n\n"
+            f"You should only respond in JSON format as described below "
+            f"\nResponse Format: \n{formatted_response_format} "
+            f"\nEnsure the response can be parsed by Python json.loads"
+        )
+
+        return prompt_string
+
+
+def get_prompt(tools: List[BaseTool]) -> str:
+    """Generates a prompt string.
+
+    It includes various constraints, commands, resources, and performance evaluations.
+
+    Returns:
+        str: The generated prompt string.
+    """
+
+    # Initialize the PromptGenerator object
+    prompt_generator = PromptGenerator()
+
+    # Add constraints to the PromptGenerator object
+    prompt_generator.add_constraint(
+        "~4000 word limit for short term memory. "
+        "Your short term memory is short, "
+        "so immediately save important information to files."
+    )
+    prompt_generator.add_constraint(
+        "If you are unsure how you previously did something "
+        "or want to recall past events, "
+        "thinking about similar events will help you remember."
+    )
+    prompt_generator.add_constraint("No user assistance")
+    prompt_generator.add_constraint(
+        'Exclusively use the commands listed in double quotes e.g. "command name"'
+    )
+
+    # Add commands to the PromptGenerator object
+    for tool in tools:
+        prompt_generator.add_tool(tool)
+
+    # Add resources to the PromptGenerator object
+    prompt_generator.add_resource(
+        "Internet access for searches and information gathering."
+    )
+    prompt_generator.add_resource("Long Term memory management.")
+    prompt_generator.add_resource(
+        "GPT-3.5 powered Agents for delegation of simple tasks."
+    )
+    prompt_generator.add_resource("File output.")
+
+    # Add performance evaluations to the PromptGenerator object
+    prompt_generator.add_performance_evaluation(
+        "Continuously review and analyze your actions "
+        "to ensure you are performing to the best of your abilities."
+    )
+    prompt_generator.add_performance_evaluation(
+        "Constructively self-criticize your big-picture behavior constantly."
+    )
+    prompt_generator.add_performance_evaluation(
+        "Reflect on past decisions and strategies to refine your approach."
+    )
+    prompt_generator.add_performance_evaluation(
+        "Every command has a cost, so be smart and efficient. "
+        "Aim to complete tasks in the least number of steps."
+    )
+
+    # Generate the prompt string
+    prompt_string = prompt_generator.generate_prompt_string()
+
+    return prompt_string

swarms/agents/utils/human_input.py

@@ -1,35 +1,24 @@
 """Tool for asking human input."""
 
-from typing import Callable, Optional
 
-from pydantic import Field
-
-from langchain.callbacks.manager import CallbackManagerForToolRun
-from langchain.tools.base import BaseTool
-
-
-def _print_func(text: str) -> None:
-    print("\n")
-    print(text)
-
-
-class HumanInputRun(BaseTool):
+class HumanInputRun:
     """Tool that asks user for input."""
 
-    name = "human"
-    description = (
-        "You can ask a human for guidance when you think you "
-        "got stuck or you are not sure what to do next. "
-        "The input should be a question for the human."
-    )
-    prompt_func: Callable[[str], None] = Field(default_factory=lambda: _print_func)
-    input_func: Callable = Field(default_factory=lambda: input)
-
-    def _run(
-        self,
-        query: str,
-        run_manager: Optional[CallbackManagerForToolRun] = None,
-    ) -> str:
+    def __init__(self, prompt_func=None, input_func=None):
+        self.name = "human"
+        self.description = (
+            "You can ask a human for guidance when you think you "
+            "got stuck or you are not sure what to do next. "
+            "The input should be a question for the human."
+        )
+        self.prompt_func = prompt_func if prompt_func else self._print_func
+        self.input_func = input_func if input_func else input
+
+    def _print_func(self, text: str) -> None:
+        print("\n")
+        print(text)
+
+    def run(self, query: str) -> str:
         """Use the Human input tool."""
         self.prompt_func(query)
         return self.input_func()

tests/agents/models/hf.py

@@ -1,7 +1,7 @@
 import pytest
 import torch
 from unittest.mock import Mock
-from swarms.agents.models.hf import HuggingFaceLLM  
+from swarms.agents.models.huggingface import HuggingFaceLLM  
 
 
 @pytest.fixture