LongArticleTaskServer/applications/api/deep_seek_official_api.py

"""
@author: luojunhui
@description: deepseek 官方api
"""

import json

from typing import Dict, List, Optional
from openai import OpenAI

from applications.config import deep_seek_official_model
from applications.config import deep_seek_official_api_key


def fetch_deepseek_completion(
    model: str,
    prompt: str,
    output_type: str = "text",
    tool_calls: bool = False,
    tools: List[Dict] = None,
) -> Optional[Dict]:
    messages = [{"role": "user", "content": prompt}]
    kwargs = {
        "model": deep_seek_official_model.get(model, "deepseek-chat"),
        "messages": messages,
    }

    # add tool calls
    if tool_calls and tools:
        kwargs["tools"] = tools
        kwargs["tool_choice"] = "auto"

    client = OpenAI(
        api_key=deep_seek_official_api_key, base_url="https://api.deepseek.com"
    )

    if output_type == "json":
        kwargs["response_format"] = {"type": "json_object"}

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

        if output_type == "text":
            return choice.message.content  # 只返回文本
        elif output_type == "json":
            return json.loads(choice.message.content)
        else:
            raise ValueError(f"Invalid output_type: {output_type}")

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