使用fastapi搭建ChatGPT对话后台

🕗 发布于 2024-09-27 08:12 fastapi chatgpt

使用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 asyncio
    asyncio.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 asyncio
    asyncio.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() == "":
                    continue
                line = line.replace("data: ","")
                if line.strip() == "[DONE]":
                    return
                data = 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:
                    return
                yield 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 asyncio
    asyncio.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() == "":
                    continue
                line = line.replace("data: ","")
                if line.strip() == "[DONE]":
                    return
                data = 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:
                    return
                yield 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 asyncio
    asyncio.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 CORSMiddleware

import json
from collections import defaultdict
from fastapi.responses import HTMLResponse

app = 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() == "":
                    continue
                line = line.replace("data: ","")
                if line.strip() == "[DONE]":
                    return
                data = 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:
                    return
                yield 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()
            break
        message.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 uvicorn
    uvicorn.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>

原文地址：https://blog.csdn.net/Weary_PJ/article/details/142548728

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：【pytorch】pytorch入门4：神经网络的卷积层
下一篇：每天一个数据分析题（四百八十八）- 非监督学习

从 ES5 到 ES14：深入解析 JavaScript 的演进与特性
本文概述了 JavaScript 从 ES5 到 ES14 的主要特性演进。ES5 引入了严格模式和新的数组方法，提升了代码的安全性和可读性；ES6 带来了 let、const、箭头函数和模块系统等核
阅读更多2024-09-28
为什么不用tensorflow而用opencv
使用 OpenCV 而不是 TensorFlow 进行图像处理和计算机视觉任务的原因取决于特定需求和场景。
阅读更多2024-09-28
“AI大语言模型+”助力大气科学相关交叉领域实践技术应用
GPT生成转化代码，将数据转化为WPS可读取的二进制格式。4.2模型堆叠：使用mlxtend库或自定义方法实现模型堆叠，结合不同模型的预测结果作为新的特征，训练一个新的模型。2.3交叉验证分割：采用交
阅读更多2024-09-28
Apache Solr：深入探索与常见误区解析
Solr 是一个功能丰富且高度可扩展的搜索引擎，适合大规模的数据搜索应用。无论是电商平台的商品搜索、日志分析还是内容管理系统的文章检索，Solr 都能提供高效的解决方案。但是，在使用 Solr 时，配
阅读更多2024-09-28
DataLight（V1.4.5）版本更新，新增 Ranger、Solr
本次更新对平台进行了较多的功能拓展和优化，新增了对 Ranger 和 Solr 服务组件的支持，同时对多项已有功能进行了改进，旨在提升更好的使用体验。进入主节点，通过以下命令启动 Master 进程，
阅读更多2024-09-28
从零开始Ubuntu24.04上Docker构建自动化部署（三）Docker安装Nginx
宿主机上拷贝docker上nginx服务上文件到本地目录。删除docker上nginx容器。停止docker 上nignx。重启启动nginx容器。
阅读更多2024-09-28
付费计量系统通用过程
值得注意的是，虽然每个过程类在普通意义上显示单一，但实际系统的实现通常都有属于同类的几个事例过程。参考条目6和图3，普通。一个过程定义描述的是数据单元或一个事件序列的流程，每一个在系统中都是被定义了的
阅读更多2024-09-28
STM32原理知识查询表
2024.09.27单片机的两种看门狗原理解析
阅读更多2024-09-28
鸿蒙开发（NEXT/API 12）【硬件（常见问题）】驱动开发服务
编译或运行时报错：“usb/usb_ddk_api.h not found”, "hid/hid_ddk_api.h not found" 等
阅读更多2024-09-28
Cilium + ebpf 系列文章- （六）Cilium-BGP与分发-EXTERNAL-IP
CiliumBGPPeerConfig配置对等连接的行为，可以引用一个或多个CiliumBGPAdvertisement资源(使用选择器)1、可以使用CiliumLoadBalancerIPPool来
阅读更多2024-09-28

使用fastapi搭建ChatGPT对话后台

使用fastapi搭建ChatGPT对话后台

ChatGPT初步调用

流式调用ChatGPT

逐行处理流式响应

封装请求与chat方法

使用fastapi进行封装

相关文章