|
from fastapi import FastAPI, status |
|
from fastapi.responses import HTMLResponse |
|
from pydantic import BaseModel |
|
from fastapi.responses import JSONResponse, StreamingResponse |
|
import requests |
|
import json |
|
import openai |
|
import time |
|
from langchain.embeddings.openai import OpenAIEmbeddings |
|
import langchain |
|
|
|
class Text(BaseModel): |
|
content: str = "" |
|
|
|
|
|
app = FastAPI() |
|
key = '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' |
|
openai.api_key = key |
|
headers = { |
|
'Content-Type': 'application/json', |
|
'Authorization': 'Bearer ' + key |
|
} |
|
|
|
|
|
@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 = 'According to the article below, generate "question and answer" QA pairs, greater than 5, in a json format per line({“question”:"xxx","answer":"xxx"})generate:\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": "You are a useful assistant to answer questions accurately using the content of the article." |
|
} |
|
] |
|
obj = json.loads(content.content) |
|
messages.append({"role": "system", "content": "Article 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("/sale") |
|
def sale_ep(content: Text = None): |
|
url = 'https://api.openai.com/v1/chat/completions' |
|
messages = [ |
|
{ |
|
"role": "system", |
|
"content": "You are a useful assistant to answer questions accurately using the content of the article" |
|
} |
|
] |
|
obj = json.loads(content.content) |
|
messages.append({"role": "system", "content": "Article 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("/chatgpt") |
|
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) |
|
|
|
|
|
async def chat_gpt_stream_fun(content: Text = None): |
|
start_time = time.time() |
|
obj = json.loads(content.content) |
|
response = openai.ChatCompletion.create( |
|
model='gpt-3.5-turbo', |
|
messages=obj['messages'], |
|
stream=True, |
|
) |
|
|
|
collected_chunks = [] |
|
collected_messages = [] |
|
|
|
for chunk in response: |
|
chunk_time = time.time() - start_time |
|
collected_chunks.append(chunk) |
|
chunk_message = chunk['choices'][0]['delta'] |
|
collected_messages.append(chunk_message) |
|
print(f"Message received {chunk_time:.2f} seconds after request: {chunk_message}") |
|
full_reply_content = ''.join([m.get('content', '') for m in collected_messages]) |
|
print(f"Full conversation received: {full_reply_content}") |
|
content = {'content': full_reply_content} |
|
print('content:', content) |
|
yield json.dumps(content) + '\n' |
|
|
|
|
|
@app.post("/chatgptstream", status_code=status.HTTP_200_OK) |
|
async def get_random_numbers(content: Text = None): |
|
return StreamingResponse(chat_gpt_stream_fun(content), media_type='application/json') |
|
|
|
|
|
@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("/embedd") |
|
def embed(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 |
|
) |
|
embeddings = OpenAIEmbeddings(openai_api_key= key) |
|
return key |
|
|
|
|
|
@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()) |