WordPress更改网站地址,给公司在百度上做网站,it软件外包公司,排名做网站优化笔者最近在测试星火大模型的时候#xff0c;他们是使用websocket 来建立对话#xff0c;而且星火大模型开放的测试代码#xff0c;质量上不咋地#xff08;20231030记录#xff09;#xff0c;还需要对websocket有一定的了解#xff0c;才适合自己微调。
安装#xff…笔者最近在测试星火大模型的时候他们是使用websocket 来建立对话而且星火大模型开放的测试代码质量上不咋地20231030记录还需要对websocket有一定的了解才适合自己微调。
安装
pip install websocket
pip install websocket-client文章目录 1 常见的websocket获取数据的方法1.1 第一种使用create_connection链接1.2 第二种WebSocketApp run_forever的方式 2 针对run_forever内容保存2.1 通过定义global变量来保存内容2.2 通过CallbackToIterator()来返回 1 常见的websocket获取数据的方法
参考【python: websocket获取实时数据的几种常见链接方式】常见的两种。
1.1 第一种使用create_connection链接
需要pip install websocket-client (此方法不建议使用链接不稳定容易断并且连接很耗时)
import time
from websocket import create_connectionurl wss://i.cg.net/wi/ws
while True: # 一直链接直到连接上就退出循环time.sleep(2)try:ws create_connection(url)print(ws)breakexcept Exception as e:print(连接异常, e)continue
while True: # 连接上退出第一个循环之后此循环用于一直获取数据ws.send({event:subscribe, channel:btc_usdt.ticker})response ws.recv()print(response)
1.2 第二种WebSocketApp run_forever的方式
import websocketdef on_message(ws, message): # 服务器有数据更新时主动推送过来的数据print(message)def on_error(ws, error): # 程序报错时就会触发on_error事件print(error)def on_close(ws):print(Connection closed ……)def on_open(ws): # 连接到服务器之后就会触发on_open事件这里用于send数据req {event:subscribe, channel:btc_usdt.deep}print(req)ws.send(req)if __name__ __main__:websocket.enableTrace(True)ws websocket.WebSocketApp(wss://i.cg.net/wi/ws,on_messageon_message,on_erroron_error,on_closeon_close)ws.on_open on_openws.run_forever(ping_timeout30)
第二种方式里面run_forever其实是流式返回内容大概可以看流式输出的样例 {code:0,sid:5ebc0d6833b54909b4a51fbe75a5051a,status:0}
### error: content{code:0,fileRefer:{\43816997a7a44a299d0bfb7c360c5838\:[2,0,1]},sid:5ebc0d6833b54909b4a51fbe75a5051a,status:99}
### error: content{code:0,content:橘,sid:5ebc0d6833b54909b4a51fbe75a5051a,status:1}橘{code:0,content:子。,sid:5ebc0d6833b54909b4a51fbe75a5051a,status:1}子。{code:0,content:,sid:5ebc0d6833b54909b4a51fbe75a5051a,status:2}
### closed ###
那么run_forever流式输出正常的内容如何保存呢进入下一章 2 针对run_forever内容保存
2.1 通过定义global变量来保存内容
参考【将Websocket数据保存到Pandas】 来看一下文中的案例
import jsonimport pandas as pd
import websocketdf pd.DataFrame(columns[foreignNotional, grossValue, homeNotional, price, side,size, symbol, tickDirection, timestamp, trdMatchID])def on_message(ws, message):msg json.loads(message)print(msg)global df# ignore_indexTrue has to be provided, otherwise youll get# Can only append a Series if ignore_indexTrue or if the Series has a name errorsdf df.append(msg, ignore_indexTrue)def on_error(ws, error):print(error)def on_close(ws):print(### closed ###)def on_open(ws):returnif __name__ __main__:ws websocket.WebSocketApp(wss://www.bitmex.com/realtime?subscribetrade:XBTUSD,on_openon_open, on_messageon_message, on_erroron_error, on_closeon_close)ws.run_forever()其中global df是在定义全局变量df可以在函数中把流式数据拿出来还是很不错的
2.2 通过CallbackToIterator()来返回
在开源项目中ChuanhuChatGPT看到了使用的方式spark.py个人还没有尝试只是贴在这里。
贴一下这个函数
class CallbackToIterator:def __init__(self):self.queue deque()self.cond Condition()self.finished Falsedef callback(self, result):with self.cond:self.queue.append(result)self.cond.notify() # Wake up the generator.def __iter__(self):return selfdef __next__(self):with self.cond:# Wait for a value to be added to the queue.while not self.queue and not self.finished:self.cond.wait()if not self.queue:raise StopIteration()return self.queue.popleft()def finish(self):with self.cond:self.finished Trueself.cond.notify() # Wake up the generator if its waiting.# 主函数截取
def get_answer_stream_iter(self):wsParam Ws_Param(self.appid, self.api_key, self.api_secret, self.spark_url)websocket.enableTrace(False)wsUrl wsParam.create_url()ws websocket.WebSocketApp(wsUrl,on_messageself.on_message,on_errorself.on_error,on_closeself.on_close,on_openself.on_open,)ws.appid self.appidws.domain self.domain# Initialize the CallbackToIteratorws.iterator CallbackToIterator()# Start the WebSocket connection in a separate threadthread.start_new_thread(ws.run_forever, (), {sslopt: {cert_reqs: ssl.CERT_NONE}})# Iterate over the CallbackToIterator instanceanswer total_tokens 0for message in ws.iterator:data json.loads(message)code data[header][code]if code ! 0:ws.close()raise Exception(f请求错误: {code}, {data})else:choices data[payload][choices]status choices[status]content choices[text][0][content]if usage in data[payload]:total_tokens data[payload][usage][text][total_tokens]answer contentif status 2:ws.iterator.finish() # Finish the iterator when the status is 2ws.close()yield answer, total_tokens截取了部分代码这里先是定义ws.iterator CallbackToIterator()然后通过迭代从for message in ws.iterator:拿出数据看上去也是可行的
