使用 FastAPI 实现聊天完成 API 详解

使用 FastAPI 实现聊天完成 API 详解

    • 简介
    • 基础概念
      • FastAPI
      • Pydantic
      • PyTorch
    • 代码详解
      • 1. 定义 API 端点
      • 2. 请求验证
      • 3. 生成参数字典
      • 4. 处理流式响应
      • 5. 工具调用处理
      • 6. 非流式响应处理
      • 7. 处理使用信息和工具调用
      • 8. 构建聊天消息
      • 9. 构建响应选择
      • 10. 更新使用信息
      • 11. 返回最终响应
    • 总结
      • 示例代码

简介

在这篇博客中，我们将详细解释一段使用 FastAPI 构建的聊天完成 API 代码。这段代码实现了一个 POST 请求的 API 端点，用于处理聊天消息并生成响应。我们将逐行解析代码，并提供必要的背景知识和示例代码。

基础概念

FastAPI

FastAPI 是一个用于构建 API 的现代、快速（高性能）的 Web 框架，基于 Python 3.6+。它使用类型提示来自动生成文档和验证请求数据。

Pydantic

Pydantic 是一个用于数据验证和设置管理的库，常与 FastAPI 一起使用。它通过 Python 的类型提示来定义数据模型，并自动验证输入数据的类型和格式。

PyTorch

PyTorch 是一个开源的深度学习框架，广泛用于研究和生产环境。它提供了灵活的张量计算和自动求导功能。

代码详解

下面是完整的代码，我们将逐段进行解释。

@app.post("/v1/chat/completions", response_model=ChatCompletionResponse)
async def create_chat_completion(request: ChatCompletionRequest):if len(request.messages) < 1 or request.messages[-1].role == "assistant":raise HTTPException(status_code=400, detail="Invalid request")gen_params = dict(messages=request.messages,temperature=request.temperature,top_p=request.top_p,max_tokens=request.max_tokens or 1024,echo=False,stream=request.stream,repetition_penalty=request.repetition_penalty,tools=request.tools,tool_choice=request.tool_choice,)logger.debug(f"==== request ====\n{gen_params}")if request.stream:predict_stream_generator = predict_stream(request.model, gen_params)output = await anext(predict_stream_generator)if output:return EventSourceResponse(predict_stream_generator, media_type="text/event-stream")logger.debug(f"First result output：\n{output}")function_call = Noneif output and request.tools:try:function_call = process_response(output, request.tools, use_tool=True)except:logger.warning("Failed to parse tool call")if isinstance(function_call, dict):function_call = ChoiceDeltaToolCallFunction(**function_call)generate = parse_output_text(request.model, output, function_call=function_call)return EventSourceResponse(generate, media_type="text/event-stream")else:return EventSourceResponse(predict_stream_generator, media_type="text/event-stream")response = ""async for response in generate_stream_glm4(gen_params):passif response["text"].startswith("\n"):response["text"] = response["text"][1:]response["text"] = response["text"].strip()usage = UsageInfo()function_call, finish_reason = None, "stop"tool_calls = Noneif request.tools:try:function_call = process_response(response["text"], request.tools, use_tool=True)except Exception as e:logger.warning(f"Failed to parse tool call: {e}")if isinstance(function_call, dict):finish_reason = "tool_calls"function_call_response = ChoiceDeltaToolCallFunction(**function_call)function_call_instance = FunctionCall(name=function_call_response.name,arguments=function_call_response.arguments)tool_calls = [ChatCompletionMessageToolCall(id=generate_id('call_', 24),function=function_call_instance,type="function")]message = ChatMessage(role="assistant",content=None if tool_calls else response["text"],function_call=None,tool_calls=tool_calls,)logger.debug(f"==== message ====\n{message}")choice_data = ChatCompletionResponseChoice(index=0,message=message,finish_reason=finish_reason,)task_usage = UsageInfo.model_validate(response["usage"])for usage_key, usage_value in task_usage.model_dump().items():setattr(usage, usage_key, getattr(usage, usage_key) + usage_value)return ChatCompletionResponse(model=request.model,choices=[choice_data],object="chat.completion",usage=usage)

