feat: +Reverse engineering related logic

metagpt/configs/llm_config.py

@@ -42,6 +42,7 @@ class LLMConfig(YamlModel):
     api_version: Optional[str] = None
 
     model: Optional[str] = None  # also stands for DEPLOYMENT_NAME
+    pricing_plan: Optional[str] = None  # Cost Settlement Plan Parameters.
 
     # For Spark(Xunfei), maybe remove later
     app_id: Optional[str] = None

metagpt/provider/azure_openai_api.py

@@ -8,14 +8,10 @@
 """
 from openai import AsyncAzureOpenAI
 from openai._base_client import AsyncHttpxClientWrapper
-from openai.types import CompletionUsage
 
 from metagpt.configs.llm_config import LLMType
-from metagpt.logs import logger
 from metagpt.provider.llm_provider_registry import register_provider
 from metagpt.provider.openai_api import OpenAILLM
-from metagpt.utils import TOKEN_COSTS, count_message_tokens, count_string_tokens
-from metagpt.utils.exceptions import handle_exception
 
 
 @register_provider(LLMType.AZURE)
@@ -29,6 +25,7 @@ def _init_client(self):
         # https://learn.microsoft.com/zh-cn/azure/ai-services/openai/how-to/migration?tabs=python-new%2Cdalle-fix
         self.aclient = AsyncAzureOpenAI(**kwargs)
         self.model = self.config.model  # Used in _calc_usage & _cons_kwargs
+        self.pricing_plan = self.config.pricing_plan
 
     def _make_client_kwargs(self) -> dict:
         kwargs = dict(
@@ -43,38 +40,3 @@ def _make_client_kwargs(self) -> dict:
             kwargs["http_client"] = AsyncHttpxClientWrapper(**proxy_params)
 
         return kwargs
-
-    def _calc_usage(self, messages: list[dict], rsp: str) -> CompletionUsage:
-        usage = CompletionUsage(prompt_tokens=0, completion_tokens=0, total_tokens=0)
-        if not self.config.calc_usage:
-            return usage
-
-        model_name = "gpt-35-turbo" if "gpt-3" in self.model.lower() else "gpt-4-turbo-preview"
-        try:
-            usage.prompt_tokens = count_message_tokens(messages, model_name)
-            usage.completion_tokens = count_string_tokens(rsp, model_name)
-        except Exception as e:
-            logger.error(f"usage calculation failed: {e}")
-
-        return usage
-
-    @handle_exception
-    def _update_costs(self, usage: CompletionUsage):
-        if self.config.calc_usage and usage and self.cost_manager:
-            model_name = self._get_azure_model()
-            # More about pricing: https://azure.microsoft.com/en-us/pricing/details/cognitive-services/openai-service/
-            self.cost_manager.update_cost(usage.prompt_tokens, usage.completion_tokens, model_name)
-
-    def _get_azure_model(self) -> str:
-        models = [i.lower() for i in TOKEN_COSTS.keys() if "azure" in i]
-        mappings = {i: set(i.split("-")) for i in models}
-        words = self.model.lower().split("-")
-        weights = []
-        for k, v in mappings.items():
-            count = 0
-            for i in words:
-                if i in v:
-                    count += 1
-            weights.append((k, count))
-        sorted_list = sorted(weights, key=lambda x: x[1], reverse=True)
-        return sorted_list[0][0]

metagpt/provider/base_llm.py

@@ -6,16 +6,20 @@
 @File    : base_llm.py
 @Desc    : mashenquan, 2023/8/22. + try catch
 """
+from __future__ import annotations
+
 import json
 from abc import ABC, abstractmethod
-from typing import Optional, Union
+from typing import Dict, Optional, Union
 
 from openai import AsyncOpenAI
+from openai.types import CompletionUsage
 
 from metagpt.configs.llm_config import LLMConfig
 from metagpt.logs import logger
 from metagpt.schema import Message
 from metagpt.utils.cost_manager import CostManager
+from metagpt.utils.exceptions import handle_exception
 
 
 class BaseLLM(ABC):
@@ -29,6 +33,7 @@ class BaseLLM(ABC):
     aclient: Optional[Union[AsyncOpenAI]] = None
     cost_manager: Optional[CostManager] = None
     model: Optional[str] = None
+    pricing_plan: Optional[str] = None
 
     @abstractmethod
     def __init__(self, config: LLMConfig):
