app250106.py

import json
import uuid  # 用于生成随机用户 ID

import requests
from flask import Flask, request, Response

app = Flask(__name__)
# 全局字典用于保存用户的上下文对话
user_contexts = {}
MAX_HISTORY_LENGTH = 15  # 最大上下文历史长度


def get_models():
    models = {
        "object": "list",
        "data": [
            {"id": "Qwen2.5-72B", "object": "model", "created": 0, "owned_by": "Qwen"},
            {"id": "Llama3.3-70B", "object": "model", "created": 0, "owned_by": "Nemotron"},
            {"id": "Pixtral-124B", "object": "model", "created": 0, "owned_by": "Pixtral"},
            {"id": "Qwen2.5-Coder-32B", "object": "model", "created": 0, "owned_by": "Qwen"},
        ]
    }
    return json.dumps(models)


# 新的方法，检查model是否有效并返回id
def get_model_by_id(model_id=None):
    # 获取所有模型数据
    models_data = json.loads(get_models())["data"]
    
    # 提取所有有效的模型ID
    valid_ids = [model["id"] for model in models_data]
    
    # 如果model_id无效，默认返回Qwen2.5-72B
    if model_id not in valid_ids:
        model_id = "Llama3.3-70B"
    
    # 根据model_id获取对应的模型数据
    model_data = next((model for model in models_data if model["id"] == model_id), None)
    
    # 返回model_data的id字段
    return model_data["id"] if model_data else None

def chat_completion(
        user_prompt, user_id: str = None, system_prompt="You are a helpful assistant.", model="Llama3.3-70B",
        project="DecentralGPT", stream=False, temperature=0.3, max_tokens=1024, top_p=0.5,
        frequency_penalty=0, presence_penalty=0):
    """处理用户请求并保留上下文"""
    url = 'https://usa-chat.degpt.ai/api/v0/chat/completion/proxy'
    headers = {
        'sec-ch-ua-platform': '"macOS"',
        'Referer': 'https://www.degpt.ai/',
        'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36',
        'sec-ch-ua': 'Google Chrome";v="131", "Chromium";v="131", "Not_A Brand";v="24"',
        'DNT': '1',
        'Content-Type': 'application/json',
        'sec-ch-ua-mobile': '?0'
    }

    # 初始化或更新用户的对话历史
    if user_id is not None:
        if user_id not in user_contexts:
            user_contexts[user_id] = [{"role": "system", "content": system_prompt}]

        user_contexts[user_id].append({"role": "user", "content": user_prompt})

        # 检查是否需要修剪历史记录，保留 `system` 提示词
        while len(user_contexts[user_id]) > MAX_HISTORY_LENGTH:
            # 删除最早的用户问题和系统回复，但保留 `system` 提示词
            if len(user_contexts[user_id]) > 2:
                # 检查删除的条目是否有匹配的系统回复，如果没有，只删除用户输入
                if user_contexts[user_id][2]["role"] == "user":
                    user_contexts[user_id] = [user_contexts[user_id][0]] + user_contexts[user_id][2:]
                else:
                    user_contexts[user_id] = [user_contexts[user_id][0]] + user_contexts[user_id][2:]
            else:
                break

        messages = user_contexts[user_id]
    else:
        # 如果没有提供 user_id，不保留上下文
        messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}]

    payload = {
        "model": get_model_by_id(model),
        "messages": messages,
        "project": project,
        "stream": stream,
        "temperature": temperature,
        "max_tokens": max_tokens,
        "top_p": top_p,
        "frequency_penalty": frequency_penalty,
        "presence_penalty": presence_penalty
    }

    try:
        response = requests.post(url, headers=headers, json=payload)
        response.encoding = 'utf-8'
        response.raise_for_status()
        ## print(response.text)

        # 获取响应并添加到上下文
        response_content = response.json()["choices"][0]["message"]["content"]
        # print(
        #     f"=========== {user_id}:{user_prompt} ====================\r\n请求内容:{messages}\r\n完整响应:{response.text}")

        # 将系统的回复添加到用户上下文中
        if user_id is not None:
            user_contexts[user_id].append({"role": "assistant", "content": response_content})
        return response.text
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return "请求失败，请检查网络或参数配置。"
    except (KeyError, IndexError) as e:
        print(f"解析响应时出错: {e}")
        return "解析响应内容失败。"
    return {}

