Merge 'integration_2025-03-27_808630449922' into 'master'

merge branch integration_2025-03-27_808630449922 into master

See merge request: !553

Author: 修杰

meta.json

@@ -1,4 +1,4 @@
 {
-	"lasted": "1.1.0",
-	"meta_commit": "fb5ff08e537bd963d0f03359b644c7347f2a0645"
+	"lasted": "1.1.1",
+	"meta_commit": "eb656d5036a263d4e68911ee390d36802e05547b"
 }

setup.py

@@ -3,7 +3,7 @@
 from setuptools import setup, find_packages  # noqa: H301
 
 NAME = "volcengine-python-sdk"
-VERSION = "1.0.130"
+VERSION = "1.1.1"
 # To install the library, run the following
 #
 # python setup.py install

volcenginesdkarkruntime/_utils/_model_breaker.py

@@ -1,23 +1,83 @@
-from datetime import datetime, timedelta
+import asyncio
+import time
+import uuid
+import threading
 
 
-class ModelBreaker:
+class _QuerySet(object):
+    def __init__(self):
+        self._items = list()
+        self._index = dict()
+        self._lock = threading.Lock()
+
+    def add(self, item: int) -> None:
+        with self._lock:
+            if item in self._index:
+                return
+
+            self._items.append(item)
+            self._index[item] = len(self._items) - 1
+
+    def remove(self, item: int) -> None:
+        with self._lock:
+            if item not in self._index:
+                return
+
+            index = self._index[item]
+            self._items[index] = self._items[-1]
+            self._index[self._items[-1]] = index
+            self._items.pop()
+            del self._index[item]
+
+    def query(self, item: int) -> int:
+        with self._lock:
+            return self._index[item]
+
+
+class ModelBreaker(object):
     def __init__(self):
         # 初始化 allow_time 为当前时间
-        self.allow_time = datetime.now()
+        self._allow_time = time.perf_counter()
+        self._waiters = _QuerySet()
 
-    def allow(self):
-        # 检查当前时间是否在 allow_time 之后
-        return datetime.now() > self.allow_time
+    def _allow(self, id: int) -> bool:
+        cur = time.perf_counter()
+        # 如果当前时间小于等于 allow_time，不允许通过
+        if cur <= self._allow_time:
+            return 0
+        # 如果当前时间与 allow_time 的差值大于 10，允许通过
+        if cur - self._allow_time > 10:
+            return True
+        # 如果当前时间与 allow_time 的差值小于等于 10，慢启动通过
+        return self._waiters.query(id) < 2 ** (cur - self._allow_time)
 
-    def reset(self, duration):
+    def _get_allowed_duration(self) -> float:
+        # 计算当前时间与 allow_time 之间的持续时间
+        allow_duration = self._allow_time - time.perf_counter()
+
+        # 至少有 1 秒的等待时间
+        return max(allow_duration, 1)
+
+    def _acquire(self) -> int:
+        id = uuid.uuid4().int
+        self._waiters.add(id)
+        return id
+
+    def _release(self, id: int) -> None:
+        self._waiters.remove(id)
+
+    def reset(self, duration: float) -> None:
         # 将 allow_time 重置为当前时间加上指定的持续时间
-        self.allow_time = datetime.now() + timedelta(seconds=duration.total_seconds())
+        self._allow_time = time.perf_counter() + duration
 
-    def get_allowed_duration(self):
-        # 计算当前时间与 allow_time 之间的持续时间
-        allow_duration = self.allow_time - datetime.now()
-        # 如果持续时间为负，则返回一个零时长的 timedelta 对象
-        if allow_duration.total_seconds() < 0:
-            return timedelta(0)
-        return allow_duration
+    def wait(self) -> None:
+        id = self._acquire()
+        while not self._allow(id):
+            time.sleep(self._get_allowed_duration())
+        self._release(id)
+
+    async def asyncwait(self) -> None:
+        id = self._acquire()
+        while not self._allow(id):
+            await asyncio.sleep(self._get_allowed_duration())
+        self._release(id)

volcenginesdkarkruntime/resources/batch_chat/completions.py

