app.py

from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from fastapi.responses import JSONResponse
import requests
import json


class Text(BaseModel):
    content: str = ""


app = FastAPI()

headers = {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer ' + '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'
}


@app.get("/")
def home():
    html_content = open('index.html').read()
    return HTMLResponse(content=html_content, status_code=200)


@app.post("/qa_maker")
def sentiment_analysis_ep(content: Text = None):
    url = 'https://api.openai.com/v1/chat/completions'
    prompt = '根据下面的文章，生成的“问题和回答”QA对，大于5个，以一行一个json格式（{“question”:"xxx","answer":"xxx"}）生成：\n'
    messages = [{"role": "user", "content": prompt + content.content}]
    data = {
        "model": "gpt-3.5-turbo",
        "messages": messages
    }
    print("messages = \n", messages)
    result = requests.post(url=url,
                           data=json.dumps(data),
                           headers=headers
                           )
    res = str(result.json()['choices'][0]['message']['content']).strip()
    print('res:', res)
    res = {'content': res}
    return JSONResponse(content=res)


@app.post("/chatpdf")
def chat_pdf_ep(content: Text = None):
    url = 'https://api.openai.com/v1/chat/completions'
    messages = [
        {
            "role": "system",
            "content": "你是一个有用的助手，可以使用文章内容准确地回答问题。使用提供的文章来生成你的答案，但避免逐字复制文章。尽可能使用自己的话。准确、有用、简洁、清晰。"
        }
    ]
    obj = json.loads(content.content)
    messages.append({"role": "system", "content": "文章内容：\n" + obj['doc']})
    history = obj['history']
    for his in history:
        messages.append({"role": "user", "content": his[0]})
        messages.append({"role": "assistant", "content": his[1]})
    messages.append({"role": "user", "content": obj['question']})
    data = {
        "model": "gpt-3.5-turbo",
        "messages": messages
    }
    print("messages = \n", messages)
    result = requests.post(url=url,
                           data=json.dumps(data),
                           headers=headers
                           )
    res = str(result.json()['choices'][0]['message']['content']).strip()
    content = {'content': res}
    print('content:', content)
    return JSONResponse(content=content)


@app.post("/dic")
def dic_ep(content: Text = None):
    url = 'https://api.openai.com/v1/chat/completions'
    messages = [
        {
            "role": "system",
            "content": "你是小鹏汽车数据智能中心（简称DIC）的IT服务台的智能助手"
            + "根据提供的参考文章，生成文本来回答问题。答案包含参考文章内容中HTML样式的图片、网址、超链接。准确、有用、清晰。"
            + "模仿提供的参考对话来有用的回复用户，进行多轮对话。"
            + "根据参考文章和参考对话来回答用户的IT类问题，无法回答非IT类问题。"
            + "你是一个对话机器人，无法提供对话之外的服务（如远程协助、到现场解决故障等）。用户有对话之外服务的需求，请回复他：正在转人工。"
        }
    ]
    obj = json.loads(content.content)
    if 'ref_doc' in obj:
        messages.append({"role": "system", "content": "参考文章：\n" + obj['ref_doc']})
    if 'ref_dialog' in obj:
        messages.append({"role": "system", "content": "参考对话：\n" + obj['ref_dialog']})
    history = obj['history']
    for his in history:
        messages.append({"role": "user", "content": his[0]})
        messages.append({"role": "assistant", "content": his[1]})
    messages.append({"role": "user", "content": obj['question']})
    data = {
        "model": "gpt-3.5-turbo",
        "messages": messages
    }
    print("messages = \n", messages)
    result = requests.post(url=url,
                           data=json.dumps(data),
                           headers=headers
                           )
    res = str(result.json()['choices'][0]['message']['content']).strip()
    content = {'content': res}
    print('content:', content)
    return JSONResponse(content=content)


@app.post("/sale")
def sale_ep(content: Text = None):
    url = 'https://api.openai.com/v1/chat/completions'
    messages = [
        {
            "role": "system",
            "content": "你是小鹏汽车资深的销售，可以参考文章内容来生成你的答案，但避免逐字复制文章，尽可能使用自己的话，突出小鹏汽车的优点、说服用户来店试驾或购买小鹏汽车。准确、有用、有销售技巧、口语化、有亲和力。"
        }
    ]
    obj = json.loads(content.content)
    messages.append({"role": "system", "content": "文章内容：\n" + obj['doc']})
    history = obj['history']
    for his in history:
        messages.append({"role": "user", "content": his[0]})
        messages.append({"role": "assistant", "content": his[1]})
    messages.append({"role": "user", "content": obj['question']})
    data = {
        "model": "gpt-3.5-turbo",
        "messages": messages
    }
    print("messages = \n", messages)
    result = requests.post(url=url,
                           data=json.dumps(data),
                           headers=headers
                           )
    res = str(result.json()['choices'][0]['message']['content']).strip()
    content = {'content': res}
    print('content:', content)
    return JSONResponse(content=content)


@app.post("/chat_gpt")
def chat_gpt_ep(content: Text = None):
    url = 'https://api.openai.com/v1/chat/completions'
    obj = json.loads(content.content)
    data = {
        "model": "gpt-3.5-turbo",
        "messages": obj['messages']
    }
    print("data = \n", data)
    result = requests.post(url=url,
                           data=json.dumps(data),
                           headers=headers
                           )
    res = str(result.json()['choices'][0]['message']['content']).strip()
    content = {'content': res}
    print('content:', content)
    return JSONResponse(content=content)


@app.post("/embeddings")
def embeddings_ep(content: Text = None):
    url = 'https://api.openai.com/v1/embeddings'
    data = {
        "model": "text-embedding-ada-002",
        "input": content.content
    }
    result = requests.post(url=url,
                           data=json.dumps(data),
                           headers=headers
                           )
    return JSONResponse(content=result.json())


@app.post("/create_image")
def create_image_ep(content: Text = None):
    url = 'https://api.openai.com/v1/images/generations'
    obj = json.loads(content.content)
    data = {
        "prompt": obj["prompt"],
        "n": obj["n"],
        "size": obj["size"]
    }
    print("data = \n", data)
    result = requests.post(url=url,
                           data=json.dumps(data),
                           headers=headers
                           )
    return JSONResponse(content=result.json())