File size: 2,267 Bytes
a4b70d9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 |
from __future__ import annotations
from ...typing import Messages, AsyncResult
from ...errors import MissingAuthError
from ..template import OpenaiTemplate
from .qwenContentGenerator import QwenContentGenerator
from .qwenOAuth2 import QwenOAuth2Client
from .sharedTokenManager import TokenManagerError
class QwenCode(OpenaiTemplate):
label = "Qwen Code 🤖"
url = "https://qwen.ai"
login_url = "https://github.com/QwenLM/qwen-code"
working = True
needs_auth = True
active_by_default = True
default_model = "qwen3-coder-plus"
models = [default_model]
client = QwenContentGenerator(QwenOAuth2Client())
@classmethod
def get_models(cls, **kwargs):
if cls.live == 0:
cls.client.shared_manager.checkAndReloadIfNeeded()
creds = cls.client.shared_manager.getCurrentCredentials()
if creds:
cls.client.shared_manager.isTokenValid(creds)
cls.live += 1
return cls.models
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
api_key: str = None,
api_base: str = None,
**kwargs
) -> AsyncResult:
try:
creds = await cls.client.get_valid_token()
last_chunk = None
async for chunk in super().create_async_generator(
model,
messages,
api_key=creds.get("token", api_key),
api_base=creds.get("endpoint", api_base),
**kwargs
):
if chunk != last_chunk:
yield chunk
last_chunk = chunk
except TokenManagerError:
await cls.client.shared_manager.getValidCredentials(cls.client.qwen_client, True)
creds = await cls.client.get_valid_token()
last_chunk = None
async for chunk in super().create_async_generator(
model,
messages,
api_key=creds.get("token"),
api_base=creds.get("endpoint"),
**kwargs
):
if chunk != last_chunk:
yield chunk
last_chunk = chunk
except:
raise |