Chainlit接入FastGpt接口完美对接,实现全新的用户聊天界面

前言

由于fastgpt只提供了一个分享用的网页应用,网页访问地址没法自定义,虽然可以接入NextWeb/ChatGPT web等开源应用。但是如果我们想直接给客户应用,还需要客户去设置配置,里面还有很多我们不想展示给客户的东西怎么办?于是,我使用Chainlit实现了一个无缝快速接入fastgpt实现自定义用户使用界面的应用,代码清晰简单。还可以自定义logo、欢迎语、网站图标等。
之前写一个一篇文章 《Chainlit接入FastGpt接口快速实现自定义用户聊天界面》 文章中实现了Fastgpt的对话接口,可以实现聊天,但是有些人想。接入Fastgpt的欢迎语,获取聊天历史记录等。本此的代码更新可以实现这些,下面我主要讲一些这次代码更新实现的亮点吧!

1. 接入Fastgpt后,可以自动更新成你的网站图标和网站名称

在这里插入图片描述

2. 接入欢迎语

在这里插入图片描述

3. 接入聊天记录

在这里插入图片描述

  • 不仅可以实现聊天记录的自动添加还可实现删除,搜索聊天记录的功能,原fastgpt的分享聊天网页可没有聊天搜索功能呦!

4 可以开启黑白主题的切换

在这里插入图片描述

5.可以自定义修改访问路径

当然这个功能需要你自己修改启动地址

快速开始

获取Fastgpt的信息

获取fastgpt的base_url和share_id

登录fastgpt后台,在工作台里,点击自己创建的AI应用,点击发布渠道,点击免登录窗口,创建新连接,创建后点击开始使用按钮。

  • 复制base_urlshare_id,后面需要配置到Chainlit的环境变量中

在这里插入图片描述

获取fastgpt的API_KEY

登录fastgpt后台,在工作台里,点击自己创建的AI应用,点击发布渠道,点击API访问创建,访问APIKEY.

  • 只需要复制API_KEY即可,后面需要配置到Chainlit的环境变量中
    在这里插入图片描述

Chainlit网页搭建

创建一个文件夹,例如“chainlit_fastgpt”

mkdir chainlit_fastgpt

进入 chainlit_chat文件夹下,执行命令创建python 虚拟环境空间(需要提前安装好python sdkChainlit 需要python>=3.8。,具体操作,由于文章长度问题就不在叙述,自行百度),命令如下:

python -m venv .venv
  • 这一步是避免python第三方库冲突,省事版可以跳过
  • .venv是创建的虚拟空间文件夹可以自定义

接下来激活你创建虚拟空间,命令如下:

#linux or mac
source .venv/bin/activate
#windows
.venv\Scripts\activate

在项目根目录下创建requirements.txt,内容如下:

chainlit~=1.1.402
aiohttp~=3.10.5
requests~=2.32.3
literalai~=0.0.607

在项目根目录下创建app.py文件,代码如下:

