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

devtools/2024/10/18 5:55:53/

前言

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

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

在这里插入图片描述

2. 接入欢迎语

在这里插入图片描述

3. 接入聊天记录

在这里插入图片描述

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

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

在这里插入图片描述

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

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

快速开始

获取Fastgpt的信息

fastgptbase_urlshare_id_20">获取fastgpt的base_url和share_id

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

  • 复制base_url和share_id，后面需要配置到Chainlit的环境变量中

在这里插入图片描述

fastgptAPI_KEY_25">获取fastgpt的API_KEY

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

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

Chainlit网页搭建

chainlit_fastgpt_32">创建一个文件夹，例如“chainlit_fastgpt”

mkdir chainlit_fastgpt

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

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

接下来激活你创建虚拟空间，命令如下：

#linux or mac
uiltin class-name">source .venv/bin/activate
#windows
.venv\Scripts\activate

uirementstxt_56">在项目根目录下创建requirements.txt，内容如下：

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

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

python">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 uiltin">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.uiltin">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_datapy_202">在项目根目录下创建fastgpt_data.py文件，代码如下：

python">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: uiltin">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 = uiltin">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 uiltin">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: uiltin">str):print('get_user', identifier)return PersistedUser(uiltin">id=identifier, createdAt=now, identifier=identifier)async def update_thread(self,thread_id: uiltin">str,name: Optional[uiltin">str] = None,user_id: Optional[uiltin">str] = None,metadata: Optional[Dict] = None,tags: Optional[List[uiltin">str]] = None,):print('---------update_thread----------',thread_id)thread = uiltin">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: uiltin">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 uiltin">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 = uiltin">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: uiltin">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':uiltin">file = {"id": uiltin">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(uiltin">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: uiltin">str):print('delete_thread')thread = uiltin">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.ppmy.cn/devtools/100062.html

相关文章

Tomcat使用及负载均衡（最全源码安装及配置使用教程）

Tomcat使用及负载均衡（最全源码安装及配置使用教程）

目录 一 Tomcat概述 1.1 Tomcat 简介 1.2 Tomcat 下载 二 Tomcat 单主机配置 2.1 Tomcat 环境配置 2.2 Tomcat 安装与添加系统启动 2.3 Tomcat 启动与停止 三 Tomcat 配置文件及反向代理 3.1 配置文件详解 3.2 反向代理实现Tomcat部署 四 Memcached安装 4.1 简介 …
阅读更多...
Apache Commons-IO 库

Apache Commons-IO 库

Apache Commons-IO是Apache开源基金组织提供的一组有关IO（Input/Output）操作的小框架。这个库的主要目的是为了提高IO流的开发效率，减少在进行文件读写、目录遍历等操作时编写的样板代码量。通过使用Commons-IO库，开发者可以更加简…
阅读更多...
每日Attention学习16——Multi-layer Multi-scale Dilated Convolution

每日Attention学习16——Multi-layer Multi-scale Dilated Convolution

模块出处 [CBM 22] [link] [code] Do You Need Sharpened Details? Asking MMDC-Net: Multi-layer Multi-scale Dilated Convolution Network For Retinal Vessel Segmentation 模块名称 Multi-layer Multi-scale Dilated Convolution (MMDC) 模块作用 多尺度特征提取与融合…
阅读更多...
EXCEL 分段排序--Excel难题#86

EXCEL 分段排序--Excel难题#86

Excel某表格有3列。 ABC1A1B1512A2B27213A3B33824A4B495A5B5736A6B65777A7B7918A13B131509A14B144910A17B1770211A18B1870512A34B343313A35B3540914A36B3657915A37B3710 现在要求对表格按照第3列进行分段排序，由小到大排列。第1段：第3列小于等于50&…
阅读更多...
Python3：多行文本内容转换为标准的cURL请求参数值

Python3：多行文本内容转换为标准的cURL请求参数值

背景 在最近的工作中，经常需要处理一些接口请求的参数，参数来源形式很多，可能是Excel、知识库文档等，有些数据形式比较复杂，比如多行或者包含很多不同的字符，示例如下： **客服质检分析指引** …
阅读更多...
RPA自动化流程机器人助力企业财务数字化转型

RPA自动化流程机器人助力企业财务数字化转型

在数字经济时代，企业需要快速响应市场变化，而财务数字化转型是企业适应现代商业环境、提升竞争力的必要步骤。财务数字化转型不仅涉及企业财务能力的提升，推动了财务管理与决策模式的转变。RPA自动化流程机器人因其能通过自动化技术帮助企业实…
阅读更多...
如何从单点到集群配置Eureka：最佳实践与配置详解

如何从单点到集群配置Eureka：最佳实践与配置详解

如何从单点到集群配置Eureka：最佳实践与配置详解 1. 引言 什么是Eureka？ Eureka是Netflix开源的服务发现和注册中心，广泛应用于微服务架构中。它允许各个服务实例在启动时注册到Eureka Server，并通过Eureka Client获取其他服务…
阅读更多...
前端开发工程师面试整理-ES6+的新特性

前端开发工程师面试整理-ES6+的新特性

ES6（ECMAScript 2015）及后续版本引入了许多新特性，极大地增强了JavaScript的功能和开发体验。以下是一些主要的新特性： 变量声明 1. let 和 const: ● let 声明块作用域变量。 ● const 声明常量，值不能重新赋值。 ● 示例：
阅读更多...
最新文章

Copyright @ 2022~2023