dakaca/linux

1const express = require('express')2const fetch = require('cross-fetch')3const app = express()4var multer = require('multer');5var forms = multer({limits: { fieldSize: 10*1024*1024 }});6app.use(forms.array()); 7const cors = require('cors');8app.use(cors());9 10const bodyParser = require('body-parser')11app.use(bodyParser.json({limit : '50mb' }));  12app.use(bodyParser.urlencoded({ extended: true }));13 14const tencentcloud = require("tencentcloud-sdk-nodejs");15const TmsClient = tencentcloud.tms.v20201229.Client;16const clientConfig = {17  credential: {18    secretId: process.env.TENCENT_CLOUD_SID,19    secretKey: process.env.TENCENT_CLOUD_SKEY,20  },21  region: process.env.TENCENT_CLOUD_AP||"ap-singapore",22  profile: {23    httpProfile: {24      endpoint: "tms.tencentcloudapi.com",25    },26  },27};28const mdClient = process.env.TENCENT_CLOUD_SID && process.env.TENCENT_CLOUD_SKEY ? new TmsClient(clientConfig) : false;29 30const controller = new AbortController();31 32app.all(`*`, async (req, res) => {33  34  if(req.originalUrl) req.url = req.originalUrl;35  let url = `https://api.openai.com${req.url}`;36  // 从 header 中取得 Authorization': 'Bearer 后的 token37  const token = req.headers.authorization?.split(' ')[1];38  if( !token ) return res.status(403).send('Forbidden');39 40  const openai_key = process.env.OPENAI_KEY||token.split(':')[0];41  if( !openai_key ) return res.status(403).send('Forbidden');42  if( openai_key.startsWith("fk") ) url = url.replaceAll( "api.openai.com", "openai.api2d.net" );43 44  const proxy_key = token.split(':')[1]||"";  45  if( process.env.PROXY_KEY && proxy_key !== process.env.PROXY_KEY ) 46    return res.status(403).send('Forbidden');47 48  // console.log( req );49  const { moderation, moderation_level, ...restBody } = req.body;50  let sentence = "";51  // 建立一个句子缓冲区52  let sentence_buffer = [];53  let processing = false;54  let processing_stop = false;55 56  async function process_buffer(res)57  {58    if( processing_stop )59    {60      console.log("processing_stop",processing_stop);61      return false;62    }63 64    console.log("句子缓冲区" + new Date(), sentence_buffer);65    66    // 处理句子缓冲区67    if( processing )68    {69      // 有正在处理的，1秒钟后重试70      console.log("有正在处理的，1秒钟后重试");71      setTimeout( () => process_buffer(res), 1000 );72      return false;73    }74    75    processing = true;76    const sentence = sentence_buffer.shift();77    console.log("取出句子", sentence);78    if( sentence )79    {80      if( sentence === '[DONE]' )81      {82        console.log("[DONE]", "结束输出");83        res.write("data: "+sentence+"\n\n" );84        processing = false;85        res.end();86        return true;87      }else88      {89        // 开始对句子进行审核90        let data_array = JSON.parse(sentence);91        console.log("解析句子数据为array",data_array);92 93        const sentence_content = data_array.choices[0]?.delta?.content;94        console.log("sentence_content", sentence_content);95        if( sentence_content )96        {97          const params = {"Content": Buffer.from(sentence_content).toString('base64')};  98          const md_result = await mdClient.TextModeration(params);99          // console.log("审核结果", md_result);100          let md_check = moderation_level == 'high' ? md_result.Suggestion != 'Pass' : md_result.Suggestion == 'Block';101          if( md_check )102          {103            // 终止输出104            console.log("审核不通过", sentence_content, md_result);105            let forbidden_array = data_array;106            forbidden_array.choices[0].delta.content = "这个话题不适合讨论，换个话题吧。";107            res.write("data: "+JSON.stringify(forbidden_array)+"\n\n" );108            res.write("data: [DONE]\n\n" );109            res.end();110            controller.abort();111            processing = false;112            processing_stop = true;113            return false;114          }else115          {116            console.log("审核通过", sentence_content);117            res.write("data: "+sentence+"\n\n" );118            processing = false;119            console.log("processing",processing);120            return true;121          }122        }123 124      }125    }else126    {127      // console.log("句子缓冲区为空");128    }129 130    processing = false;131  }132 133  134  const options = {135      method: req.method,136      timeout: process.env.TIMEOUT||30000,137      signal: controller.signal,138      headers: {139        'Content-Type': 'application/json; charset=utf-8',140        'Authorization': 'Bearer '+ openai_key,141      },142      onMessage: async (data) => {143        // console.log(data);144        if( data === '[DONE]' )145        {146          sentence_buffer.push(data);147          await process_buffer(res);  148        }else149        {150          if( moderation && mdClient )151          {152            try {153              let data_array = JSON.parse(data);154              const char = data_array.choices[0]?.delta?.content;155              if( char ) sentence += char;156              // console.log("sentence",sentence );157              if( char == '。' || char == '？' || char == '！' || char == "\n" )158              {159                // 将 sentence 送审160                console.log("遇到句号，将句子放入缓冲区", sentence);161                data_array.choices[0].delta.content = sentence;162                sentence = "";163                sentence_buffer.push(JSON.stringify(data_array));164                await process_buffer(res);165              }166            } catch (error) {167              // 因为开头已经处理的了 [DONE] 的情况，这里应该不会出现无法解析json的情况 168              console.log( "error", error );169            }   170          }else171          {172            // 如果没有文本审核参数或者设置，直接输出173            res.write("data: "+data+"\n\n" );  174          }175        }176      }177  };178  179  if( req.method.toLocaleLowerCase() === 'post' && req.body ) options.body = JSON.stringify(restBody);180  // console.log({url, options});181 182  try {183    184    // 如果是 chat completion 和 text completion，使用 SSE185    if( (req.url.startsWith('/v1/completions') || req.url.startsWith('/v1/chat/completions')) && req.body.stream ) {186      console.log("使用 SSE");187      const response = await myFetch(url, options);188      if( response.ok )189      {190        // write header191        res.writeHead(200, {192          'Content-Type': 'text/event-stream',193          'Cache-Control': 'no-cache',194          'Connection': 'keep-alive',195        });196        const  { createParser } = await import("eventsource-parser");197        const parser = createParser((event) => {198          // console.log(event);    199          if (event.type === "event") {200            options.onMessage(event.data);201          }202        });203        if (!response.body.getReader) {204          const body = response.body;205          if (!body.on || !body.read) {206            throw new error('unsupported "fetch" implementation');207          }208          body.on("readable", () => {209            let chunk;210            while (null !== (chunk = body.read())) {211              // console.log(chunk.toString());212              parser.feed(chunk.toString());213            }214          });215        } else {216          for await (const chunk of streamAsyncIterable(response.body)) {217            const str = new TextDecoder().decode(chunk);218            parser.feed(str);219          }220        }221      }else222      {223        const body = await response.text();224        res.status(response.status).send(body);225      }226      227    }else228    {229      console.log("使用 fetch");230      const response = await myFetch(url, options);231      // console.log(response);232      const data = await response.json();233      // 审核结果234      if( moderation && mdClient )235      {236        const params = {"Content": Buffer.from(data.choices[0].message.content).toString('base64')};  237        const md_result = await mdClient.TextModeration(params);238        // console.log("审核结果", md_result);239        let md_check = moderation_level == 'high' ? md_result.Suggestion != 'Pass' : md_result.Suggestion == 'Block';240        if( md_check )241        {242          // 终止输出243          console.log("审核不通过", data.choices[0].message.content, md_result);244          data.choices[0].message.content = "这个话题不适合讨论，换个话题吧。";245        }else246        {247          console.log("审核通过", data.choices[0].message.content);248        }249      }250 251      res.json(data);252    }253    254    255  } catch (error) {256    console.error(error);257    res.status(500).json({"error":error.toString()});  258  }259})260 261async function* streamAsyncIterable(stream) {262  const reader = stream.getReader();263  try {264    while (true) {265      const { done, value } = await reader.read();266      if (done) {267        return;268      }269      yield value;270    }271  } finally {272    reader.releaseLock();273  }274}275 276async function myFetch(url, options) {277  const {timeout, ...fetchOptions} = options;278  const controller = new AbortController();279  const timeoutId = setTimeout(() => controller.abort(), timeout||30000)280  const res = await fetch(url, {...fetchOptions,signal:controller.signal});281  clearTimeout(timeoutId);282  return res;283}284 285// Error handler286app.use(function(err, req, res, next) {287  console.error(err)288  res.status(500).send('Internal Serverless Error')289})290 291const port = process.env.PORT||7860;292app.listen(port, () => {293  console.log(`Server start on http://localhost:${port}`);294})