打造个人知识库-chatwithrtx接口研究-编程知识

前言

之前安装了chatwithrtx，确实挺好用的。但是如果想用其对外提供服务的话，还需要研究是否能够提供api接口进行调用，所以今天来进行一下研究。

gradio介绍

web的访问是通过gradio框架进行开发的。在user_interface.py中可以发现如下引用

import gradio as gr

Gradio 是一个用于快速创建和共享机器学习模型的Web界面的Python库。它允许开发者和研究人员轻松地为他们的模型创建交互式的、可分享的Web应用，无需编写前端代码。通过Gradio，用户可以通过简单的图形界面与机器学习模型进行交互，比如上传图片进行分类、输入文本进行翻译或情感分析、或者使用其他输入数据类型。

Gradio 的主要特点包括：
1、易于使用：只需几行代码，就可以为模型创建交互式界面。
2、支持多种输入和输出类型：包括文本、图片、音频、视频等。
3、可分享：Gradio 应用可以生成URL，方便与他人分享或嵌入到网站中。
4、集成：可以轻松集成到Jupyter笔记本中，也支持与主流的机器学习库（如Ten5、sorFlow、PyTorch）和数据科学库（如pandas）一起使用。
6、自动文档：Gradio 可以自动生成接口的使用文档。
7、Gradio 被广泛用于机器学习、数据科学项目的原型设计、演示和评估阶段，让非技术用户也能轻松地理解和使用复杂的模型。

前端调试

通过F12调试，我们发现如下结论

问答结果推送

问答结果，是以event-stream方式进行推送，如下图

为json的方式进行返回

{"msg": "process_completed","event_id": "a0ae8b3a9deb4b21a40c4d7c2beba5fe","output": {"data": [[["你好","你好！我很高兴能为您提供帮助。如何可以帮助您？"]],null],"is_generating": true,"duration": 0.031991243362426758,"average_duration": 0.05873284415294059},"success": true
}

text/event-stream是一种MIME类型，用于HTTP服务器推送技术，这项技术通常称为Server-Sent Events（SSE）。它允许一个web服务器实时地向浏览器推送事件或消息。这种通信方式是单向的：从服务器到客户端（浏览器），而不需要客户端周期性地向服务器发起请求来检查更新。

服务器设置：在服务器端，响应的Content-Type设置为text/event-stream，然后通过一个持续开放的HTTP连接发送数据。
Server-Sent Events适用于需要实时数据更新的应用，比如实时新闻报道、社交媒体更新、在线交易信息、或是游戏得分板等。与WebSocket相比，SSE只支持单向通信，但它更简单，且在某些只需要服务器到客户端单向数据流的场景下，SSE是一个更轻量级的选择。

查询接口

查询接口如下图，url地址为

http://127.0.0.1:38504/queue/join?__theme=dark

请求的数据格式

{"data": [[["你好",null]],null],"event_data": null,"fn_index": 40,"trigger_id": 45,"session_hash": "w0s2r1jsq5g"
}

代码解读

由于gradio框架不太熟悉，所以只能一边看一边理解。可能理解不太到位。

监听端口

监听的端口号每次都不一样，都是随机生成的，代码如下，从1024-49000范围随机产生。

def _get_free_port(self):# Create a socket objectsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)# Set a short timeout for the connection attemptsock.settimeout(1)port = Nonewhile port is None:port = random.randint(1024, 49000)try:# Attempt to bind to the portsock.bind(("127.0.0.1", port))except OSError as e:port = Noneif e.errno != 98:  # errno 98: Address already in useprint('OS error', e)break sock.close()return port

安全校验

在启动之后，会生成一个cookie

def _open_app(self, port):def launch_thread(cookie):launch_url = f'http://127.0.0.1:{port}?cookie={cookie}&__theme=dark'print(f'Open {launch_url} in browser to start Chat with RTX')webbrowser.open(launch_url)return Noneself._secure_cookie = str(uuid.uuid4())threading.Thread(target=launch_thread, args=(self._secure_cookie,)).start()return None