1. 定义 API 端点

@app.post("/v1/chat/completions", response_model=ChatCompletionResponse)
async def create_chat_completion(request: ChatCompletionRequest):

这行代码定义了一个 POST 请求的 API 端点 /v1/chat/completions，并指定了请求的响应模型为 ChatCompletionResponse。create_chat_completion 函数将处理传入的 ChatCompletionRequest 请求。

2. 请求验证

    if len(request.messages) < 1 or request.messages[-1].role == "assistant":raise HTTPException(status_code=400, detail="Invalid request")

这里我们进行请求验证：

  • 检查 messages 列表的长度是否小于 1。
  • 检查最后一条消息的角色是否为 “assistant”。

如果以上任一条件为真，则抛出 HTTP 400 错误。

3. 生成参数字典

    gen_params = dict(messages=request.messages,temperature=request.temperature,top_p=request.top_p,max_tokens=request.max_tokens or 1024,echo=False,stream=request.stream,repetition_penalty=request.repetition_penalty,tools=request.tools,tool_choice=request.tool_choice,)logger.debug(f"==== request ====\n{gen_params}")

将请求中的参数转化为一个字典 gen_params，用于后续的生成操作。同时，记录调试信息。

4. 处理流式响应

    if request.stream:predict_stream_generator = predict_stream(request.model, gen_params)output = await anext(predict_stream_generator)if output:return EventSourceResponse(predict_stream_generator, media_type="text/event-stream")logger.debug(f"First result output：\n{output}")

如果请求中指定了流式响应，则调用 predict_stream 函数生成流式响应生成器，并返回 EventSourceResponse。如果第一个输出存在，则直接返回生成器作为事件流响应。

5. 工具调用处理

        function_call = Noneif output and request.tools:try:function_call = process_response(output, request.tools, use_tool=True)except:logger.warning("Failed to parse tool call")if isinstance(function_call, dict):function_call = ChoiceDeltaToolCallFunction(**function_call)generate = parse_output_text(request.model, output, function_call=function_call)return EventSourceResponse(generate, media_type="text/event-stream")else:return EventSourceResponse(predict_stream_generator, media_type="text/event-stream")

如果输出存在且请求包含工具调用，则尝试解析工具调用。如果解析成功，则处理工具调用并生成新的事件流响应；否则，继续返回原始的事件流生成器。

6. 非流式响应处理

    response = ""async for response in generate_stream_glm4(gen_params):passif response["text"].startswith("\n"):response["text"] = response["text"][1:]response["text"] = response["text"].strip()

如果请求未指定流式响应，则调用 generate_stream_glm4 生成响应。在生成响应后，去掉开头的换行符并修剪两端空白。

7. 处理使用信息和工具调用

    usage = UsageInfo()function_call, finish_reason = None, "stop"tool_calls = Noneif request.tools:try:function_call = process_response(response["text"], request.tools, use_tool=True)except Exception as e:logger.warning(f"Failed to parse tool call: {e}")if isinstance(function_call, dict):finish_reason = "tool_calls"function_call_response = ChoiceDeltaToolCallFunction(**function_call)function_call_instance = FunctionCall(name=function_call_response.name,arguments=function_call_response.arguments)tool_calls = [ChatCompletionMessageToolCall(id=generate_id('call_', 24),function=function_call_instance,type="function")]

在处理响应后，创建 UsageInfo 实例并检查是否有工具调用。如果有工具调用，则解析并生成工具调用响应。

8. 构建聊天消息

    message = ChatMessage(role="assistant",content=None if tool_calls else response["text"],function_call=None,tool_calls=tool_calls,)

根据生成的响应和工具调用信息，创建一个 ChatMessage 实例。

