89 lines
2.5 KiB
Python
89 lines
2.5 KiB
Python
from collections.abc import Mapping
|
|
|
|
try:
|
|
from .qwen_api_models import (
|
|
QWEN_API_MULTIMODAL_ENDPOINT_MODEL_IDS,
|
|
QWEN_API_THINKING_CONTROL_MODEL_IDS,
|
|
normalize_api_model_name,
|
|
)
|
|
except ImportError:
|
|
from qwen_api_models import (
|
|
QWEN_API_MULTIMODAL_ENDPOINT_MODEL_IDS,
|
|
QWEN_API_THINKING_CONTROL_MODEL_IDS,
|
|
normalize_api_model_name,
|
|
)
|
|
|
|
|
|
def _get_field(value, name, default=None):
|
|
if isinstance(value, Mapping):
|
|
return value.get(name, default)
|
|
return getattr(value, name, default)
|
|
|
|
|
|
def _to_multimodal_messages(messages):
|
|
normalized = []
|
|
for message in messages:
|
|
item = dict(message)
|
|
content = item.get("content")
|
|
if item.get("role") != "system" and isinstance(content, str):
|
|
item["content"] = [{"text": content}]
|
|
normalized.append(item)
|
|
return normalized
|
|
|
|
|
|
def call_qwen_api(
|
|
dashscope_module,
|
|
*,
|
|
api_key,
|
|
model,
|
|
messages,
|
|
response_format=None,
|
|
force_multimodal=False,
|
|
):
|
|
model = normalize_api_model_name(model)
|
|
use_multimodal = (
|
|
force_multimodal or model in QWEN_API_MULTIMODAL_ENDPOINT_MODEL_IDS
|
|
)
|
|
call_kwargs = {
|
|
"api_key": api_key,
|
|
"model": model,
|
|
"messages": (
|
|
_to_multimodal_messages(messages) if use_multimodal else messages
|
|
),
|
|
"result_format": "message",
|
|
"response_format": response_format,
|
|
}
|
|
if model in QWEN_API_THINKING_CONTROL_MODEL_IDS:
|
|
call_kwargs["enable_thinking"] = False
|
|
|
|
endpoint = (
|
|
dashscope_module.MultiModalConversation
|
|
if use_multimodal
|
|
else dashscope_module.Generation
|
|
)
|
|
return endpoint.call(**call_kwargs)
|
|
|
|
|
|
def extract_qwen_api_response_text(response):
|
|
output = _get_field(response, "output")
|
|
choices = _get_field(output, "choices")
|
|
if choices:
|
|
message = _get_field(choices[0], "message")
|
|
content = _get_field(message, "content")
|
|
if isinstance(content, str):
|
|
return content
|
|
if isinstance(content, list):
|
|
text_parts = []
|
|
for part in content:
|
|
text = _get_field(part, "text")
|
|
if isinstance(text, str):
|
|
text_parts.append(text)
|
|
if text_parts:
|
|
return "".join(text_parts)
|
|
|
|
text = _get_field(output, "text")
|
|
if isinstance(text, str):
|
|
return text
|
|
|
|
raise ValueError("Qwen API response did not contain text output")
|