安全校验如下，我相信把这一段注释掉应该就没有安全校验了

def _validate_request(self, request: gr.Request):headers = request.headerssession_key = Noneif 'cookie' in headers:cookies = headers['cookie']if '_s_chat_=' in cookies:cookies = cookies.split('; ')for i, cookie in enumerate(cookies):key, value = cookie.split('=')if key == '_s_chat_':session_key = valueif session_key == None or session_key != self._secure_cookie:raise 'session validation failed'return True

发送命令的时候添加了一个_s_chat_ 的字段

问答交互

问题主要有两个控件构成，一个是Chat with RTX 一个是SEND。

我们在_render_chatbot函数中可以找到。

def _render_chatbot(self, show_chatbot):chatbot_window = gr.Chatbot(show_label=False,elem_classes="chat-window",visible=show_chatbot,elem_id="main-chatbot-window",sanitize_html=False)with gr.Group() as query_group:with gr.Row():query_input = gr.Textbox(placeholder="Chat with RTX...", scale=9, container=False)submit_button = gr.Button("SEND", variant="primary", scale=1)

在def _handle_chatbot_events(self):函数中实现处理发送事件函数。

gr.on([self._chat_query_input_textbox.submit, self._chat_submit_button.click],self._validate_session,None,self._get_validate_session_output()).then(self._validate_session_and_raise,None,None).success(self._show_hide_sample_questions,self._get_show_hide_sample_questions_inputs(),self._get_show_hide_sample_questions_outputs()).then(process_input,[self._chat_query_input_textbox, self._chat_bot_window], [self._chat_query_input_textbox, self._chat_bot_window]).then(process_output,[self._chat_bot_window, self._state],[self._chat_bot_window, self._state])

重点的问题消息处理的函数在

then(process_input,[self._chat_query_input_textbox, self._chat_bot_window], [self._chat_query_input_textbox, self._chat_bot_window])

这里有个语法问题，我看了好久。第一个是处理函数，第二个是输入参数，第三个数输出参数。
第一个 [self._chat_query_input_textbox, self._chat_bot_window] 指定了处理函数的输入来源。这意味着：当用户触发某个事件（比如提交文本或点击按钮）时，process_input 函数将会接收来自 self._chat_query_input_textbox（用户输入的文本框）和 self._chat_bot_window（聊天窗口，可能用于显示之前的交互历史）的数据作为输入。
第二个 [self._chat_query_input_textbox, self._chat_bot_window] 指明了处理函数的输出目标。这表示：process_input 函数处理完输入数据后，将会更新 self._chat_query_input_textbox 和 self._chat_bot_window 的内容。具体来说，这可能涉及到清空用户的输入框以准备下一次输入，以及在聊天窗口中显示处理结果或者更新交互历史。

然后数据处理函数找到了，在这个函数处理_query_handler

最后发现是在初始化的时候外部传入进来的

def __init__(self, chatbot=None, streaming = False) -> None:self._interface = Noneself._query_handler = chatbot

最后在app.py 中找到了其实现的地方

def stream_chatbot(query, chat_history, session_id):

之后就是处理问答的逻辑了。这个下一次来研究。

总结

本文主要研究了chatwithrtx的用户接口，本来想有暴露合适的api，进行直接调用，但是通过研发发现，这个框架并不是一个前后端分离的框架，界面的渲染和函数调用混合在一起。所以接下来，需要研究如何在不使用这个界面的前提下，进行问答的调用。

我的知识星球

请关注公众号g0415shenw 加入知识星球。
星球地址 https://t.zsxq.com/15EvfoA7n
星球有本人经验心得全部总结涵盖音视频，gb28181、虚幻引擎、其他编程工具等等。另外还可以在星球提问，我会尽力答复，等于给您多了一个引路人。