9. 构建响应选择

    logger.debug(f"==== message ====\n{message}")choice_data = ChatCompletionResponseChoice(index=0,message=message,finish_reason=finish_reason,)

这段代码将生成的 ChatMessage 实例记录到日志中，并且创建一个 ChatCompletionResponseChoice 实例，其中包含了消息的索引、消息内容和完成原因。

10. 更新使用信息

    task_usage = UsageInfo.model_validate(response["usage"])for usage_key, usage_value in task_usage.model_dump().items():setattr(usage, usage_key, getattr(usage, usage_key) + usage_value)

从响应中提取使用信息，并将其添加到 usage 实例中。UsageInfo.model_validate 方法用于验证并创建一个包含使用信息的实例。

11. 返回最终响应

    return ChatCompletionResponse(model=request.model,choices=[choice_data],object="chat.completion",usage=usage)

最后，创建并返回一个 ChatCompletionResponse 实例，其中包含了模型名称、选项列表和使用信息。

总结

通过这篇博客，我们详细解析了一个基于 FastAPI 实现的聊天完成 API 的代码。我们逐行解释了代码的功能，并介绍了相关的基础概念和库。

示例代码

为了帮助理解，我们提供一个简化版的示例代码，用于实现类似的聊天完成 API：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Listapp = FastAPI()class ChatMessage(BaseModel):role: strcontent: strclass ChatCompletionRequest(BaseModel):messages: List[ChatMessage]temperature: floatmax_tokens: intclass ChatCompletionResponse(BaseModel):message: ChatMessage@app.post("/v1/chat/completions", response_model=ChatCompletionResponse)
async def create_chat_completion(request: ChatCompletionRequest):if len(request.messages) < 1 or request.messages[-1].role == "assistant":raise HTTPException(status_code=400, detail="Invalid request")# Simplified response generation logicresponse_text = "This is a response."response_message = ChatMessage(role="assistant", content=response_text)return ChatCompletionResponse(message=response_message)

这段简化代码定义了一个基本的聊天完成 API 端点，处理请求并返回简单的响应。通过这个示例，可以更好地理解完整代码的工作原理。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/pingmian/37981.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

SQL Server触发器深度解析：数据完整性的守护者

SQL Server触发器深度解析：数据完整性的守护者

标题&#xff1a;SQL Server触发器深度解析&#xff1a;数据完整性的守护者 摘要 在SQL Server中&#xff0c;触发器是一种特殊的存储过程&#xff0c;它在特定数据库事件发生时自动执行。触发器主要用于维护数据的完整性和实施复杂的业务规则。本文将详细介绍SQL Server中触…
阅读更多...
ubuntu 添加PATH

ubuntu 添加PATH

在Ubuntu中&#xff0c;PATH是一个环境变量&#xff0c;用于指定系统查找可执行文件的目录列表。如果你想将新的目录添加到PATH中&#xff0c;可以按照以下步骤操作&#xff1a; 临时添加PATH 你可以在终端中使用export命令临时修改PATH环境变量。例如&#xff0c;如果你想将…
阅读更多...
Python使用彩虹表来尝试对MD5哈希进行破解

Python使用彩虹表来尝试对MD5哈希进行破解

MD5是一种散列算法&#xff0c;它是不可逆的&#xff0c;无法直接解密。它的主要作用是将输入数据进行散列&#xff0c;生成一个固定长度的唯一哈希值。 然而&#xff0c;可以使用预先计算好的MD5哈希值的彩虹表&#xff08;Rainbow Table&#xff09;来尝试对MD5进行破解。彩…
阅读更多...
c++函数（一）习题讲解

c++函数（一）习题讲解

1.【单选题】 void swap (int a,int b) { a a ^ b; b a ^ b; a a ^ b; } int a {120},b {130}; swap{a,b}; a? b? A 120,130 B 130,120 C130,0 解析&#xff1a;这道题中&#xff0c;函数体的内容是交换两个变量的值&#xff0c;采用的是位运算的…
阅读更多...
Java中泛型的概念和使用场景

