VentureBeat 拓展企业级 AI 研究并任命首位首席分析师

import time
import logging
import requests

# 配置日志输出
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

class ResilientAIClient:
    def __init__(self, api_key: str, base_url: str = "https://api.n1n.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

    def generate_completion(self, prompt: str, model_fallback_chain: list, temperature: float = 0.7) -> dict:
        """
        按照优先级列表尝试调用不同的模型，直到成功为止。
        """
        for model in model_fallback_chain:
            logging.info(f"正在尝试使用模型: {model}")
            payload = {
                "model": model,
                "messages": [{"role": "user", "content": prompt}],
                "temperature": temperature
            }

            start_time = time.time()
            try:
                response = requests.post(
                    f"{self.base_url}/chat/completions",
                    json=payload,
                    headers=self.headers,
                    timeout=12.0  # 设置合理超时时间以快速触发备用方案
                )

                latency = time.time() - start_time

                if response.status_code == 200:
                    logging.info(f"成功使用 {model} 生成响应，耗时 {latency:.2f} 秒")
                    return {
                        "success": True,
                        "model_used": model,
                        "latency": latency,
                        "data": response.json()
                    }
                else:
                    logging.warning(
                        f"模型 {model} 响应失败，状态码 {response.status_code}: {response.text}"
                    )
            except requests.exceptions.RequestException as e:
                logging.error(f"请求模型 {model} 时发生网络错误或超时: {str(e)}")

            # 在尝试链中的下一个模型前稍作停顿
            time.sleep(0.5)

        raise RuntimeError("故障转移链中的所有模型均未能成功响应。")

# 调用示例
if __name__ == "__main__":
    # 请将此处替换为您在 n1n.ai 申请的 API 密钥
    N1N_API_KEY = "your_n1n_api_key_here"

    client = ResilientAIClient(api_key=N1N_API_KEY)

    # 定义候选模型链：首选 Claude 3.5，备选 GPT-4o，最后使用性价比高的 DeepSeek-V3 兜底
    models_to_try = [
        "anthropic/claude-3.5-sonnet",
        "openai/gpt-4o",
        "deepseek/deepseek-v3"
    ]

    user_prompt = "请详细对比检索增强生成（RAG）与模型微调（Fine-Tuning）的适用场景。"

    try:
        result = client.generate_completion(prompt=user_prompt, model_fallback_chain=models_to_try)
        print("\n--- 调用结果摘要 ---")
        print(f"实际使用模型: {result['model_used']}")
        print(f"接口响应延迟: {result['latency']:.2f} 秒")
        print(f"回复内容预览: {result['data']['choices'][0]['message']['content'][:150]}...")
    except Exception as error:
        print(f"业务执行失败: {str(error)}")