LongArticleTaskServer/app/infra/external/deepseek_volcengine.py

"""
@author: luojunhui
@description: deepseek 火山引擎 Ark API 版本 (Responses API)
"""

from typing import Any, Dict, List, Optional
from volcenginesdkarkruntime import Ark

from app.core.config import GlobalConfigSettings
from app.infra.shared.tools import safe_json_parse

config = GlobalConfigSettings()

model_map = config.deepseek_volcengine.get_model_map()
api_key = config.deepseek_volcengine.api_key
base_url = config.deepseek_volcengine.base_url
chat_model = config.deepseek_volcengine.chat_model
reasoner_model = config.deepseek_volcengine.reasoner_model


def fetch_deepseek_completion(
    model: str,
    prompt: str,
    output_type: str = "text",
) -> Optional[Dict | List]:
    input_messages = [{"role": "user", "content": prompt}]
    kwargs: dict[str, Any] = {
        "model": model_map.get(model, chat_model),
        "messages": input_messages,
    }

    client = Ark(base_url=base_url, api_key=api_key)

    try:
        response = client.chat.completions.create(**kwargs)
        output_text = response.choices[0].message.content

        if output_type == "text":
            return output_text
        elif output_type == "json":
            return safe_json_parse(output_text)
        else:
            raise ValueError(f"Invalid output_type: {output_type}")

    except Exception as e:
        print(f"[ERROR] fetch_deepseek_completion (volcengine) failed: {e}")
        return None