import hashlib
import json
import os
from typing import Optional, Dictimport aiohttp
import chainlit as cl
import chainlit.data as cl_data
import requestsfrom fastgpt_data import FastgptDataLayer, now, share_id, app_name, welcome_textfastgpt_base_url = os.getenv("FASTGPT_BASE_URL")
fastgpt_api_key = os.getenv("FASTGPT_API_KEY")
fastgpt_api_detail = os.getenv("FASTGPT_API_DETAIL", False)cl_data._data_layer = FastgptDataLayer()
cl.config.ui.name = app_namedef download_logo():local_filename = "./public/favicon.svg"directory = os.path.dirname(local_filename)os.makedirs(directory, exist_ok=True)# Streaming, so we can iterate over the response.with requests.get(f"{fastgpt_base_url}/icon/logo.svg", stream=True) as r:r.raise_for_status()  # Check if the request was successfulwith open(local_filename, 'wb') as f:for chunk in r.iter_content(chunk_size=8192):# If you have chunk encoded response uncomment if# and set chunk_size parameter to None.f.write(chunk)download_logo()@cl.on_chat_start
async def chat_start():if welcome_text:# elements = [cl.Text(content=welcomeText, display="inline")]await cl.Message(content=welcome_text).send()@cl.on_message
async def handle_message(message: cl.Message):msg = cl.Message(content="")url = f"{fastgpt_base_url}/api/v1/chat/completions"print('message.thread_id',message.thread_id)headers = {"Authorization": f"Bearer {fastgpt_api_key}","Content-Type": "application/json"}data = {"messages": [{"role": "user","content": message.content}],"variables": {"cTime": now},"responseChatItemId": message.id,"shareId": share_id,"chatId": message.thread_id,"appType": "advanced","outLinkUid": cl.context.session.user.identifier,"detail": fastgpt_api_detail,"stream": True}async for data in fetch_sse(url, headers=headers, data=json.dumps(data), detail=fastgpt_api_detail):delta = data['choices'][0]['delta']if delta:await msg.stream_token(delta['content'])await msg.send()@cl.header_auth_callback
def header_auth_callback(headers: Dict) -> Optional[cl.User]:print(headers)# 创建一个md5 hash对象md5_hash = hashlib.md5()user_agent_bytes = headers.get('user-agent').encode('utf-8')# 更新这个hash对象的内容md5_hash.update(user_agent_bytes)# 获取md5哈希值的十六进制表示形式md5_hex_digest = md5_hash.hexdigest()out_link_uid = md5_hex_digestprint("MD5加密后的结果:", md5_hex_digest)return cl.User(identifier=out_link_uid, display_name="visitor")@cl.on_chat_resume
async def on_chat_resume():passasync def fetch_sse(url, headers, data, detail):async with aiohttp.ClientSession() as session:async with session.post(url, headers=headers, data=data) as response:async for line in response.content:if line:  # 过滤掉空行data = line.decode('utf-8').rstrip('\n\r')# print(f"Received: {data}")# 检查是否为数据行,并且是我们感兴趣的事件类型if detail:if data.startswith('event:'):event_type = data.split(':', 1)[1].strip()  # 提取事件类型elif data.startswith('data:') and event_type == 'flowNodeStatus':data = data.split(':', 1)[1].strip()flowNodeStatus = json.loads(data)current_step = cl.context.current_stepcurrent_step.name = flowNodeStatus['name']elif data.startswith('data:') and event_type == 'answer':data = data.split(':', 1)[1].strip()  # 提取数据内容# 如果数据包含换行符,可能需要进一步处理(这取决于你的具体需求)# 这里我们简单地打印出来if data != '[DONE]':yield json.loads(data)else:if data.startswith('data:'):data = data.split(':', 1)[1].strip()  # 提取数据内容# 如果数据包含换行符,可能需要进一步处理(这取决于你的具体需求)# 这里我们简单地打印出来if data != '[DONE]':yield json.loads(data)
  • 传入的model,temperature等参数字段均无效,这些字段由编排决定,不会根据 API 参数改变。

  • 不会返回实际消耗Token值,如果需要,可以设置detail=true,并手动计算 responseData 里的tokens值。

在项目根目录下创建fastgpt_data.py文件,代码如下:

import json
import os
import uuid
from typing import Optional, List, Dictimport requests
from chainlit import PersistedUser
from chainlit.data import BaseDataLayer
from chainlit.types import PageInfo, ThreadFilter, ThreadDict, Pagination, PaginatedResponse
from literalai.helper import utc_nowfastgpt_base_url = os.getenv("FASTGPT_BASE_URL")
share_id = os.getenv("FASTGPT_SHARE_ID")
now = utc_now()
user_cur_threads = []
thread_user_dict = {}def change_type(user_type: str):if user_type == 'AI':return 'assistant_message'if user_type == 'Human':return 'user_message'def get_app_info():with requests.get(f"{fastgpt_base_url}/api/core/chat/outLink/init?chatId=&shareId={share_id}&outLinkUid=123456") as resp:app = {}if resp.status_code == 200:res = json.loads(resp.content)app = res.get('data').get('app')appId = res.get('data').get('appId')app['id'] = appIdreturn appapp_info = get_app_info()app_id = app_info.get('id')
app_name = app_info.get('name')
welcome_text = app_info.get('chatConfig').get('welcomeText')def getHistories(user_id):histories = []if user_id:with requests.post(f"{fastgpt_base_url}/api/core/chat/getHistories",data={"shareId": share_id, "outLinkUid": user_id}) as resp:if resp.status_code == 200:res = json.loads(resp.content)data = res["data"]print(data)histories = [{"id": item["chatId"],"name": item["title"],"createdAt": item["updateTime"],"userId": user_id,"userIdentifier": user_id}for item in data]if user_cur_threads:thread = next((t for t in user_cur_threads if t["userId"] == user_id), None)if thread:  # 确保 thread 不为 Nonethread_id = thread.get("id")if histories:# 检查 thread 的 ID 是否已存在于 threads 中if not any(t.get("id") == thread_id for t in histories):histories.insert(0, thread)else:# 如果 threads 是空列表,则直接插入 threadhistories.insert(0, thread)for item in histories:thread_user_dict[item.get('id')] = item.get('userId')return historiesclass FastgptDataLayer(BaseDataLayer):async def get_user(self, identifier: str):print('get_user', identifier)return PersistedUser(id=identifier, createdAt=now, identifier=identifier)async def update_thread(self,thread_id: str,name: Optional[str] = None,user_id: Optional[str] = None,metadata: Optional[Dict] = None,tags: Optional[List[str]] = None,):print('---------update_thread----------',thread_id)thread = next((t for t in user_cur_threads if t["userId"] == user_id), None)if thread:if thread_id:thread["id"] = thread_idif name:thread["name"] = nameif user_id:thread["userId"] = user_idthread["userIdentifier"] = user_idif metadata:thread["metadata"] = metadataif tags:thread["tags"] = tagsthread["createdAt"] = utc_now()else:print('---------update_thread----------thread_id ', thread_id, name)user_cur_threads.append({"id": thread_id,"name": name,"metadata": metadata,"tags": tags,"createdAt": utc_now(),"userId": user_id,"userIdentifier": user_id,})async def get_thread_author(self, thread_id: str):print('get_thread_author')return thread_user_dict.get(thread_id, None)async def list_threads(self, pagination: Pagination, filters: ThreadFilter) -> PaginatedResponse[ThreadDict]:threads = []if filters:threads = getHistories(filters.userId)search = filters.search if filters.search else ""filtered_threads = [thread for thread in threads if search in thread.get('name', '')]start = 0if pagination.cursor:for i, thread in enumerate(filtered_threads):if thread["id"] == pagination.cursor:  # Find the start index using pagination.cursorstart = i + 1breakend = start + pagination.firstpaginated_threads = filtered_threads[start:end] or []has_next_page = len(paginated_threads) > endstart_cursor = paginated_threads[0]["id"] if paginated_threads else Noneend_cursor = paginated_threads[-1]["id"] if paginated_threads else Nonereturn PaginatedResponse(pageInfo=PageInfo(hasNextPage=has_next_page,startCursor=start_cursor,endCursor=end_cursor,),data=paginated_threads,)async def get_thread(self, thread_id: str):print('get_thread', thread_id)user_id = thread_user_dict.get(thread_id, None)thread = Noneif user_id:params = {'chatId': thread_id,'shareId': share_id,'outLinkUid': user_id,}with requests.get(f"{fastgpt_base_url}/api/core/chat/outLink/init",params=params,) as resp:if resp.status_code == 200:res = json.loads(resp.content)data = res["data"]if data:history = data['history']files = []texts = []for item in history:for entry in item['value']:if entry.get('type') == 'text':text = {"id": item["_id"],"threadId": thread_id,"name": item["obj"],"type": change_type(item["obj"]),"input": None,"createdAt": utc_now(),"output": entry.get('text').get('content'),}texts.append(text)if entry.get('type') == 'file':file = {"id": str(uuid.UUID),"threadId": thread_id,"forId": item["_id"],"name": entry.get('file').get('name'),"type": entry.get('file').get('type'),"url": entry.get('file').get('url'),"display": "inline","size": "medium"}files.append(file)thread = {"id": thread_id,"name": data.get("title", ''),"createdAt": utc_now(),"userId": "admin","userIdentifier": "admin","metadata": {"appId": data["appId"]},"steps": texts,"elements": files,}return threadreturn threadasync def delete_thread(self, thread_id: str):print('delete_thread')thread = next((t for t in user_cur_threads if t["id"] == thread_id), None)user_id = thread_user_dict.get(thread_id, None)if thread:user_cur_threads.remove(thread)if user_id:params = {'appId': app_id,'chatId': thread_id,'shareId': share_id,'outLinkUid': user_id,}requests.get(f"{fastgpt_base_url}/api/core/chat/delHistory",params=params)

在项目根目录下创建.env环境变量,配置如下:

CHAINLIT_AUTH_SECRET="xOIPIMBGfI7N*VK6O~KOVIRC/cGRNSmk%bmO4Q@el647hR?^mdW6=8KlQBuWWTbk"
FASTGPT_BASE_URL="https://share.fastgpt.in"
FASTGPT_API_KEY="fastgpt-key"
FASTGPT_SHARE_ID=""
FASTGPT_API_DETAIL=False
  • 项目根目录下,执行 chainlit create-secret命令,可以获得CHAINLIT_AUTH_SECRET
  • FASTGPT_BASE_URL 为你Fastgpt服务器,网页端分享地址的的base_url
  • FASTGPT_API_KEY 为你Fastgpt服务器f发布渠道中->API访问的密匙。
  • FASTGPT_SHARE_ID 为你Fastgpt服务器f发布渠道中->免登录窗口中的url的中参数shareId=后面的值。