def chat_completion_messages(
        messages, user_id: str = None, system_prompt="You are a helpful assistant.", model="Qwen2.5-72B",
        project="DecentralGPT", stream=False, temperature=0.3, max_tokens=1024, top_p=0.5,
        frequency_penalty=0, presence_penalty=0):
    """处理用户请求并保留上下文"""
    url = 'https://usa-chat.degpt.ai/api/v0/chat/completion/proxy'
    headers = {
        'sec-ch-ua-platform': '"macOS"',
        'Referer': 'https://www.degpt.ai/',
        'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36',
        'sec-ch-ua': 'Google Chrome";v="131", "Chromium";v="131", "Not_A Brand";v="24"',
        'DNT': '1',
        'Content-Type': 'application/json',
        'sec-ch-ua-mobile': '?0'
    }

    # # 初始化或更新用户的对话历史
    # if user_id is not None:
    #     if user_id not in user_contexts:
    #         user_contexts[user_id] = [{"role": "system", "content": system_prompt}]

    #     user_contexts[user_id].append({"role": "user", "content": user_prompt})

    #     # 检查是否需要修剪历史记录，保留 `system` 提示词
    #     while len(user_contexts[user_id]) > MAX_HISTORY_LENGTH:
    #         # 删除最早的用户问题和系统回复，但保留 `system` 提示词
    #         if len(user_contexts[user_id]) > 2:
    #             # 检查删除的条目是否有匹配的系统回复，如果没有，只删除用户输入
    #             if user_contexts[user_id][2]["role"] == "user":
    #                 user_contexts[user_id] = [user_contexts[user_id][0]] + user_contexts[user_id][2:]
    #             else:
    #                 user_contexts[user_id] = [user_contexts[user_id][0]] + user_contexts[user_id][2:]
    #         else:
    #             break

    #     messages = user_contexts[user_id]
    # else:
    #     # 如果没有提供 user_id，不保留上下文
    #     messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}]

    payload = {
        "model": get_model_by_id(model),
        "messages": messages,
        "project": project,
        "stream": stream,
        "temperature": temperature,
        "max_tokens": max_tokens,
        "top_p": top_p,
        "frequency_penalty": frequency_penalty,
        "presence_penalty": presence_penalty
    }

    try:
        response = requests.post(url, headers=headers, json=payload)
        response.encoding = 'utf-8'
        response.raise_for_status()
        ## print(response.text)

        # # 获取响应并添加到上下文
        # response_content = response.json()["choices"][0]["message"]["content"]
        # # print(
        # #     f"=========== {user_id}:{user_prompt} ====================\r\n请求内容:{messages}\r\n完整响应:{response.text}")

        # # 将系统的回复添加到用户上下文中
        # if user_id is not None:
        #     user_contexts[user_id].append({"role": "assistant", "content": response_content})
        return response.text
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return "请求失败，请检查网络或参数配置。"
    except (KeyError, IndexError) as e:
        print(f"解析响应时出错: {e}")
        return "解析响应内容失败。"
    return {}
@app.route('/api/models', methods=['GET'])
@app.route('/api/v1/models', methods=['GET'])
@app.route('/hf/v1/models', methods=['GET'])
def models():
    """返回可用模型列表"""
    return get_models()


@app.route('/api/chat/completion', methods=['POST'])
@app.route('/api/v1/chat/completions', methods=['POST'])
@app.route('/hf/v1/chat/completions', methods=['POST'])
def chat_completion_api():
    """处理用户请求并保留上下文"""
    data = request.json
    # user_prompt = data.get("prompt")
    user_id = data.get("user_id", str(uuid.uuid4()))  # 如果未提供 user_id，生成随机值
    model=get_model_by_id(data.get("model"))
    temperature=data.get("temperature", 0.3)
    max_tokens =data.get("max_tokens", 1024)
    messages = data.get("messages", [])
    stream = data.get("stream", False)
    response_content = chat_completion_messages(
        messages=messages,
        user_id=user_id,
        temperature=temperature,
        max_tokens=max_tokens,
        stream=stream,
        model=model
    )

    # maybe \uxxxx
    # return jsonify(response_content)
    ## maybe \"xxx\"
    # return Response(
    #     json.dumps(response_content, ensure_ascii=False),
    #     content_type="application/json; charset=utf-8"
    # )
    # support Chinese
    if isinstance(response_content, str):  # 如果已经是 JSON 字符串
        return Response(response_content, content_type="application/json; charset=utf-8")


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=7860)