Java中泛型的概念和使用场景

技术难点 Java中的泛型&#xff08;Generics&#xff09;是JDK 5.0引入的一项新特性&#xff0c;它允许在定义类、接口和方法时使用类型参数&#xff08;type parameters&#xff09;。泛型的主要技术难点在于类型擦除&#xff08;type erasure&#xff09;和类型推断&#xf…
阅读更多...
Day48

Day48

Day48 手写Spring-MVC之前后置处理器与异常处理 前后置处理器 概念&#xff1a;从服务器获取的JSON数据可能是加密后的&#xff0c;因此服务端获取的时候需要进行解密&#xff08;前置处理器&#xff09;。 而从服务器传出的JSON数据可能需要加密&#xff0c;因此需要在处理返…
阅读更多...
VMware虚拟机迁移：兼用性踩坑和复盘

VMware虚拟机迁移：兼用性踩坑和复盘

文章目录 方法失败情况分析&#xff1a;参考文档 方法 虚拟机关机&#xff0c;整个文件夹压缩后拷贝到新机器中&#xff0c;开机启用即可 成功的情况&#xff1a; Mac (intel i5) -> Mac (intel i7)Mac (intel, MacOS - VMware Fusion) -> DELL (intel, Windows - VMw…
阅读更多...
Zynq7000系列FPGA中的DMA控制器简介（二）

Zynq7000系列FPGA中的DMA控制器简介（二）

AXI互连上的DMA传输 所有DMA事务都使用AXI接口在PL中的片上存储器、DDR存储器和从外设之间传递数据。PL中的从设备通过DMAC的外部请求接口与DMAC通信&#xff0c;以控制数据流。这意味着从设备可以请求DMA交易&#xff0c;以便将数据从源地址传输到目标地址。 虽然DMAC在技术…
阅读更多...
mysql5.7安装使用

mysql5.7安装使用

mysql5.7安装包&#xff1a;百度网盘 提取码: 0000 一、 安装步骤 双击安装文件 选择我接受许可条款–Next 选择自定义安装&#xff0c;下一步 选择电脑对应的系统版本后(我的系统是64位)&#xff0c;点击中间的右箭头&#xff0c;选择Next 选择安装路径–Next 执行…
阅读更多...
.NET之C#编程：懒汉模式的终结，单例模式的正确打开方式

.NET之C#编程：懒汉模式的终结，单例模式的正确打开方式

概述 在C#编程世界中&#xff0c;单例模式是一种常见的设计模式&#xff0c;用于确保一个类只有一个实例&#xff0c;并提供一个全局访问点。然而&#xff0c;传统的懒汉模式实现方式在多线程环境下存在安全隐患。本文将深入探讨单例模式的正确实现姿势&#xff0c;带你走出懒汉…
阅读更多...
matlab可以把图像数据转换为小波分析吗

matlab可以把图像数据转换为小波分析吗

&#x1f3c6;本文收录于《CSDN问答解答》专栏&#xff0c;主要记录项目实战过程中的Bug之前因后果及提供真实有效的解决方案&#xff0c;希望能够助你一臂之力&#xff0c;帮你早日登顶实现财富自由&#x1f680;&#xff1b;同时&#xff0c;欢迎大家关注&&收藏&…
阅读更多...
【后端面试题】【中间件】【NoSQL】ElasticSearch 节点角色、写入数据过程、Translog和索引与分片

【后端面试题】【中间件】【NoSQL】ElasticSearch 节点角色、写入数据过程、Translog和索引与分片

中间件的常考方向&#xff1a; 中间件如何做到高可用和高性能的&#xff1f; 你在实践中怎么做的高可用和高性能的&#xff1f; Elasticsearch节点角色 Elasticsearch的节点可以分为很多种角色&#xff0c;并且一个节点可以扮演多种角色&#xff0c;下面列举几种主要的&…
阅读更多...
【软件测试】白盒测试（知识点 + 习题 + 答案）