执行以下命令安装依赖:

pip install -r .\requirements.txt
  • 安装后,项目根目录下会多出.chainlit.files文件夹和chainlit.md文件

运行应用程序

要启动 Chainlit 应用程序,请打开终端并导航到包含的目录app.py。然后运行以下命令:

 chainlit run app.py -w   
  • -w标志告知 Chainlit 启用自动重新加载,因此您无需在每次更改应用程序时重新启动服务器。您的聊天机器人 UI 现在应该可以通过http://localhost:8000访问。
  • 自定义端口可以追加--port 80

启动后界面如下:
在这里插入图片描述

  • 由于时间关系,这个应用和fastgpt的文件上传接口、语音对话还未实现,后续会在更新一次,实现完美对接!

相关文章推荐

《使用 Xinference 部署本地模型》
《Fastgpt接入Whisper本地模型实现语音输入》
《Fastgpt部署和接入使用重排模型bge-reranker》
《Fastgpt部署接入 M3E和chatglm2-m3e文本向量模型》
《Fastgpt 无法启动或启动后无法正常使用的讨论(启动失败、用户未注册等问题这里)》
《vllm推理服务兼容openai服务API》
《vLLM模型推理引擎参数大全》
《解决vllm推理框架内在开启多显卡时报错问题》
《Ollama 在本地快速部署大型语言模型,可进行定制并创建属于您自己的模型》

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

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

相关文章

[数据集][目标检测]街灯路灯检测数据集VOC+YOLO格式1893张1类别

数据集格式:Pascal VOC格式YOLO格式(不包含分割路径的txt文件,仅仅包含jpg图片以及对应的VOC格式xml文件和yolo格式txt文件) 图片数量(jpg文件个数):1893 标注数量(xml文件个数):1893 标注数量(txt文件个数):1893 标注…

数据结构(Java实现):链表习题

文章目录 1. 题目列表及链接2. 题目解析及代码2.1 删除链表中等于给定值 val 的所有节点2.2 反转一个单链表2.3 给定一个带有头结点 head 的非空单链表,返回链表的中间结点。如果有两个中间结点,则返回第二个中间结点2.4 输入一个链表,输出该…

iTransformer时序模型改进——基于SENet和TCN的倒置Transformer,性能暴涨

1数据集介绍 ETT(电变压器温度):由两个小时级数据集(ETTh)和两个 15 分钟级数据集(ETTm)组成。它们中的每一个都包含 2016 年 7 月至 2018 年 7 月的七种石油和电力变压器的负载特征。 数据集链接: https…

深度学习入门-第4章-神经网络的学习

学习就是从训练数据中自动获取最优权重参数的过程。引入损失函数这一指标,学习的目的是找出使损失函数达到最小的权重参数。使用函数斜率的梯度法来找这个最小值。 人工智能有两派,一派认为实现人工智能必须用逻辑和符号系统,自顶向下看问题…

Sass实现网页背景主题切换

Sass 实现网页背景主题切换 前言准备工作一、 简单的两种主题黑白切换1.定义主题2. 添加主题切换功能3. 修改 data-theme 属性 二、多种主题切换1. 定义主题2. 动态生成 CSS 变量1.遍历列表2.遍历映射3.高级用法 3. 设置默认主题4. 切换功能HTML 三、多种主题多种样式切换1. 定…

Java数组的定义与使用

目录 1. 数组的基本概念 1.1为什么要使用数组 1.2 什么是数组 1.3 数组的创建及初始化 1.3.1 数组的创建 1.3.2 数组的初始化 1. 动态初始化 2. 静态初始化 1.4 数组的使用 1.4.1 数组中元素访问 1.4.2 遍历数组 2. 数组是引用类型 2.1 基本类型变量与引用类型变量…

【C++从小白到大牛】C++智能指针的使用、原理和分类

目录 1、我们为什么需要智能指针? 2、内存泄露 2.1 什么是内存泄漏,内存泄漏的危害 2.2如何避免内存泄漏 总结一下: 3.智能指针的使用及原理 3.1 RAII 3.2关于深拷贝和浅拷贝更深层次的理解: 3.3 std::auto_ptr 3.4 std::unique_pt…

