File size: 2,541 Bytes
0eb03b6 | 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 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | #!/opt/homebrew/bin/python3.12
import os
import sys
from qwen_api import Qwen
from qwen_api.client import QwenAPIError
from qwen_api.core.exceptions import AuthError, RateLimitError
from qwen_api.core.types.chat import ChatMessage, ChatResponse, MessageRole
LLM_LOGIN_USER = os.getenv('LLM_LOGIN_USER','')
LLM_LOGIN_PASSWORD = os.getenv('LLM_LOGIN_PASSWORD','')
LLM_MODEL = os.getenv('LLM_MODEL','qwen-max-latest')
def qwen_chat(prompt: str) -> tuple[bool, str]:
response_status = False
try:
# Initialize client
# Full configuration
client = Qwen(
email=LLM_LOGIN_USER,
password=LLM_LOGIN_PASSWORD,
# api_key = token, # API authentication token
# api_key = QWEN_AUTH_TOKEN, # API authentication token
# cookie = cookie, # Authentication cookie
base_url = "https://chat.qwen.ai", # API base URL
timeout = 600, # Request timeout in seconds
log_level = "ERROR", # Logging level
save_logs = False # Whether to save logs to file
)
# Create a simple chat message
messages = [ChatMessage(
role = MessageRole.USER,
content = prompt,
web_search = False,
thinking = False
)]
# Get response
response: ChatResponse = client.chat.create(
messages=messages,
model=LLM_MODEL, # type: ignore
#model="qwen3-coder-30b-a3b-instruct", # qwen3 coder flash
stream=True
) # type: ignore
response_status = True
content = ""
# Process the stream
for chunk in response:
delta = chunk.choices[0].delta # type: ignore
content += delta.content
return response_status, content
except AuthError as e:
print(f"Authentication failed: {e}")
except RateLimitError as e:
print(f"Rate limit exceeded: {e}")
except QwenAPIError as e:
print(f"API error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")
return response_status, "Error"
if __name__ == "__main__":
if (len(sys.argv) <= 1):
print("No message provided")
sys.exit(1)
role = ""
prompt = role + ' '.join(sys.argv[1:])
response_ok, result = qwen_chat(prompt)
if response_ok:
print(result)
|