@@ -168,3 +173,30 @@ def get_choice_function_arguments(self, rsp: dict) -> dict:
             {'language': 'python', 'code': "print('Hello, World!')"}
         """
         return json.loads(self.get_choice_function(rsp)["arguments"])
+
+    @handle_exception
+    def _update_costs(self, usage: CompletionUsage | Dict):
+        """
+        Updates the costs based on the provided usage information.
+
+        Args:
+            usage (Union[CompletionUsage, Dict]): The usage information used to calculate and update costs.
+                It can be either an instance of CompletionUsage or a dictionary.
+
+        Returns:
+            None: This method does not return any value.
+
+        Raises:
+            ValueError: If the provided usage is not a valid format.
+
+        Example:
+            Usage example goes here, demonstrating how to call and utilize this method.
+        """
+        if self.config.calc_usage and usage and self.cost_manager:
+            if isinstance(usage, Dict):
+                prompt_tokens = int(usage.get("prompt_tokens", 0))
+                completion_tokens = int(usage.get("completion_tokens", 0))
+            else:
+                prompt_tokens = usage.prompt_tokens
+                completion_tokens = usage.completion_tokens
+            self.cost_manager.update_cost(prompt_tokens, completion_tokens, self.pricing_plan)

metagpt/provider/fireworks_api.py

@@ -81,14 +81,6 @@ def _make_client_kwargs(self) -> dict:
         kwargs = dict(api_key=self.config.api_key, base_url=self.config.base_url)
         return kwargs
 
-    def _update_costs(self, usage: CompletionUsage):
-        if self.config.calc_usage and usage:
-            try:
-                # use FireworksCostManager not context.cost_manager
-                self.cost_manager.update_cost(usage.prompt_tokens, usage.completion_tokens, self.model)
-            except Exception as e:
-                logger.error(f"updating costs failed!, exp: {e}")
-
     def get_costs(self) -> Costs:
         return self.cost_manager.get_costs()
 

metagpt/provider/google_gemini_api.py

@@ -53,6 +53,7 @@ def __init__(self, config: LLMConfig):
         self.__init_gemini(config)
         self.config = config
         self.model = "gemini-pro"  # so far only one model
+        self.pricing_plan = self.config.pricing_plan or self.model
         self.llm = GeminiGenerativeModel(model_name=self.model)
 
     def __init_gemini(self, config: LLMConfig):
@@ -70,16 +71,6 @@ def _const_kwargs(self, messages: list[dict], stream: bool = False) -> dict:
         kwargs = {"contents": messages, "generation_config": GenerationConfig(temperature=0.3), "stream": stream}
         return kwargs
 
-    def _update_costs(self, usage: dict):
-        """update each request's token cost"""
-        if self.config.calc_usage:
-            try:
-                prompt_tokens = int(usage.get("prompt_tokens", 0))
-                completion_tokens = int(usage.get("completion_tokens", 0))
-                self.cost_manager.update_cost(prompt_tokens, completion_tokens, self.model)
-            except Exception as e:
-                logger.error(f"google gemini updats costs failed! exp: {e}")
-
     def get_choice_text(self, resp: GenerateContentResponse) -> str:
         return resp.text
 

metagpt/provider/metagpt_api.py

@@ -10,19 +10,11 @@
 from metagpt.configs.llm_config import LLMType
 from metagpt.provider import OpenAILLM
 from metagpt.provider.llm_provider_registry import register_provider
-from metagpt.utils.exceptions import handle_exception
 
 
 @register_provider(LLMType.METAGPT)
 class MetaGPTLLM(OpenAILLM):
     def _calc_usage(self, messages: list[dict], rsp: str) -> CompletionUsage:
-        usage = CompletionUsage(prompt_tokens=0, completion_tokens=0, total_tokens=0)
-
         # The current billing is based on usage frequency. If there is a future billing logic based on the
         # number of tokens, please refine the logic here accordingly.
-
-        return usage
-
-    @handle_exception
-    def _update_costs(self, usage: CompletionUsage):
-        pass
+        return CompletionUsage(prompt_tokens=0, completion_tokens=0, total_tokens=0)

metagpt/provider/ollama_api.py

@@ -36,26 +36,17 @@ def __init__(self, config: LLMConfig):
         self.suffix_url = "/chat"
         self.http_method = "post"
         self.use_system_prompt = False