Springboot里集成Mybatis-plus、ClickHouse

🌹作者主页:青花锁 🌹简介:Java领域优质创作者🏆、Java微服务架构公号作者😄 🌹简历模板、学习资料、面试题库、技术互助 🌹文末获取联系方式 📝 Springboot里集成Mybati…

Overleaf参考文献由 BibTex 转 \bibitem 格式

目录 Overleaf参考文献由 BibTex 转 \bibitem 格式一、获取引用论文的BibTex二、编写引用论文对应的bib文件三、编写生成bibitem的tex文件四、转化bibitem格式 参考资料 Overleaf参考文献由 BibTex 转 \bibitem 格式 一、获取引用论文的BibTex 搜索论文引用点击BibTex 跳转出…

怎样快速搭建 Linux 虚拟机呢?(vagrant 篇)

作为一名Coder(程序员或码农),供职于中小型互联网公司,而你恰恰偏向于服务端,那么,产品部署在生产环境的艰巨任务,便毫无疑问的落在你身上了。 只有大厂(大型互联网)企业…

Ps:首选项 - 界面

Ps菜单:编辑/首选项 Edit/Preferences 快捷键:Ctrl K Photoshop 首选项中的“界面” Interface选项卡可以定制 Photoshop 的界面外观和行为,从而创建一个最适合自己工作习惯和需求的工作环境。这些设置有助于提高工作效率,减轻眼…

Simple RPC - 07 从零开始设计一个服务端(下)_RPC服务的实现

文章目录 PreRPC服务实现服务注册请求处理 设计: 请求分发机制 Pre Simple RPC - 01 框架原理及总体架构初探 Simple RPC - 02 通用高性能序列化和反序列化设计与实现 Simple RPC - 03 借助Netty实现异步网络通信 Simple RPC - 04 从零开始设计一个客户端&#…

# 利刃出鞘_Tomcat 核心原理解析(九)-- Tomcat 安全

利刃出鞘_Tomcat 核心原理解析(九)-- Tomcat 安全 一、Tomcat专题 - Tomcat安全 - 配置安全 1、 删除 tomcat 的 webapps 目录下的所有文件,禁用 tomcat 管理界面. 如下目录均可删除: D:\java-test\apache-tomcat-8.5.42-wind…

数据结构-KMP算法

先解决 前缀与后缀串的最长匹配长度信息(前缀或后缀都不能取整体)。如下 位置6的前缀最长串就是abcab(不能取全部,即不能为abcabc) 位置6的后缀最长串就是bcabc(不能取全部,即不能为abcabc)

[Linux#47][网络] 网络协议 | TCP/IP模型 | 以太网通信

目录 1.网络协议 2.协议分层 2.1 OSI七层模型 2.2TCP/IP五层(四层)模型 2.3 以太网通信 1.网络协议 "协议"本质就是一种约定 计算机之间的传输媒介是光信号和电信号. 通过 "频率" 和 "强弱" 来表示 0 和 1 这样的 信息. 要想传递各种不同…

HTML+CSS浮动和清除浮动的效果及其应用场景举例

一、清除浮动的效果 解释 .container:用于展示浮动和清除浮动效果的容器,具有边框和背景色以便于区分。 .float-box:浮动元素,用不同的背景色标识。 .clearfix:使用伪元素清除浮动的类,应用于第二个容器。 …

IDEA 2024.2.0.2 使用 Jrebel and XRebel 热部署

安装 激活 工具网站中url和邮箱复制进去 设置 允许项目自动构建 允许开发过程中自动部署

python面向对象—封装、继承、多态

封装 ① 把现实世界中的主体中的属性和方法书写到类的里面的操作即为封装 ② 封装可以为属性和方法添加为私有权限,不能直接被外部访问 在面向对象代码中,我们可以把属性和方法分为两大类:公有(属性、方法)、私有&…

SQLSugar进阶使用:高级查询与性能优化

文章目录 前言一、高级查询1.查所有2.查询总数3.按条件查询4.动态OR查询5.查前几条6.设置新表名7.分页查询8.排序 OrderBy9.联表查询10.动态表达式11.原生 Sql 操作 ,Sql和存储过程 二、性能优化1.二级缓存2.批量操作3.异步操作4.分表组件,自动分表5.查询…

LCP:60 排列序列[leetcode-4]

LCP:60 排列序列 给出集合 [1,2,3,...,n],其所有元素共有 n! 种排列。 按大小顺序列出所有排列情况,并一一标记,当 n 3 时, 所有排列如下: "123""132""213""231""312"&quo…