当前位置： 首页 > news >正文

使用fastapi搭建ChatGPT对话后台

news 来源：原创 2024/10/1 9:43:09

使用fastapi搭建ChatGPT对话后台

参考资料：使用fastapi搭建ChatGPT对话后台
效果：在本地构建网页达成类似chatgpt的对话效果，一个字一个字的返回生成结果

ChatGPT初步调用

import os
import fastapi
import dotenv
from httpx import AsyncClient
from typing import List,Dict
dotenv.load_dotenv('./env')# print(os.getenv('OPENAI_API_BASE'))
async def request(val: List[dict[str,str]]):"""发起请求val: 对话内容"""url = "https://xiaoai.plus/v1/chat/completions"headers ={"Content-Type": "application/json","Authorization": "Bearer " + os.getenv("OPENAI_API_KEY")}params = {"model": "gpt-3.5-turbo","messages": val, # [{"role": "user", "content": "Say this is a test!"}]"temperature": 0.7,"n": 1,"max_tokens": 3000,"stream": False}async with AsyncClient() as clinet:response = await clinet.post(url, headers=headers,json=params,timeout=60)print(response.json())if __name__ == '__main__':import asyncioasyncio.run(request([{"role": "user", "content": "Hello!"}]))

{'id': 'chatcmpl-ABYGZNDqhtZn5igtaukBbLPWrdTPZ', 'object': 'chat.completion', 'created': 1727316691, 'model': 'gpt-3.5-turbo', 'choices': [{'index': 0, 'message': {'role': 'assistant', 'content': 'Hello! How can I assist you today?'}, 'finish_reason': 'stop'}], 'usage': {'prompt_tokens': 9, 'completion_tokens': 9, 'total_tokens': 18}, 'system_fingerprint': 'fp_808245b034'}

对回答进行解析, 这里的结果是一次性返回消息内容

{"id": "chatcmpl-ABYGZNDqhtZn5igtaukBbLPWrdTPZ",  # 唯一标识符，用于追踪请求"object": "chat.completion",  # 对象类型，表示这是一个聊天完成事件"created": 1727316691,  # 创建时间戳，表示响应创建的时间"model": "gpt-3.5-turbo",  # 使用的模型名称"choices": [  # 选择列表，可能包含多个回复，这里只有一个{"index": 0,  # 当前选择的索引"message": {  # 选择的消息内容"role": "assistant",  # 消息角色，这里是助手"content": "Hello! How can I assist you today?"  # 消息内容},"finish_reason": "stop"  # 完成原因，这里是"stop"，表示模型决定停止生成更多内容}],"usage": {  # 使用情况，包括token使用情况"prompt_tokens": 9,  # 提示token的数量"completion_tokens": 9,  # 完成token的数量"total_tokens": 18  # 总token的数量},"system_fingerprint": "fp_808245b034"  # 系统指纹，用于识别请求的系统环境
}

流式调用ChatGPT

修改上述代码中的"stream": True与print(response.text)部分

import os
import fastapi
import dotenv
from httpx import AsyncClient
from typing import List,Dict
dotenv.load_dotenv('./env')# print(os.getenv('OPENAI_API_BASE'))
async def request(val: List[dict[str,str]]):"""发起请求val: 对话内容"""url = "https://xiaoai.plus/v1/chat/completions"headers ={"Content-Type": "application/json","Authorization": "Bearer " + os.getenv("OPENAI_API_KEY")}params = {"model": "gpt-3.5-turbo","messages": val, # [{"role": "user", "content": "Say this is a test!"}]"temperature": 0.7,"n": 1,"max_tokens": 3000,"stream": True}async with AsyncClient() as clinet:response = await clinet.post(url, headers=headers,json=params,timeout=60)print(response.text)if __name__ == '__main__':import asyncioasyncio.run(request([{"role": "user", "content": "Hello!"}]))

可以看到GPT的结果是一个词一个词的返回的

data: {"id":"chatcmpl-ABYJD1L22azzx2PK9IyqCaw2RrC7J","object":"chat.completion.chunk","created":1727316855,"model":"gpt-3.5-turbo","system_fingerprint":"fp_808245b034","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":null}]}data: {"id":"chatcmpl-ABYJD1L22azzx2PK9IyqCaw2RrC7J","object":"chat.completion.chunk","created":1727316855,"model":"gpt-3.5-turbo","system_fingerprint":"fp_808245b034","choices":[{"index":0,"delta":{"content":"Hello"},"finish_reason":null}]}data: {"id":"chatcmpl-ABYJD1L22azzx2PK9IyqCaw2RrC7J","object":"chat.completion.chunk","created":1727316855,"model":"gpt-3.5-turbo","system_fingerprint":"fp_808245b034","choices":[{"index":0,"delta":{"content":"!"},"finish_reason":null}]}data: {"id":"chatcmpl-ABYJD1L22azzx2PK9IyqCaw2RrC7J","object":"chat.completion.chunk","created":1727316855,"model":"gpt-3.5-turbo","system_fingerprint":"fp_808245b034","choices":[{"index":0,"delta":{"content":" How"},"finish_reason":null}]}data: {"id":"chatcmpl-ABYJD1L22azzx2PK9IyqCaw2RrC7J","object":"chat.completion.chunk","created":1727316855,"model":"gpt-3.5-turbo","system_fingerprint":"fp_808245b034","choices":[{"index":0,"delta":{"content":" can"},"finish_reason":null}]}data: {"id":"chatcmpl-ABYJD1L22azzx2PK9IyqCaw2RrC7J","object":"chat.completion.chunk","created":1727316855,"model":"gpt-3.5-turbo","system_fingerprint":"fp_808245b034","choices":[{"index":0,"delta":{"content":" I"},"finish_reason":null}]}data: {"id":"chatcmpl-ABYJD1L22azzx2PK9IyqCaw2RrC7J","object":"chat.completion.chunk","created":1727316855,"model":"gpt-3.5-turbo","system_fingerprint":"fp_808245b034","choices":[{"index":0,"delta":{"content":" assist"},"finish_reason":null}]}data: {"id":"chatcmpl-ABYJD1L22azzx2PK9IyqCaw2RrC7J","object":"chat.completion.chunk","created":1727316855,"model":"gpt-3.5-turbo","system_fingerprint":"fp_808245b034","choices":[{"index":0,"delta":{"content":" you"},"finish_reason":null}]}data: {"id":"chatcmpl-ABYJD1L22azzx2PK9IyqCaw2RrC7J","object":"chat.completion.chunk","created":1727316855,"model":"gpt-3.5-turbo","system_fingerprint":"fp_808245b034","choices":[{"index":0,"delta":{"content":" today"},"finish_reason":null}]}data: {"id":"chatcmpl-ABYJD1L22azzx2PK9IyqCaw2RrC7J","object":"chat.completion.chunk","created":1727316855,"model":"gpt-3.5-turbo","system_fingerprint":"fp_808245b034","choices":[{"index":0,"delta":{"content":"?"},"finish_reason":null}]}data: {"id":"chatcmpl-ABYJD1L22azzx2PK9IyqCaw2RrC7J","object":"chat.completion.chunk","created":1727316855,"model":"gpt-3.5-turbo","system_fingerprint":"fp_808245b034","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}data: [DONE]

逐行处理流式响应

修改相应的代码

    # async with AsyncClient() as clinet:#     response = await clinet.post(url, headers=headers,json=params,timeout=60)#     print(response.text)async with AsyncClient() as clinet:async with clinet.stream("POST", url, headers=headers, json=params, timeout=60) as response:async for line in response.aiter_lines():print(line)

import os
import fastapi
import dotenv
from httpx import AsyncClient
from typing import List,Dict
import json
from collections import defaultdict
dotenv.load_dotenv('./env')# print(os.getenv('OPENAI_API_BASE'))
async def request(val: List[dict[str,str]]):"""发起请求val: 对话内容"""url = "https://xiaoai.plus/v1/chat/completions"headers ={"Content-Type": "application/json","Authorization": "Bearer " + os.getenv("OPENAI_API_KEY")}params = {"model": "gpt-3.5-turbo","messages": val, # [{"role": "user", "content": "Say this is a test!"}]"temperature": 0.7,"n": 1,"max_tokens": 3000,"stream": True}# async with AsyncClient() as clinet:#     response = await clinet.post(url, headers=headers,json=params,timeout=60)#     print(response.text)async with AsyncClient() as clinet:async with clinet.stream("POST", url, headers=headers, json=params, timeout=60) as response:async for line in response.aiter_lines():if line.strip() == "":continueline = line.replace("data: ","")if line.strip() == "[DONE]":returndata = json.loads(line)if data.get("choices") is None or len(data.get("choices")) == 0 or data.get("choices")[0].get("finish_reason") is not None:returnyield data.get("choices")[0]async def chat(inp: str):message = [{"role": "user", "content": inp}]async for i in request(message):print(i)
if __name__ == '__main__':import asyncioasyncio.run(chat("你好啊"))

{'index': 0, 'delta': {'role': 'assistant', 'content': ''}, 'finish_reason': None}
{'index': 0, 'delta': {'content': '你'}, 'finish_reason': None}
{'index': 0, 'delta': {'content': '好'}, 'finish_reason': None}
{'index': 0, 'delta': {'content': '，'}, 'finish_reason': None}
{'index': 0, 'delta': {'content': '有'}, 'finish_reason': None}
{'index': 0, 'delta': {'content': '什'}, 'finish_reason': None}
{'index': 0, 'delta': {'content': '么'}, 'finish_reason': None}
{'index': 0, 'delta': {'content': '可以'}, 'finish_reason': None}
{'index': 0, 'delta': {'content': '帮'}, 'finish_reason': None}
{'index': 0, 'delta': {'content': '助'}, 'finish_reason': None}
{'index': 0, 'delta': {'content': '你'}, 'finish_reason': None}
{'index': 0, 'delta': {'content': '的'}, 'finish_reason': None}
{'index': 0, 'delta': {'content': '吗'}, 'finish_reason': None}
{'index': 0, 'delta': {'content': '？'}, 'finish_reason': None}

封装请求与chat方法

import os
import fastapi
import dotenv
from httpx import AsyncClient
from typing import List,Dict
import json
from collections import defaultdict
dotenv.load_dotenv('./env')# print(os.getenv('OPENAI_API_BASE'))
async def request(val: List[dict[str,str]]):"""发起请求val: 对话内容"""url = "https://xiaoai.plus/v1/chat/completions"headers ={"Content-Type": "application/json","Authorization": "Bearer " + os.getenv("OPENAI_API_KEY")}params = {"model": "gpt-3.5-turbo","messages": val, # [{"role": "user", "content": "Say this is a test!"}]"temperature": 0.7,"n": 1,"max_tokens": 3000,"stream": True}# async with AsyncClient() as clinet:#     response = await clinet.post(url, headers=headers,json=params,timeout=60)#     print(response.text)async with AsyncClient() as clinet:async with clinet.stream("POST", url, headers=headers, json=params, timeout=60) as response:async for line in response.aiter_lines():if line.strip() == "":continueline = line.replace("data: ","")if line.strip() == "[DONE]":returndata = json.loads(line)if data.get("choices") is None or len(data.get("choices")) == 0 or data.get("choices")[0].get("delta").get("finish_reason") is not None:returnyield data.get("choices")[0]async def chat(inp: str):message = [{"role": "user", "content": inp}]chat_msg = defaultdict(str)async for i in request(message):if i.get("delta").get("role"):chat_msg["role"] = i.get("delta").get("role")if i.get("delta").get("content"):chat_msg["content"] += i.get("delta").get("content")print(chat_msg)
if __name__ == '__main__':import asyncioasyncio.run(chat("你好啊"))

defaultdict(<class 'str'>, {'role': 'assistant', 'content': '你好！有什么我可以帮助你的吗？'}

使用fastapi进行封装

import os
import fastapi
import dotenv
from httpx import AsyncClient
from typing import List,Dict
from fastapi import FastAPI, WebSocket
from fastapi.middleware.cors import CORSMiddlewareimport json
from collections import defaultdict
from fastapi.responses import HTMLResponseapp = FastAPI()
dotenv.load_dotenv('./env')# app.add_middleware(
#     CORSMiddleware,
#     allow_origins=["*"],
#     allow_credentials=True,
#     allow_methods=["*"],
#     allow_headers=["*"],
# )@app.get("/")
async def root():return {"message": "Hello World"}# print(os.getenv('OPENAI_API_BASE'))
async def request(val: List[dict[str,str]]):"""发起请求val: 对话内容"""url = "https://xiaoai.plus/v1/chat/completions"headers ={"Content-Type": "application/json","Authorization": "Bearer " + os.getenv("OPENAI_API_KEY"),}params = {"model": "gpt-3.5-turbo","messages": val, # [{"role": "user", "content": "Say this is a test!"}]"temperature": 0.7,"n": 1,"max_tokens": 3000,"stream": True}# async with AsyncClient() as clinet:#     response = await clinet.post(url, headers=headers,json=params,timeout=60)#     print(response.text)async with AsyncClient() as clinet:async with clinet.stream("POST", url, headers=headers, json=params, timeout=60) as response:async for line in response.aiter_lines():if line.strip() == "":continueline = line.replace("data: ","")if line.strip() == "[DONE]":returndata = json.loads(line)if data.get("choices") is None or len(data.get("choices")) == 0 or data.get("choices")[0].get("delta").get("finish_reason") is not None:returnyield data.get("choices")[0]@app.websocket("/chat")
async def chat(websocket: WebSocket):await websocket.accept()message = []while True:data = await websocket.receive_text()if data == "quit": await websocket.close()breakmessage.append({"role": "user", "content": data})chat_msg = defaultdict(str)async for i in request(message):if i.get("delta").get("role"):chat_msg["role"] = i.get("delta").get("role")if i.get("delta").get("content"):chat_msg["content"] += i.get("delta").get("content")await websocket.send_text(i.get("delta").get("content"))message.append(chat_msg)if __name__ == '__main__':import uvicornuvicorn.run("main:app", host="127.0.0.1",port=8080,reload=True)

使用html界面调用

<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>Titile</title>
</head>
<body><p>连接状态: <span id="status">未连接</span></p><p>回复消息: <span id="message"></span></p><p><input id="inp"></p><button type="submit" id="submit">提交</button>
<script>let status = document.getElementById("status")let message = document.getElementById("message")let inp = document.getElementById("inp")let submit = document.getElementById("submit")let socket = new WebSocket("ws://127.0.0.1:8080/chat")socket.addEventListener("open", (event)=>{status.innerText = "已连接"})socket.addEventListener("error", (event)=>{status.innerText = "已失败"})socket.addEventListener("close", (event)=>{status.innerText = "已关闭"console.log("WebSocket closed:", event)})socket.addEventListener("message", (event)=>{message.innerText += event.data})submit.addEventListener("click", ()=>{socket.send(inp.value)})
</script>
</body>
</html>

相关文章：

  • 前端规范工程-5：Git提交信息规范（commitlint + czg）
  • 企业数据安全从0到1建设方法
  • sql server每天定时执行sql语句
  • RabbitMQ 界面管理说明
  • 【Orange Pi 5嵌入式应用编程】-用户空间GPIO控制
  • Lenovo SR850服务器亮黄灯维修和升级CPU扩展模块
  • qt使用QDomDocument读写xml文件
  • SpringBoot 流式输出时，正常输出后为何突然报错？
  • 《Windows PE》3.2.3 NT头-扩展头
  • Vscode、小皮面板安装
  • 智能招聘系统小程序的设计
  • OpenCL 学习(1)---- OpenCL 基本概念
  • PGMP-03战略一致性
  • 解决docker一直出现“=＞ ERROR [internal] load metadata for docker.io/library/xxx“的问题
  • 【Kubernetes】常见面试题汇总（四十）
  • 《Javascript数据结构和算法》笔记-「字典和散列表」
  • android 一些 utils
  • angular学习第一篇-----环境搭建
  • Dubbo 整合 Pinpoint 做分布式服务请求跟踪
  • ES6语法详解(一)
  • JavaScript异步流程控制的前世今生
  • Java方法详解
  • LeetCode刷题——29. Divide Two Integers（Part 1靠自己）
  • Linux各目录及每个目录的详细介绍
  • MySQL Access denied for user 'root'@'localhost' 解决方法
  • nginx 负载服务器优化
  • OpenStack安装流程（juno版）- 添加网络服务（neutron）- controller节点
  • Python中eval与exec的使用及区别
  • SegmentFault 技术周刊 Vol.27 - Git 学习宝典：程序员走江湖必备
  • Shadow DOM 内部构造及如何构建独立组件
  • 从零到一：用Phaser.js写意地开发小游戏（Chapter 3 - 加载游戏资源）
  • 高性能JavaScript阅读简记（三）
  • 极限编程 (Extreme Programming) - 发布计划 (Release Planning)
  • 手机app有了短信验证码还有没必要有图片验证码？
  • 我是如何设计 Upload 上传组件的
  • 一个6年java程序员的工作感悟，写给还在迷茫的你
  • TPG领衔财团投资轻奢珠宝品牌APM Monaco
  • 摩拜创始人胡玮炜也彻底离开了，共享单车行业还有未来吗？ ...
  • ​Linux·i2c驱动架构​
  • !$boo在php中什么意思,php前戏
  • #laravel部署安装报错loadFactoriesFrom是undefined method #
  • #控制台大学课堂点名问题_课堂随机点名
  • （1）STL算法之遍历容器
  • (13)Hive调优——动态分区导致的小文件问题
  • （19）夹钳(用于送货)
  • (3) cmake编译多个cpp文件
  • (NO.00004)iOS实现打砖块游戏(九):游戏中小球与反弹棒的碰撞
  • (ZT) 理解系统底层的概念是多么重要(by趋势科技邹飞）
  • （黑马点评）二、短信登录功能实现
  • （企业 / 公司项目）前端使用pingyin-pro将汉字转成拼音
  • （转）visual stdio 书签功能介绍
  • (转载)微软数据挖掘算法:Microsoft 时序算法(5)
  • （自用）learnOpenGL学习总结-高级OpenGL-抗锯齿
  • .cfg\.dat\.mak（持续补充）
  • .NET 4.0网络开发入门之旅-- 我在“网” 中央（下）