()
| 189 | @app.route('/completions', methods=['POST']) |
| 190 | @app.route('/v1/completions', methods=['POST']) |
| 191 | def completion(): |
| 192 | if (args.api_key != "" and request.headers["Authorization"].split()[1] != args.api_key): |
| 193 | return Response(status=403) |
| 194 | body = request.get_json() |
| 195 | stream = False |
| 196 | tokenize = False |
| 197 | if(is_present(body, "stream")): stream = body["stream"] |
| 198 | if(is_present(body, "tokenize")): tokenize = body["tokenize"] |
| 199 | postData = make_postData(body, chat=False, stream=stream) |
| 200 | |
| 201 | promptToken = [] |
| 202 | if (tokenize): |
| 203 | tokenData = requests.request("POST", urllib.parse.urljoin(args.llama_api, "/tokenize"), data=json.dumps({"content": postData["prompt"]})).json() |
| 204 | promptToken = tokenData["tokens"] |
| 205 | |
| 206 | if (not stream): |
| 207 | data = requests.request("POST", urllib.parse.urljoin(args.llama_api, "/completion"), data=json.dumps(postData)) |
| 208 | print(data.json()) |
| 209 | resData = make_resData(data.json(), chat=False, promptToken=promptToken) |
| 210 | return jsonify(resData) |
| 211 | else: |
| 212 | def generate(): |
| 213 | data = requests.request("POST", urllib.parse.urljoin(args.llama_api, "/completion"), data=json.dumps(postData), stream=True) |
| 214 | time_now = int(time.time()) |
| 215 | for line in data.iter_lines(): |
| 216 | if line: |
| 217 | decoded_line = line.decode('utf-8') |
| 218 | resData = make_resData_stream(json.loads(decoded_line[6:]), chat=False, time_now=time_now) |
| 219 | yield 'data: {}\n'.format(json.dumps(resData)) |
| 220 | return Response(generate(), mimetype='text/event-stream') |
| 221 | |
| 222 | if __name__ == '__main__': |
| 223 | app.run(args.host, port=args.port) |
nothing calls this directly
no test coverage detected