python库

• openai
• flask
• dataclasses
• typing

废话不多说直接上干货

# 引入相关库
import datetime
import json
import logging
from logging.handlers import RotatingFileHandler
from openai import OpenAI
from flask import Flask, request, jsonify, Response
from dataclasses import dataclass, asdict, is_dataclass
from typing import List, Optional# 定义数据类
@dataclass
class ChatCompletionMessage:content: strrole: strfunction_call: Optional[str] = Nonetool_calls: Optional[str] = None@dataclass
class Choice:finish_reason: strindex: intmessage: ChatCompletionMessagelogprobs: Optional[str] = None@dataclass
class CompletionUsage:completion_tokens: intprompt_tokens: inttotal_tokens: intdef custom_to_dict(obj):"""Recursively convert an object to a dictionary."""if is_dataclass(obj):return asdict(obj)elif hasattr(obj, '__dict__'):return {key: custom_to_dict(value) for key, value in obj.__dict__.items()}elif isinstance(obj, list):return [custom_to_dict(item) for item in obj]elif isinstance(obj, dict):return {key: custom_to_dict(value) for key, value in obj.items()}else:return obj# openai相关的接口，主要是对接chatgpt等等的
@app.route('/openai/', methods=['POST', 'GET', 'PUT'], defaults={'subpath': ''})
@app.route('/openai/<path:subpath>', methods=['POST', 'GET', 'PUT'])
def openai_proxy(subpath):# 从请求中获取参数data = request.json# data['stream'] 若存在则赋予本来的值，若不存在则为Falsedata['stream'] = data.get('stream', False)data['top_p'] = data.get('top_p', 0.7)data['temperature'] = data.get('temperature', 0.9)data['max_tokens'] = data.get('max_tokens', 2048)try:client = OpenAI(api_key="sk-xxxxxxxxxx",base_url="https://api.openai.com")if data['stream'] == True:response = client.chat.completions.create(model=data['model'],messages=data['messages'],messages=data['messages'],top_p=data['top_p'],temperature=data['temperature'],stream=True,max_tokens=data['max_tokens'])def generate():for chunk in response:chunk_dict = {'id': chunk.id,'created': chunk.created,'model': chunk.model,'object': chunk.object,'choices': [{'delta': {'content': choice.delta.content,'role': choice.delta.role,# 添加其他delta属性},'finish_reason': choice.finish_reason,'index': choice.index,# 添加其他choice属性}for choice in chunk.choices],# 添加其他chunk属性}# 将字典转换为JSON字符串chunk_json_str = json.dumps(chunk_dict)yield f"data: {chunk_json_str}".encode('utf-8')return Response(generate(), mimetype='text/event-stream')else:response = client.chat.completions.create(model=data['model'],messages=data['messages'],top_p=data['top_p'],temperature=data['temperature'],max_tokens=data['max_tokens'])# 将响应对象手动转换为字典response_dict = custom_to_dict(response)# 转换为 JSON 格式的字符串json_result = json.dumps(response_dict, ensure_ascii=False, indent=4)return jsonify(json.loads(json_result))except Exception as e:print(e)return jsonify({"error": str(e)})# GLM相关接口
@app.route('/glm', methods=['POST', 'GET', 'PUT'], defaults={'subpath': ''})
@app.route('/glm/<path:subpath>', methods=['POST', 'GET', 'PUT'])
def glm_proxy(subpath):# 从请求中获取参数data = request.jsonprint(data['messages'])# data['stream'] 若存在则赋予本来的值，若不存在则为Falsedata['stream'] = data.get('stream', False)data['top_p'] = data.get('top_p', 0.7)data['temperature'] = data.get('temperature', 0.9)data['max_tokens'] = data.get('max_tokens', 2048)try:client = OpenAI(api_key="xxxxxxx", base_url="https://open.bigmodel.cn/api/paas/v4/")if data['stream'] == True:response = client.chat.completions.create(model="GLM-4-air",messages=data['messages'],top_p=data['top_p'],temperature=data['temperature'],stream=True,max_tokens=data['max_tokens'])def generate():for chunk in response:chunk_dict = {'id': chunk.id,'created': chunk.created,'model': chunk.model,'object': chunk.object,'choices': [{'delta': {'content': choice.delta.content,'role': choice.delta.role,# 添加其他delta属性},'finish_reason': choice.finish_reason,'index': choice.index,# 添加其他choice属性}for choice in chunk.choices],# 添加其他chunk属性}# 将字典转换为JSON字符串chunk_json_str = json.dumps(chunk_dict)yield f"data: {chunk_json_str}".encode('utf-8')return Response(generate(), mimetype='text/event-stream')else:response = client.chat.completions.create(model=data["model"],messages=data['messages'],top_p=data['top_p'],temperature=data['temperature'],max_tokens=data['max_tokens'])# 将响应对象手动转换为字典response_dict = custom_to_dict(response)# 转换为 JSON 格式的字符串json_result = json.dumps(response_dict, ensure_ascii=False, indent=4)print(json_result)return jsonify(json.loads(json_result))except Exception as e:print(e)return jsonify({"error": str(e)})
# ....
# 至于其他的我就不一一展示了，大同小异，我写的比较简单，当然中间也能加其他逻辑# 运行 Flask 服务
if __name__ == '__main__':log_handler = RotatingFileHandler('app.log', maxBytes=10000, backupCount=1)log_handler.setLevel(logging.INFO)app.logger.addHandler(log_handler)app.logger.setLevel(logging.INFO)app.run(host="127.0.0.1", port=8000)