update get rpm from api.openai.com

config/config.yaml

@@ -9,6 +9,10 @@ OPENAI_API_BASE: "https://api.openai.com/v1"
 #OPENAI_PROXY: "http://127.0.0.1:8118"
 #OPENAI_API_KEY: "YOUR_API_KEY"
 OPENAI_API_MODEL: "gpt-4"
+## If you want to automate the setup for RPM and MAX_TOKENS, please log in through your own browser at https://platform.openai.com/account/limits, enable developer mode, and find 'rate_limits' in the network section. 
+## If it's not found, refresh the recording using Ctrl+R. After locating 'rate_limits', find 'Authorization: sess-xxx' in the request header, where 'sess-xxx' is the session key.
+## If you wish to manually set RPM, please set RPM and either do not set OPENAI_SESSION_KEY or set it to "".
+# OPENAI_SESSION_KEY: ""
 MAX_TOKENS: 1500
 RPM: 10
 

metagpt/config.py

@@ -44,6 +44,7 @@ def __init__(self, yaml_file=default_yaml_file):
         logger.info("Config loading done.")
         self.global_proxy = self._get("GLOBAL_PROXY")
         self.openai_api_key = self._get("OPENAI_API_KEY")
+        self.openai_session_key = self._get("OPENAI_SESSION_KEY", "")
         self.anthropic_api_key = self._get("Anthropic_API_KEY")
         if (not self.openai_api_key or "YOUR_API_KEY" == self.openai_api_key) and (
                 not self.anthropic_api_key or "YOUR_API_KEY" == self.anthropic_api_key

metagpt/provider/openai_api.py

@@ -5,10 +5,13 @@
 @File    : openai.py
 """
 import asyncio
+import aiohttp
 import time
+import json
 from typing import NamedTuple, Union
 
 import openai
+import requests
 from openai.error import APIConnectionError
 from tenacity import (
     after_log,
@@ -33,16 +36,57 @@
 class RateLimiter:
     """Rate control class, each call goes through wait_if_needed, sleep if rate control is needed"""
 
-    def __init__(self, rpm):
+    def __init__(self):
         self.last_call_time = 0
+        self.rpm = None
+        self.interval = None
         # Here 1.1 is used because even if the calls are made strictly according to time,
         # they will still be QOS'd; consider switching to simple error retry later
-        self.interval = 1.1 * 60 / rpm
-        self.rpm = rpm
 
     def split_batches(self, batch):
+        if self.rpm is None:
+            raise ValueError("Your must run update_rpm before calling split_batches.")
         return [batch[i : i + self.rpm] for i in range(0, len(batch), self.rpm)]
 
+    async def update_rpm(self):
+        if self.rpm is None:
+            self.rpm = await self._aget_rpm()
+            self.interval = 1.1 * 60 / self.rpm
+            logger.info(f'Setting rpm to {self.rpm}')
+
+    async def _aget_rpm(self):
+        session_key = CONFIG.get("OPENAI_SESSION_KEY", "")
+        default_rpm = int(CONFIG.get("RPM", 10))
+        if len(session_key) > 0:
+            try:
+                async with aiohttp.ClientSession() as session:
+                    async with session.get(
+                                        "https://api.openai.com/dashboard/rate_limits",
+                                        headers={"Authorization": f"Bearer {session_key}"},
+                                        timeout=10,
+                                        proxy=openai.proxy
+                    ) as response:
+                        if response.status == 200:
+                            response_content = json.loads(await response.text())
+                            if CONFIG.openai_api_model not in response_content:
+                                raise ValueError("Get rpm from api.openai.com error. \
+                                                 You have entered a model name that is not supported by OpenAI, or the input is incorrect. \
+                                                 Please enter the correct name in the configuration file. \
+                                                 Setting rpm to default parameter.")
+
+                            limit_dict = response_content[CONFIG.openai_api_model]
+                            return limit_dict["max_requests_per_1_minute"]
+                        else:
+                            error = json.loads(await response.text())["error"]
+                            logger.error(f"Connection to api.openai.com failed:{error}.Setting rpm to default parameter.")
+                            return default_rpm
+
+            except Exception as exp:
+                logger.error(f"Connection to api.openai.com failed, error type:{type(exp).__name__}, error message:{str(exp)}.Setting rpm to default parameter.")
+                return default_rpm
+        else:
+            return default_rpm
+
     async def wait_if_needed(self, num_requests):
         current_time = time.time()
         elapsed_time = current_time - self.last_call_time
@@ -143,21 +187,20 @@ class OpenAIGPTAPI(BaseGPTAPI, RateLimiter):
     """
 
     def __init__(self):
-        self.__init_openai(CONFIG)
+        self.__init_openai()
         self.llm = openai
         self.model = CONFIG.openai_api_model
         self.auto_max_tokens = False
         self._cost_manager = CostManager()
-        RateLimiter.__init__(self, rpm=self.rpm)
 
-    def __init_openai(self, config):
-        openai.api_key = config.openai_api_key
-        if config.openai_api_base:
-            openai.api_base = config.openai_api_base
-        if config.openai_api_type:
-            openai.api_type = config.openai_api_type
-            openai.api_version = config.openai_api_version
-        self.rpm = int(config.get("RPM", 10))
+    def __init_openai(self):
+        openai.api_key = CONFIG.openai_api_key
+        if CONFIG.openai_api_base:
+            openai.api_base = CONFIG.openai_api_base
+        if CONFIG.openai_api_type:
+            openai.api_type = CONFIG.openai_api_type
+            openai.api_version = CONFIG.openai_api_version
+        self.rpm = None
 
     async def _achat_completion_stream(self, messages: list[dict]) -> str:
         response = await openai.ChatCompletion.acreate(**self._cons_kwargs(messages), stream=True)
@@ -255,6 +298,8 @@ def _calc_usage(self, messages: list[dict], rsp: str) -> dict:
 
     async def acompletion_batch(self, batch: list[list[dict]]) -> list[dict]:
         """Return full JSON"""
+
+        await self.update_rpm()
         split_batches = self.split_batches(batch)
         all_results = []