代码语言：python

所属分类：其他

代码描述：python搭建一个http同步的阿里百炼DASHSCOPE的所有模型api集合代码，将阿里百炼中主要的模型（文生图、聊天、图片编辑、图生视频、数字人克隆，口型同步等）异步请求api转成http的同步api。

代码标签： python 搭建 http 同步 阿里 百炼 DASHSCOPE 所有 模型 api 集合 代码

下面为部分代码预览，完整代码请点击下载或在bfwstudio webide中打开

import os
import time
import json
import logging
import hashlib
import threading
import requests
import asyncio
from typing import Optional, List, Any, Dict, Union
from fastapi import FastAPI, HTTPException, Body
from fastapi.concurrency import run_in_threadpool
from pydantic import BaseModel, Field

# ================= 配置区域 =================
# 请设置你的 API KEY
API_KEY = os.getenv("DASHSCOPE_API_KEY", "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx") 
api_key = API_KEY # 兼容旧变量名

# 用于计算 authstring 的盐值
AUTH_SALT = "your_secret_salt_123456" 

# 全局任务锁（简单的内存锁，生产环境建议用 Redis）
TASK_CACHE = {}
TASK_LOCK = threading.Lock()

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

app = FastAPI(title="DashScope AI All-in-One API")

# ==========================================
# 1. 辅助函数 & 原始业务逻辑 (Service Layer)
# ==========================================
# 注意：为了让代码在一个文件中运行，这里包含所有原始逻辑的封装

def compute_auth_hash(input_str: str) -> str:
    """计算认证哈希: MD5(input + salt)"""
    if not input_str:
        input_str = "default"
    raw = f"{input_str}{AUTH_SALT}"
    return hashlib.md5(raw.encode('utf-8')).hexdigest()

# --- 原始 SDK/API 调用函数 (直接集成) ---

# [聊天]
def aiask(prompt, history=[]):
    url = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
    messages = [{"role": "system", "content": "You are a helpful assistant."}]
    # 简单的 history 适配，实际使用需更严谨处理
    messages.extend(history)
    messages.append({"role": "user", "content": prompt})
    
    body = {
        "model": "qwen-plus", 
        "messages": messages, 
        "max_tokens": 1000, 
        "stream": False
    }
    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"}
    try:
        response = requests.post(url, headers=headers, json=body)
        if response.status_code != 200: return True, response.text
        data = response.json()
        if 'choices' in data and len(data['choices']) > 0:
            return False, data['choices'][0]['message']['content']
        return True, str(data)
    except Exception as e:
        return True, str(e)

def vlaiask(prompt, usersay, imgurl, history=[]):
    url = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
    messages = [{"role": "system", "content": prompt}]
    messages.extend(history)
    messages.append({
        "role": "user",
        "content": [
            {"type": "text", "text": usersay},
            {"type": "image_url", "imag.........完整代码请登录后点击上方下载按钮下载查看

网友评论0

取消

评论提交中...

提交评论