問(wèn)題描述

我這邊是用thinkphp6的workerman請(qǐng)求openai接口，用微信小程序調(diào)用wss地址，經(jīng)常會(huì)出現(xiàn)*4556 recv() failed (104: Connection reset by peer) while proxying upgraded connection,報(bào)錯(cuò)

程序代碼

public function onMessage($connection, $data)
    {
        ob_start(null,4096);
        $open_ai_key = "xxxx";//你的key\
        $show = json_decode($data,true);
        if(!empty($show['type'])){
            if (ob_get_length() > 0) {
                ob_flush();
                flush();
            } else if (ob_get_level() > 0) {
                ob_end_clean();
            }
            return;
        }
        if(!$this->is_vip_expired($show['uid'])){
            $vipNum = Member::where('id',$show['uid'])->value('vipNum');
            if($vipNum==0){
                $vipMsg = '您好，您的可用次數(shù)為<span style="color:red">0</span>，需要繼續(xù)使用，請(qǐng)開(kāi)通會(huì)員。';
                //更新模型時(shí)間
                Dialog::where('id',$show['dialogId'])->update(['updateTime'=>date('Y-m-d H:i:s')]);
                $connection->send($vipMsg);
                if (ob_get_length() > 0) {
                    ob_flush();
                    flush();
                } else if (ob_get_level() > 0) {
                    ob_end_clean();
                }
                return;
            }else{
                Member::where('id',$show['uid'])->dec('vipNum')->update();;
            }
        }
        $messageData=[];
        if($show['message']==''){
            $classDataID=Dialog::where('id',$show['dialogId'])->value('classDataId');
            $classData =FzhelpClassPresets::where('classDataId',$classDataID)->select();
            $show['msg']=FzhelpClassData::where('classDataId',$classDataID)->value('desc');
            foreach($classData as $k=>$v){
                $messageData[$k]['role']=$v['type'];
                $messageData[$k]['content']=$v['content'];
            }
        }else{
            $messageDialog = DialogData::where('dialogId',$show['dialogId'])->limit(getFzhelpConfig('histNews'))->order('createTime desc')->select();
            $strlen=0;
            $count = count($messageDialog);
            foreach($messageDialog as $k=>$v){
                if($count-1!==$k){
                    $messageData[$k]['role']=$v['type']==2?'assistant':'user';
                    $messageData[$k]['content']=$v['content'];
                    $strlen+=strlen($v['content']);
                }
            }
            $messageNewData[]=[
                "role" => "user",
                "content" => $show['message']
            ];
            $messageData = array_merge($messageData, $messageNewData);
        }
        $opts = [
            "stream" => true,
            'model' => getFzhelpConfig('apiModel'),
            'messages' => $messageData,
            'temperature' => (float)getFzhelpConfig('temperature'),
            'max_tokens' => (int)getFzhelpConfig('maxTokens'),
            'frequency_penalty' => 0,
            'presence_penalty' => (float) getFzhelpConfig('presence'),
            'user'=>$show['openId']
        ];
        $headers  = [
            'Accept: application/json',
            'Content-Type: application/json',
            'Authorization: Bearer ' . $open_ai_key
        ];
        //返回?cái)?shù)據(jù)
        $response = '';
        $imperfect='';
        $callback = function ($ch, $data) use ($connection,$show,&$response,&$imperfect){
            if($show['message']==''){
//                添加機(jī)器人消息入庫(kù)
                DialogData::create([
                    'dialogId'=>$show['dialogId'],
                    'type'=>2,
                    'content'=>$show['msg'],
                    'zan'=>0,
                    'cai'=>0,
                ]);
                //更新模型時(shí)間
                Dialog::where('id',$show['dialogId'])->update(['updateTime'=>date('Y-m-d H:i:s')]);
                $connection->send($show['msg']);
                if (ob_get_length() > 0) {
                    ob_flush();
                    flush();
                } else if (ob_get_level() > 0) {
                    ob_end_clean();
                }
                return $show['msg'];
            }
            $dataLength = strlen($data);
//有可能會(huì)返回不完整的數(shù)據(jù)
            if($imperfect){
                $data = $imperfect . $data;
                $imperfect = '';
            }else {
                if (substr($data, -1) !== "\n") {
                    $imperfect = $data;
                    return $dataLength;
                }
            }
            $pword = $this->parseData($data);
            if(strpos($pword, 'data: [DONE]') !== false){
                $un_word = str_replace("data: [DONE]","",$pword);
                $response .= $un_word;
                if (!empty($response)) {
                    $addDialog=DialogData::create([
                        'dialogId'=>$show['dialogId'],
                        'type'=>2,
                        'content'=>$response,
                        'zan'=>0,
                        'cai'=>0,
                    ]);
                    //更新模型時(shí)間
                    Dialog::where('id',$show['dialogId'])->update(['updateTime'=>date('Y-m-d H:i:s')]);
                    $connection->send('endId_miaowenzhen:['.$addDialog->id.']');
                    ob_flush();flush();
                    $response = '';
                }
                $connection->send($pword);
                ob_flush();flush();
            }else{
                $response .= $pword;
                $connection->send($pword);
                ob_flush();flush();
            }
            return strlen($data);
        };
//        header('Content-Type: text/event-stream');
//        header('Cache-Control: no-cache');
//        header('Connection: keep-alive');
//        header('X-Accel-Buffering: no');
        $ch = curl_init();
        curl_setopt($ch, CURLOPT_TIMEOUT, 60);
        curl_setopt($ch, CURLOPT_SSL_VERIFYPEER, FALSE);
        curl_setopt($ch, CURLOPT_SSL_VERIFYHOST, FALSE);
        curl_setopt($ch, CURLOPT_URL, getFzhelpConfig('apiUrl'));
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
        curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);
        curl_setopt($ch, CURLOPT_BUFFERSIZE,8192);
        curl_setopt($ch, CURLOPT_POST, 1);
        curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($opts));
        curl_setopt($ch, CURLOPT_WRITEFUNCTION, $callback);
        curl_exec($ch);
    }

報(bào)錯(cuò)信息

*4556 recv() failed (104: Connection reset by peer) while proxying upgraded connection, client: *.*.*.*, server: xxx.xxx.com, request: ＂GET /wss HTTP/1.1＂, upstream: ＂http://127.0.0.1:2345/wss＂, host: ＂xxx.xxx.com＂, referrer: ＂https://servicewechat.com/xxxxxxx/devtools/page-frame.html＂

截圖報(bào)錯(cuò)信息里報(bào)錯(cuò)文件相關(guān)代碼

操作系統(tǒng)及workerman/webman等框架組件具體版本

Debian GNU/Linux 11 x86_64
nginx:1.22.1
php:8.1.13
thinkphp:6.1
workerman:3.5.34