-        self._cost_manager = TokenCostManager()
+        self.cost_manager = TokenCostManager()
 
     def __init_ollama(self, config: LLMConfig):
         assert config.base_url, "ollama base url is required!"
         self.model = config.model
+        self.pricing_plan = self.model
 
     def _const_kwargs(self, messages: list[dict], stream: bool = False) -> dict:
         kwargs = {"model": self.model, "messages": messages, "options": {"temperature": 0.3}, "stream": stream}
         return kwargs
 
-    def _update_costs(self, usage: dict):
-        """update each request's token cost"""
-        if self.config.calc_usage:
-            try:
-                prompt_tokens = int(usage.get("prompt_tokens", 0))
-                completion_tokens = int(usage.get("completion_tokens", 0))
-                self._cost_manager.update_cost(prompt_tokens, completion_tokens, self.model)
-            except Exception as e:
-                logger.error(f"ollama updats costs failed! exp: {e}")
-
     def get_choice_text(self, resp: dict) -> str:
         """get the resp content from llm response"""
         assist_msg = resp.get("message", {})

metagpt/provider/open_llm_api.py

@@ -16,7 +16,7 @@
 class OpenLLM(OpenAILLM):
     def __init__(self, config: LLMConfig):
         super().__init__(config)
-        self._cost_manager = TokenCostManager()
+        self.cost_manager = TokenCostManager()
 
     def _make_client_kwargs(self) -> dict:
         kwargs = dict(api_key="sk-xxx", base_url=self.config.base_url)
@@ -35,13 +35,5 @@ def _calc_usage(self, messages: list[dict], rsp: str) -> CompletionUsage:
 
         return usage
 
-    def _update_costs(self, usage: CompletionUsage):
-        if self.config.calc_usage and usage:
-            try:
-                # use OpenLLMCostManager not CONFIG.cost_manager
-                self._cost_manager.update_cost(usage.prompt_tokens, usage.completion_tokens, self.model)
-            except Exception as e:
-                logger.error(f"updating costs failed!, exp: {e}")
-
     def get_costs(self) -> Costs:
-        return self._cost_manager.get_costs()
+        return self.cost_manager.get_costs()

metagpt/provider/openai_api.py

@@ -6,6 +6,7 @@
 @Modified By: mashenquan, 2023/11/21. Fix bug: ReadTimeout.
 @Modified By: mashenquan, 2023/12/1. Fix bug: Unclosed connection caused by openai 0.x.
 """
+from __future__ import annotations
 
 import json
 from typing import AsyncIterator, Optional, Union
@@ -62,6 +63,7 @@ def __init__(self, config: LLMConfig):
 
     def _init_model(self):
         self.model = self.config.model  # Used in _calc_usage & _cons_kwargs
+        self.pricing_plan = self.config.pricing_plan or self.model
 
     def _init_client(self):
         """https://github.com/openai/openai-python#async-usage"""
@@ -210,18 +212,13 @@ def _calc_usage(self, messages: list[dict], rsp: str) -> CompletionUsage:
             return usage
 
         try:
-            usage.prompt_tokens = count_message_tokens(messages, self.model)
-            usage.completion_tokens = count_string_tokens(rsp, self.model)
+            usage.prompt_tokens = count_message_tokens(messages, self.pricing_plan)
+            usage.completion_tokens = count_string_tokens(rsp, self.pricing_plan)
         except Exception as e:
             logger.error(f"usage calculation failed: {e}")
 
         return usage
 
-    @handle_exception
-    def _update_costs(self, usage: CompletionUsage):
-        if self.config.calc_usage and usage and self.cost_manager:
-            self.cost_manager.update_cost(usage.prompt_tokens, usage.completion_tokens, self.model)
-
     def get_costs(self) -> Costs:
         if not self.cost_manager:
             return Costs(0, 0, 0, 0)

metagpt/provider/zhipuai_api.py

@@ -3,6 +3,7 @@
 # @Desc   : zhipuai LLM from https://open.bigmodel.cn/dev/api#sdk
 
 from enum import Enum
+from typing import Optional
 
 import openai
 import zhipuai
@@ -21,6 +22,7 @@
 from metagpt.provider.llm_provider_registry import register_provider
 from metagpt.provider.openai_api import log_and_reraise
 from metagpt.provider.zhipuai.zhipu_model_api import ZhiPuModelAPI
+from metagpt.utils.cost_manager import CostManager
 
 
 class ZhiPuEvent(Enum):