@@ -149,10 +149,7 @@ def create(
         last_time = self._get_request_last_time(timeout)
         model_breaker = self._client.get_model_breaker(model)
         while True:
-            while not model_breaker.allow():
-                if datetime.now() + timedelta(seconds=model_breaker.get_allowed_duration().total_seconds()) > last_time:
-                    raise ArkAPITimeoutError()
-                time.sleep(model_breaker.get_allowed_duration().total_seconds())
+            model_breaker.wait()
             if datetime.now() > last_time:
                 raise ArkAPITimeoutError()
             try:
@@ -198,7 +195,7 @@ def create(
             except ArkAPIStatusError as err:
                 retry_after = _get_retry_after(err.response)
                 if retry_after is not None:
-                    model_breaker.reset(timedelta(seconds=retry_after))
+                    model_breaker.reset(retry_after)
                 if _should_retry(err.response):
                     continue
                 else:
@@ -295,10 +292,7 @@ async def create(
         last_time = self._get_request_last_time(timeout)
         model_breaker = await self._client.get_model_breaker(model)
         while True:
-            while not model_breaker.allow():
-                if datetime.now() + timedelta(seconds=model_breaker.get_allowed_duration().total_seconds()) > last_time:
-                    raise ArkAPITimeoutError()
-                await asyncio.sleep(model_breaker.get_allowed_duration().total_seconds())
+            await model_breaker.asyncwait()
             if datetime.now() > last_time:
                 raise ArkAPITimeoutError()
             try:
@@ -344,7 +338,7 @@ async def create(
             except ArkAPIStatusError as err:
                 retry_after = _get_retry_after(err.response)
                 if retry_after is not None:
-                    model_breaker.reset(timedelta(seconds=retry_after))
+                    model_breaker.reset(retry_after)
                 if _should_retry(err.response):
                     continue
                 else:

volcenginesdkcore/api_client.py

@@ -67,7 +67,7 @@ def __init__(self, configuration=None, header_name=None, header_value=None,
             self.default_headers[header_name] = header_value
         self.cookie = cookie
         # Set default User-Agent.
-        self.user_agent = 'volcstack-python-sdk/1.0.130'
+        self.user_agent = 'volcstack-python-sdk/1.1.1'
         self.client_side_validation = configuration.client_side_validation
 
     def __del__(self):

volcenginesdkcore/configuration.py

@@ -221,5 +221,5 @@ def to_debug_report(self):
                "OS: {env}\n"\
                "Python Version: {pyversion}\n"\
                "Version of the API: 0.1.0\n"\
-               "SDK Package Version: 1.0.130".\
+               "SDK Package Version: 1.1.1".\
                format(env=sys.platform, pyversion=sys.version)

volcenginesdkexamples/volcenginesdkarkruntime/async_batch_chat_completions.py

@@ -2,11 +2,8 @@
 import sys
 from datetime import datetime
 
-import uvloop
-
 from volcenginesdkarkruntime import AsyncArk
 
-
 # Authentication
 # 1.If you authorize your endpoint using an API key, you can set your api key to environment variable "ARK_API_KEY"
 # or specify api key by Ark(api_key="${YOUR_API_KEY}").
@@ -18,45 +15,67 @@
 # To get your ak&sk, please refer to this document(https://www.volcengine.com/docs/6291/65568)
 # For more information，please check this document（https://www.volcengine.com/docs/82379/1263279）
 
-async def worker(worker_id, task_num):
-    client = AsyncArk()
+
+async def worker(
+    worker_id: int,
+    client: AsyncArk,
+    requests: asyncio.Queue[dict],
+):
     print(f"Worker {worker_id} is starting.")
-    for i in range(task_num):
-        print(f"Worker {worker_id} task {i} is running.")
+
+    while True:
+        request = await requests.get()
         try:
-            completion = await client.batch_chat.completions.create(
-                model="${YOUR_ENDPOINT_ID}",
-                messages=[
-                    {"role": "system", "content": "你是豆包，是由字节跳动开发的 AI 人工智能助手"},
-                    {"role": "user", "content": "常见的十字花科植物有哪些？"},
-                ],
-            )
-            print(completion.choices[0].message.content)
+            completion = await client.batch_chat.completions.create(**request)
+            print(completion)
         except Exception as e:
-            print(f"Worker {worker_id} task {i} failed with error: {e}")
-        else:
-            print(f"Worker {worker_id} task {i} is completed.")
-    print(f"Worker {worker_id} is completed.")
+            print(e, file=sys.stderr)
+        finally:
+            requests.task_done()
 
 
 async def main():
     start = datetime.now()
-    max_concurrent_tasks = 1000
-    task_num = 5
+    max_concurrent_tasks, task_num = 1000, 10000
+
+    requests = asyncio.Queue()
+    client = AsyncArk(timeout=24 * 3600)
+
+    # mock `task_num` tasks
+    for _ in range(task_num):
+        await requests.put(
+            {
+                "model": "${YOUR_ENDPOINT_ID}",
+                "messages": [
+                    {
+                        "role": "system",
+                        "content": "你是豆包，是由字节跳动开发的 AI 人工智能助手",
+                    },
+                    {"role": "user", "content": "常见的十字花科植物有哪些？"},
+                ],
+            }
+        )
+
+    # create `max_concurrent_tasks` workers and start them
+    tasks = [
+        asyncio.create_task(worker(i, client, requests))
+        for i in range(max_concurrent_tasks)
+    ]
+
+    # wait for all requests is done
+    await requests.join()
+
+    # stop workers
+    for task in tasks:
+        task.cancel()
 
-    # 创建任务列表
-    tasks = [worker(i, task_num) for i in range(max_concurrent_tasks)]
+    # wait for all workers is canceled
+    await asyncio.gather(*tasks, return_exceptions=True)
+    await client.close()
 
-    # 等待所有任务完成
-    await asyncio.gather(*tasks)
     end = datetime.now()
-    print(f"Total time: {end - start}, Total task: {max_concurrent_tasks * task_num}")
+    print(f"Total time: {end - start}, Total task: {task_num}")
 
 
 if __name__ == "__main__":
-    if sys.version_info >= (3, 11):
-        with asyncio.Runner(loop_factory=uvloop.new_event_loop) as runner:
-            runner.run(main())
-    else:
-        uvloop.install()
-        asyncio.run(main())
+    asyncio.run(main())

volcenginesdkexamples/volcenginesdkarkruntime/batch_chat_completions.py

@@ -1,3 +1,8 @@
+import queue
+import sys
+from datetime import datetime
+from multiprocessing.pool import ThreadPool
+
 from volcenginesdkarkruntime import Ark
 
 # Authentication
@@ -10,16 +15,74 @@
 # or specify ak&sk by Ark(ak="${YOUR_AK}", sk="${YOUR_SK}").
 # To get your ak&sk, please refer to this document(https://www.volcengine.com/docs/6291/65568)
 # For more information，please check this document（https://www.volcengine.com/docs/82379/1263279）
-client = Ark()
+
+
+def worker(
+    worker_id: int,
+    client: Ark,
+    requests: queue.Queue[dict],
+):
+    print(f"Worker {worker_id} is starting.")
+
+    while True:
+        request = requests.get()
+
+        # check for signal of no more request
+        if not request:
+            # put back on the queue for other workers
+            requests.put(request)
+            return
+
+        try:
+            # do request
+            completion = client.batch_chat.completions.create(**request)
+            print(completion)
+        except Exception as e:
+            print(e, file=sys.stderr)
+        finally:
+            requests.task_done()
+
+
+def main():
+    start = datetime.now()
+    max_concurrent_tasks, task_num = 1000, 10000
+
+    requests = queue.Queue()
+    client = Ark(timeout=24 * 3600)
+
+    # mock `task_num` tasks
+    for _ in range(task_num):
+        requests.put(
+            {
+                "model": "${YOUR_ENDPOINT_ID}",
+                "messages": [
+                    {
+                        "role": "system",
+                        "content": "你是豆包，是由字节跳动开发的 AI 人工智能助手",
+                    },
+                    {"role": "user", "content": "常见的十字花科植物有哪些？"},
+                ],
+            }
+        )
+
+    # put a signal of no more request
+    requests.put(None)
+
+    # create `max_concurrent_tasks` workers and start them
+    with ThreadPool(max_concurrent_tasks) as pool:
+        for i in range(max_concurrent_tasks):
+            pool.apply_async(worker, args=(i, client, requests))
+        pool.apply_async(worker, args=(i, client, requests))
+
+        # wait for all request to done
+        pool.close()
+        pool.join()
+
+    client.close()
+
+    end = datetime.now()
+    print(f"Total time: {end - start}, Total task: {task_num}")
+
 
 if __name__ == "__main__":
-    # Non-streaming:
-    print("----- standard request -----")
-    completion = client.batch_chat.completions.create(
-        model="${YOUR_ENDPOINT_ID}",
-        messages=[
-            {"role": "system", "content": "你是豆包，是由字节跳动开发的 AI 人工智能助手"},
-            {"role": "user", "content": "常见的十字花科植物有哪些？"},
-        ],
-    )
-    print(completion.choices[0].message.content)
+    main()