【软件测试】白盒测试（知识点 + 习题 + 答案）

《 软件测试基础持续更新中》 最近大家总是催更……&#xff0c;我也是百忙之中给大家详细总结了白盒测试的重点内容&#xff01; 知识点题型答案&#xff0c;让你用最短的时间&#xff0c;学到最高效的知识&#xff01; 整理不易&#xff0c;求个三连 ₍ᐢ..ᐢ₎ ♡ 目录 一、…
阅读更多...
Spring专题一：源码编译

Spring专题一：源码编译

下载源码 因为公司使用的是Spring5.2.x所以就下载了这个版本&#xff0c;github源码地址如下&#xff1a; GitHub - spring-projects/spring-framework at v5.2.6.RELEASE&#xff1a; 如果网络不稳定可以使用下载压缩版即可&#xff0c;网络稳定的话还是建议使用git clone …
阅读更多...
JDBC中的元数据是什么？如何获取？

JDBC中的元数据是什么？如何获取？

JDBC中的元数据&#xff08;MetaData&#xff09;是关于数据的数据&#xff0c;它描述了数据库的结构、表的结构、列的数据类型、存储过程、支持的SQL语法和数据库产品的版本等信息。元数据在JDBC中主要通过DatabaseMetaData和ResultSetMetaData两个接口来获取。 1. DatabaseM…
阅读更多...
代理IP用什麼協議？

代理IP用什麼協議？

代理IP的運作主要依賴兩種協議&#xff1a;HTTP代理協議和SOCKS代理協議。 HTTP代理協議 HTTP代理協議是最常見的代理協議。它主要用於HTTP請求&#xff0c;即我們常說的網頁流覽。當你通過HTTP代理伺服器訪問網站時&#xff0c;你的設備會先向代理伺服器發送HTTP請求&#x…
阅读更多...
【redis】redis RDB

【redis】redis RDB

1、概述 1.1定义 RDB (Redis Database) 是 Redis 的默认持久化机制&#xff0c;它能够在指定的时间间隔内将内存中的数据集快照写入磁盘。RDB 持久化产生的文件是一个经过压缩的二进制文件&#xff0c;通过该文件可以还原生成 RDB 文件时的数据库状态。 1.2特点 一次性全量备…
阅读更多...
RStudio学习笔记（三）：其他数据结构

RStudio学习笔记（三）：其他数据结构

1、矩阵 在R语言中&#xff0c;矩阵是二维的&#xff0c;包括行和列&#xff0c;其中分为数值型、字符型、逻辑型三种&#xff0c;在每个矩阵中的矩阵元素的类型必须一致&#xff0c;可以通过matrix函数创建矩阵。 m <- matrix(1:20, nrow 4, ncol 5) # 创建一个四行…
阅读更多...
【工具分享】SQLmap

【工具分享】SQLmap

文章目录 工具介绍安装方式环境准备安装 sqlmap 工具介绍 sqlmap 是一个非常强大的自动化 SQL 注入工具&#xff0c;主要用于渗透测试和安全审计。它能够检测和利用 SQL 注入漏洞&#xff0c;进而访问数据库服务器。 GitHub&#xff1a;https://github.com/sqlmapproject/sql…
阅读更多...
JS：防抖与节流函数的实现与应用

JS：防抖与节流函数的实现与应用

一、防抖 防抖&#xff1a;一连串操作只执行一次。通过设置一个延迟时间&#xff0c;如果这段时间没有再次执行操作则运行目标函数&#xff0c;否则重新计时。 下面是一个防抖函数代码&#xff1a; let a1; const add () > {a; }; const debounce (fun: Function, dela…
阅读更多...
最新文章

Copyright @ 2022~2023