@@ -39,10 +41,12 @@ class ZhiPuAILLM(BaseLLM):
 
     def __init__(self, config: LLMConfig):
         self.__init_zhipuai(config)
+        self.config = config
         self.llm = ZhiPuModelAPI
         self.model = "chatglm_turbo"  # so far only one model, just use it
+        self.pricing_plan = self.config.pricing_plan or self.model
         self.use_system_prompt: bool = False  # zhipuai has no system prompt when use api
-        self.config = config
+        self.cost_manager: Optional[CostManager] = None
 
     def __init_zhipuai(self, config: LLMConfig):
         assert config.api_key
@@ -57,16 +61,6 @@ def _const_kwargs(self, messages: list[dict], stream: bool = False) -> dict:
         kwargs = {"model": self.model, "messages": messages, "stream": stream, "temperature": 0.3}
         return kwargs
 
-    def _update_costs(self, usage: dict):
-        """update each request's token cost"""
-        if self.config.calc_usage:
-            try:
-                prompt_tokens = int(usage.get("prompt_tokens", 0))
-                completion_tokens = int(usage.get("completion_tokens", 0))
-                self.config.cost_manager.update_cost(prompt_tokens, completion_tokens, self.model)
-            except Exception as e:
-                logger.error(f"zhipuai updats costs failed! exp: {e}")
-
     def completion(self, messages: list[dict], timeout=3) -> dict:
         resp = self.llm.chat.completions.create(**self._const_kwargs(messages))
         usage = resp.usage.model_dump()

metagpt/utils/common.py

@@ -26,7 +26,7 @@
 import typing
 from io import BytesIO
 from pathlib import Path
-from typing import Any, List, Tuple, Union
+from typing import Any, Callable, List, Tuple, Union
 from urllib.parse import quote, unquote
 
 import aiofiles
@@ -784,3 +784,45 @@ async def awrite_bin(filename: str | Path, data: bytes):
     pathname.parent.mkdir(parents=True, exist_ok=True)
     async with aiofiles.open(str(pathname), mode="wb") as writer:
         await writer.write(data)
+
+
+def is_coroutine_func(func: Callable) -> bool:
+    return inspect.iscoroutinefunction(func)
+
+
+def load_mc_skills_code(skill_names: list[str] = None, skills_dir: Path = None) -> list[str]:
+    """load mincraft skill from js files"""
+    if not skills_dir:
+        skills_dir = Path(__file__).parent.absolute()
+    if skill_names is None:
+        skill_names = [skill[:-3] for skill in os.listdir(f"{skills_dir}") if skill.endswith(".js")]
+    skills = [skills_dir.joinpath(f"{skill_name}.js").read_text() for skill_name in skill_names]
+    return skills
+
+
+def encode_image(image_path_or_pil: Union[Path, Image], encoding: str = "utf-8") -> str:
+    """encode image from file or PIL.Image into base64"""
+    if isinstance(image_path_or_pil, Image.Image):
+        buffer = BytesIO()
+        image_path_or_pil.save(buffer, format="JPEG")
+        bytes_data = buffer.getvalue()
+    else:
+        if not image_path_or_pil.exists():
+            raise FileNotFoundError(f"{image_path_or_pil} not exists")
+        with open(str(image_path_or_pil), "rb") as image_file:
+            bytes_data = image_file.read()
+    return base64.b64encode(bytes_data).decode(encoding)
+
+
+def decode_image(img_url_or_b64: str) -> Image:
+    """decode image from url or base64 into PIL.Image"""
+    if img_url_or_b64.startswith("http"):
+        # image http(s) url
+        resp = requests.get(img_url_or_b64)
+        img = Image.open(BytesIO(resp.content))
+    else:
+        # image b64_json
+        b64_data = re.sub("^data:image/.+;base64,", "", img_url_or_b64)
+        img_data = BytesIO(base64.b64decode(b64_data))
+        img = Image.open(img_data)
+    return img

metagpt/utils/cost_manager.py

@@ -39,7 +39,7 @@ def update_cost(self, prompt_tokens, completion_tokens, model):
         completion_tokens (int): The number of tokens used in the completion.
         model (str): The model used for the API call.
         """
-        if prompt_tokens + completion_tokens == 0:
+        if prompt_tokens + completion_tokens == 0 or not model:
             return
         self.total_prompt_tokens += prompt_tokens
         self.total_completion_tokens += completion_tokens