siliconflow_api.py

import requests
import logging
from typing import Optional

class SiliconflowAPIWrapper:
    """封装硅基流动API的调用逻辑"""
    BASE_URL = "https://api.siliconflow.cn/v1/chat/completions"

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

    def generate_content(self, model: str, prompt: str, system_instruction: Optional[str],
                         temperature: float, max_tokens: int, top_p: float,
                         top_k: int, frequency_penalty: float, min_p: float) -> str:
        if not self.api_key:
            return "错误：未设置硅基流动API密钥。"

        messages = []
        if system_instruction and system_instruction.strip():
            messages.append({"role": "system", "content": system_instruction.strip()})
        messages.append({"role": "user", "content": prompt})

        data = {
            "model": model,
            "messages": messages,
            "stream": False,
            "max_tokens": max_tokens,
            "min_p": min_p,
            "temperature": temperature,
            "top_p": top_p,
            "top_k": top_k,
            "frequency_penalty": frequency_penalty,
        }

        try:
            response = requests.post(self.BASE_URL, headers=self.headers, json=data)
            response.raise_for_status()
            result = response.json()
            
            if "choices" in result and len(result["choices"]) > 0:
                return result["choices"][0]["message"]["content"]
            else:
                logging.error(f"硅基流动API返回异常数据: {result}")
                return "错误：API返回结果格式异常。"

        except requests.exceptions.RequestException as e:
            return f"错误：调用硅基流动API失败 - {str(e)}"