We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
bug-fix: when using stream is False, continuous batching doesn't work
## Motivation

bug fix : when using stream=False, the continuous batching is not working.
See details in this issue: https://github.com/InternLM/lmdeploy/issues/308

## Modification

In non-streaming mode, the main thread which is responsible for receiving request is stuck in the following code since self.que.get() will wait until the queue is not empty.

But when using non-stream mode, the queue is always empty unless a request is finished.

So the newly coming request won't be processed before the last request is done. So the batching doesn't work.

```python
 while True:
            while self.que.qsize() > 1:
                self.que.get()
            finish, tm_outputs = self.que.get()
            outputs = _tm_dict_to_torch_dict(tm_outputs)
```

I modify here to avoid the long wait. just see the commits.

I'm not sure if there is a better way to fix this. If any, please comment.

I have a question, what is this line for?  Won't it cause output tokens being lost???
```python
while self.que.qsize() > 1: self.que.get()
```

## BC-breaking (Optional)

Does the modification introduce changes that break the backward-compatibility of the downstream repositories?
If so, please describe how it breaks the compatibility and how the downstream projects should modify their code to keep compatibility with this PR.

## Use cases (Optional)

If this PR introduces a new feature, it is better to list some use cases here, and update the documentation.

## Checklist

1. Pre-commit or other linting tools are used to fix the potential lint issues.
2. The modification is covered by complete unit tests. If not, please add more unit tests to ensure the correctness.
3. If the modification has a dependency on downstream projects of a newer version, this PR should be tested with all supported versions of downstream projects.
4. The documentation has been modified accordingly, like docstring or example tutorials.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `lmdeploy/serve/openai/api_server.py`
Content:
```
1 # Copyright (c) OpenMMLab. All rights reserved.
2 import json
3 import os
4 import time
5 from http import HTTPStatus
6 from typing import AsyncGenerator, Optional
7 
8 import fire
9 import uvicorn
10 from fastapi import BackgroundTasks, FastAPI, Request
11 from fastapi.responses import JSONResponse, StreamingResponse
12 
13 from lmdeploy.serve.async_engine import AsyncEngine
14 from lmdeploy.serve.openai.protocol import (  # noqa: E501
15     ChatCompletionRequest, ChatCompletionResponse,
16     ChatCompletionResponseChoice, ChatCompletionResponseStreamChoice,
17     ChatCompletionStreamResponse, ChatMessage, DeltaMessage, EmbeddingsRequest,
18     EmbeddingsResponse, ErrorResponse, GenerateRequest, ModelCard, ModelList,
19     ModelPermission, UsageInfo)
20 
21 os.environ['TM_LOG_LEVEL'] = 'ERROR'
22 
23 
24 class VariableInterface:
25     """A IO interface maintaining variables."""
26     async_engine: AsyncEngine = None
27     request_hosts = []
28 
29 
30 app = FastAPI(docs_url='/')
31 
32 
33 def get_model_list():
34     """Available models.
35 
36     Only provided one now.
37     """
38     return [VariableInterface.async_engine.tm_model.model_name]
39 
40 
41 @app.get('/v1/models')
42 def available_models():
43     """Show available models."""
44     model_cards = []
45     for model_name in get_model_list():
46         model_cards.append(
47             ModelCard(id=model_name,
48                       root=model_name,
49                       permission=[ModelPermission()]))
50     return ModelList(data=model_cards)
51 
52 
53 def create_error_response(status: HTTPStatus, message: str):
54     """Create error response according to http status and message.
55 
56     Args:
57         status (HTTPStatus): HTTP status codes and reason phrases
58         message (str): error message
59     """
60     return JSONResponse(ErrorResponse(message=message,
61                                       type='invalid_request_error').dict(),
62                         status_code=status.value)
63 
64 
65 async def check_request(request) -> Optional[JSONResponse]:
66     """Check if a request is valid."""
67     if request.model in get_model_list():
68         return
69     ret = create_error_response(
70         HTTPStatus.NOT_FOUND, f'The model `{request.model}` does not exist.')
71     return ret
72 
73 
74 @app.post('/v1/chat/completions')
75 async def chat_completions_v1(request: ChatCompletionRequest,
76                               raw_request: Request = None):
77     """Completion API similar to OpenAI's API.
78 
79     Refer to  `https://platform.openai.com/docs/api-reference/chat/create`
80     for the API specification.
81 
82     The request should be a JSON object with the following fields:
83     - model: model name. Available from /v1/models.
84     - messages: string prompt or chat history in OpenAI format.
85     - temperature (float): to modulate the next token probability
86     - top_p (float): If set to float < 1, only the smallest set of most
87         probable tokens with probabilities that add up to top_p or higher
88         are kept for generation.
89     - n (int): How many chat completion choices to generate for each input
90         message. Only support one here.
91     - stream: whether to stream the results or not. Default to false.
92     - max_tokens (int): output token nums
93     - repetition_penalty (float): The parameter for repetition penalty.
94         1.0 means no penalty
95 
96     Additional arguments supported by LMDeploy:
97     - renew_session (bool): Whether renew the session. Can be used when the
98         session length is exceeded.
99     - ignore_eos (bool): indicator for ignoring eos
100 
101     Currently we do not support the following features:
102     - function_call (Users should implement this by themselves)
103     - logit_bias (not supported yet)
104     - presence_penalty (replaced with repetition_penalty)
105     - frequency_penalty (replaced with repetition_penalty)
106     """
107     instance_id = int(raw_request.client.host.replace('.', ''))
108 
109     error_check_ret = await check_request(request)
110     if error_check_ret is not None:
111         return error_check_ret
112 
113     model_name = request.model
114     request_id = str(instance_id)
115     created_time = int(time.time())
116 
117     result_generator = VariableInterface.async_engine.generate_openai(
118         request.messages,
119         instance_id,
120         request.stream,
121         request.renew_session,
122         request_output_len=request.max_tokens if request.max_tokens else 512,
123         stop=request.stop,
124         top_p=request.top_p,
125         temperature=request.temperature,
126         repetition_penalty=request.repetition_penalty,
127         ignore_eos=request.ignore_eos)
128 
129     async def abort_request() -> None:
130         async for _ in VariableInterface.async_engine.generate_openai(
131                 request.messages,
132                 instance_id,
133                 request.stream,
134                 request.renew_session,
135                 stop=True):
136             pass
137 
138     def create_stream_response_json(
139         index: int,
140         text: str,
141         finish_reason: Optional[str] = None,
142     ) -> str:
143         choice_data = ChatCompletionResponseStreamChoice(
144             index=index,
145             delta=DeltaMessage(role='assistant', content=text),
146             finish_reason=finish_reason,
147         )
148         response = ChatCompletionStreamResponse(
149             id=request_id,
150             created=created_time,
151             model=model_name,
152             choices=[choice_data],
153         )
154         response_json = response.json(ensure_ascii=False)
155 
156         return response_json
157 
158     async def completion_stream_generator() -> AsyncGenerator[str, None]:
159         # First chunk with role
160         for i in range(request.n):
161             choice_data = ChatCompletionResponseStreamChoice(
162                 index=i,
163                 delta=DeltaMessage(role='assistant'),
164                 finish_reason=None,
165             )
166             chunk = ChatCompletionStreamResponse(id=request_id,
167                                                  choices=[choice_data],
168                                                  model=model_name)
169             data = chunk.json(exclude_unset=True, ensure_ascii=False)
170             yield f'data: {data}\n\n'
171 
172         async for res in result_generator:
173             response_json = create_stream_response_json(
174                 index=0,
175                 text=res.response,
176             )
177             yield f'data: {response_json}\n\n'
178         yield 'data: [DONE]\n\n'
179 
180     # Streaming response
181     if request.stream:
182         background_tasks = BackgroundTasks()
183         # Abort the request if the client disconnects.
184         background_tasks.add_task(abort_request)
185         return StreamingResponse(completion_stream_generator(),
186                                  media_type='text/event-stream',
187                                  background=background_tasks)
188 
189     # Non-streaming response
190     final_res = None
191     async for res in result_generator:
192         if await raw_request.is_disconnected():
193             # Abort the request if the client disconnects.
194             await abort_request()
195             return create_error_response(HTTPStatus.BAD_REQUEST,
196                                          'Client disconnected')
197         final_res = res
198     assert final_res is not None
199     choices = []
200     choice_data = ChatCompletionResponseChoice(
201         index=0,
202         message=ChatMessage(role='assistant', content=final_res.response),
203         finish_reason=final_res.finish_reason,
204     )
205     choices.append(choice_data)
206 
207     total_tokens = sum([
208         final_res.history_token_len, final_res.input_token_len,
209         final_res.generate_token_len
210     ])
211     usage = UsageInfo(
212         prompt_tokens=final_res.input_token_len,
213         completion_tokens=final_res.generate_token_len,
214         total_tokens=total_tokens,
215     )
216     response = ChatCompletionResponse(
217         id=request_id,
218         created=created_time,
219         model=model_name,
220         choices=choices,
221         usage=usage,
222     )
223 
224     return response
225 
226 
227 @app.post('/v1/embeddings')
228 async def create_embeddings(request: EmbeddingsRequest,
229                             raw_request: Request = None):
230     """Creates embeddings for the text."""
231     error_check_ret = await check_request(request)
232     if error_check_ret is not None:
233         return error_check_ret
234 
235     embedding = await VariableInterface.async_engine.get_embeddings(
236         request.input)
237     data = [{'object': 'embedding', 'embedding': embedding, 'index': 0}]
238     token_num = len(embedding)
239     return EmbeddingsResponse(
240         data=data,
241         model=request.model,
242         usage=UsageInfo(
243             prompt_tokens=token_num,
244             total_tokens=token_num,
245             completion_tokens=None,
246         ),
247     ).dict(exclude_none=True)
248 
249 
250 @app.post('/generate')
251 async def generate(request: GenerateRequest, raw_request: Request = None):
252     """Generate completion for the request.
253 
254     The request should be a JSON object with the following fields:
255     - prompt: the prompt to use for the generation.
256     - instance_id: determine which instance will be called. If not specified
257         with a value other than -1, using host ip directly.
258     - sequence_start (bool): indicator for starting a sequence.
259     - sequence_end (bool): indicator for ending a sequence
260     - stream: whether to stream the results or not.
261     - stop: whether to stop the session response or not.
262     - request_output_len (int): output token nums
263     - step (int): the offset of the k/v cache
264     - top_p (float): If set to float < 1, only the smallest set of most
265         probable tokens with probabilities that add up to top_p or higher
266         are kept for generation.
267     - top_k (int): The number of the highest probability vocabulary
268         tokens to keep for top-k-filtering
269     - temperature (float): to modulate the next token probability
270     - repetition_penalty (float): The parameter for repetition penalty.
271         1.0 means no penalty
272     - ignore_eos (bool): indicator for ignoring eos
273     """
274     if request.instance_id == -1:
275         instance_id = int(raw_request.client.host.replace('.', ''))
276         request.instance_id = instance_id
277 
278     generation = VariableInterface.async_engine.generate(
279         request.prompt,
280         request.instance_id,
281         stream_response=True,  # always use stream to enable batching
282         sequence_start=request.sequence_start,
283         sequence_end=request.sequence_end,
284         request_output_len=request.request_output_len,
285         top_p=request.top_p,
286         top_k=request.top_k,
287         stop=request.stop,
288         temperature=request.temperature,
289         repetition_penalty=request.repetition_penalty,
290         ignore_eos=request.ignore_eos)
291 
292     # Streaming case
293     async def stream_results() -> AsyncGenerator[bytes, None]:
294         async for out in generation:
295             ret = {
296                 'text': out.response,
297                 'tokens': out.generate_token_len,
298                 'finish_reason': out.finish_reason
299             }
300             yield (json.dumps(ret) + '\0').encode('utf-8')
301 
302     if request.stream:
303         return StreamingResponse(stream_results())
304     else:
305         ret = {}
306         text = ''
307         tokens = 0
308         finish_reason = None
309         async for out in generation:
310             text += out.response
311             tokens += out.generate_token_len
312             finish_reason = out.finish_reason
313         ret = {'text': text, 'tokens': tokens, 'finish_reason': finish_reason}
314         return JSONResponse(ret)
315 
316 
317 def main(model_path: str,
318          server_name: str = 'localhost',
319          server_port: int = 23333,
320          instance_num: int = 32,
321          tp: int = 1):
322     """An example to perform model inference through the command line
323     interface.
324 
325     Args:
326         model_path (str): the path of the deployed model
327         server_name (str): host ip for serving
328         server_port (int): server port
329         instance_num (int): number of instances of turbomind model
330         tp (int): tensor parallel
331     """
332     VariableInterface.async_engine = AsyncEngine(model_path=model_path,
333                                                  instance_num=instance_num,
334                                                  tp=tp)
335     uvicorn.run(app=app, host=server_name, port=server_port, log_level='info')
336 
337 
338 if __name__ == '__main__':
339     fire.Fire(main)
340
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/lmdeploy/serve/openai/api_server.py b/lmdeploy/serve/openai/api_server.py
--- a/lmdeploy/serve/openai/api_server.py
+++ b/lmdeploy/serve/openai/api_server.py
@@ -57,9 +57,10 @@
         status (HTTPStatus): HTTP status codes and reason phrases
         message (str): error message
     """
-    return JSONResponse(ErrorResponse(message=message,
-                                      type='invalid_request_error').dict(),
-                        status_code=status.value)
+    return JSONResponse(
+        ErrorResponse(message=message,
+                      type='invalid_request_error',
+                      code=status.value).dict())
 
 
 async def check_request(request) -> Optional[JSONResponse]:
@@ -117,7 +118,7 @@
     result_generator = VariableInterface.async_engine.generate_openai(
         request.messages,
         instance_id,
-        request.stream,
+        True,  # always use stream to enable batching
         request.renew_session,
         request_output_len=request.max_tokens if request.max_tokens else 512,
         stop=request.stop,
@@ -130,7 +131,7 @@
         async for _ in VariableInterface.async_engine.generate_openai(
                 request.messages,
                 instance_id,
-                request.stream,
+                True,
                 request.renew_session,
                 stop=True):
             pass
@@ -188,6 +189,7 @@
 
     # Non-streaming response
     final_res = None
+    text = ''
     async for res in result_generator:
         if await raw_request.is_disconnected():
             # Abort the request if the client disconnects.
@@ -195,11 +197,12 @@
             return create_error_response(HTTPStatus.BAD_REQUEST,
                                          'Client disconnected')
         final_res = res
+        text += res.response
     assert final_res is not None
     choices = []
     choice_data = ChatCompletionResponseChoice(
         index=0,
-        message=ChatMessage(role='assistant', content=final_res.response),
+        message=ChatMessage(role='assistant', content=text),
         finish_reason=final_res.finish_reason,
     )
     choices.append(choice_data)
@@ -308,7 +311,7 @@
         finish_reason = None
         async for out in generation:
             text += out.response
-            tokens += out.generate_token_len
+            tokens = out.generate_token_len
             finish_reason = out.finish_reason
         ret = {'text': text, 'tokens': tokens, 'finish_reason': finish_reason}
         return JSONResponse(ret)

{"golden_diff": "diff --git a/lmdeploy/serve/openai/api_server.py b/lmdeploy/serve/openai/api_server.py\n--- a/lmdeploy/serve/openai/api_server.py\n+++ b/lmdeploy/serve/openai/api_server.py\n@@ -57,9 +57,10 @@\n         status (HTTPStatus): HTTP status codes and reason phrases\n         message (str): error message\n     \"\"\"\n-    return JSONResponse(ErrorResponse(message=message,\n-                                      type='invalid_request_error').dict(),\n-                        status_code=status.value)\n+    return JSONResponse(\n+        ErrorResponse(message=message,\n+                      type='invalid_request_error',\n+                      code=status.value).dict())\n \n \n async def check_request(request) -> Optional[JSONResponse]:\n@@ -117,7 +118,7 @@\n     result_generator = VariableInterface.async_engine.generate_openai(\n         request.messages,\n         instance_id,\n-        request.stream,\n+        True,  # always use stream to enable batching\n         request.renew_session,\n         request_output_len=request.max_tokens if request.max_tokens else 512,\n         stop=request.stop,\n@@ -130,7 +131,7 @@\n         async for _ in VariableInterface.async_engine.generate_openai(\n                 request.messages,\n                 instance_id,\n-                request.stream,\n+                True,\n                 request.renew_session,\n                 stop=True):\n             pass\n@@ -188,6 +189,7 @@\n \n     # Non-streaming response\n     final_res = None\n+    text = ''\n     async for res in result_generator:\n         if await raw_request.is_disconnected():\n             # Abort the request if the client disconnects.\n@@ -195,11 +197,12 @@\n             return create_error_response(HTTPStatus.BAD_REQUEST,\n                                          'Client disconnected')\n         final_res = res\n+        text += res.response\n     assert final_res is not None\n     choices = []\n     choice_data = ChatCompletionResponseChoice(\n         index=0,\n-        message=ChatMessage(role='assistant', content=final_res.response),\n+        message=ChatMessage(role='assistant', content=text),\n         finish_reason=final_res.finish_reason,\n     )\n     choices.append(choice_data)\n@@ -308,7 +311,7 @@\n         finish_reason = None\n         async for out in generation:\n             text += out.response\n-            tokens += out.generate_token_len\n+            tokens = out.generate_token_len\n             finish_reason = out.finish_reason\n         ret = {'text': text, 'tokens': tokens, 'finish_reason': finish_reason}\n         return JSONResponse(ret)\n", "issue": "bug-fix: when using stream is False, continuous batching doesn't work\n## Motivation\r\n\r\nbug fix : when using stream=False, the continuous batching is not working.\r\nSee details in this issue: https://github.com/InternLM/lmdeploy/issues/308\r\n\r\n## Modification\r\n\r\nIn non-streaming mode, the main thread which is responsible for receiving request is stuck in the following code since self.que.get() will wait until the queue is not empty.\r\n\r\nBut when using non-stream mode, the queue is always empty unless a request is finished.\r\n\r\nSo the newly coming request won't be processed before the last request is done. So the batching doesn't work.\r\n\r\n```python\r\n while True:\r\n            while self.que.qsize() > 1:\r\n                self.que.get()\r\n            finish, tm_outputs = self.que.get()\r\n            outputs = _tm_dict_to_torch_dict(tm_outputs)\r\n```\r\n\r\nI modify here to avoid the long wait. just see the commits.\r\n\r\nI'm not sure if there is a better way to fix this. If any, please comment.\r\n\r\nI have a question, what is this line for?  Won't it cause output tokens being lost???\r\n```python\r\nwhile self.que.qsize() > 1: self.que.get()\r\n```\r\n\r\n## BC-breaking (Optional)\r\n\r\nDoes the modification introduce changes that break the backward-compatibility of the downstream repositories?\r\nIf so, please describe how it breaks the compatibility and how the downstream projects should modify their code to keep compatibility with this PR.\r\n\r\n## Use cases (Optional)\r\n\r\nIf this PR introduces a new feature, it is better to list some use cases here, and update the documentation.\r\n\r\n## Checklist\r\n\r\n1. Pre-commit or other linting tools are used to fix the potential lint issues.\r\n2. The modification is covered by complete unit tests. If not, please add more unit tests to ensure the correctness.\r\n3. If the modification has a dependency on downstream projects of a newer version, this PR should be tested with all supported versions of downstream projects.\r\n4. The documentation has been modified accordingly, like docstring or example tutorials.\r\n\n", "before_files": [{"content": "# Copyright (c) OpenMMLab. All rights reserved.\nimport json\nimport os\nimport time\nfrom http import HTTPStatus\nfrom typing import AsyncGenerator, Optional\n\nimport fire\nimport uvicorn\nfrom fastapi import BackgroundTasks, FastAPI, Request\nfrom fastapi.responses import JSONResponse, StreamingResponse\n\nfrom lmdeploy.serve.async_engine import AsyncEngine\nfrom lmdeploy.serve.openai.protocol import (  # noqa: E501\n    ChatCompletionRequest, ChatCompletionResponse,\n    ChatCompletionResponseChoice, ChatCompletionResponseStreamChoice,\n    ChatCompletionStreamResponse, ChatMessage, DeltaMessage, EmbeddingsRequest,\n    EmbeddingsResponse, ErrorResponse, GenerateRequest, ModelCard, ModelList,\n    ModelPermission, UsageInfo)\n\nos.environ['TM_LOG_LEVEL'] = 'ERROR'\n\n\nclass VariableInterface:\n    \"\"\"A IO interface maintaining variables.\"\"\"\n    async_engine: AsyncEngine = None\n    request_hosts = []\n\n\napp = FastAPI(docs_url='/')\n\n\ndef get_model_list():\n    \"\"\"Available models.\n\n    Only provided one now.\n    \"\"\"\n    return [VariableInterface.async_engine.tm_model.model_name]\n\n\n@app.get('/v1/models')\ndef available_models():\n    \"\"\"Show available models.\"\"\"\n    model_cards = []\n    for model_name in get_model_list():\n        model_cards.append(\n            ModelCard(id=model_name,\n                      root=model_name,\n                      permission=[ModelPermission()]))\n    return ModelList(data=model_cards)\n\n\ndef create_error_response(status: HTTPStatus, message: str):\n    \"\"\"Create error response according to http status and message.\n\n    Args:\n        status (HTTPStatus): HTTP status codes and reason phrases\n        message (str): error message\n    \"\"\"\n    return JSONResponse(ErrorResponse(message=message,\n                                      type='invalid_request_error').dict(),\n                        status_code=status.value)\n\n\nasync def check_request(request) -> Optional[JSONResponse]:\n    \"\"\"Check if a request is valid.\"\"\"\n    if request.model in get_model_list():\n        return\n    ret = create_error_response(\n        HTTPStatus.NOT_FOUND, f'The model `{request.model}` does not exist.')\n    return ret\n\n\n@app.post('/v1/chat/completions')\nasync def chat_completions_v1(request: ChatCompletionRequest,\n                              raw_request: Request = None):\n    \"\"\"Completion API similar to OpenAI's API.\n\n    Refer to  `https://platform.openai.com/docs/api-reference/chat/create`\n    for the API specification.\n\n    The request should be a JSON object with the following fields:\n    - model: model name. Available from /v1/models.\n    - messages: string prompt or chat history in OpenAI format.\n    - temperature (float): to modulate the next token probability\n    - top_p (float): If set to float < 1, only the smallest set of most\n        probable tokens with probabilities that add up to top_p or higher\n        are kept for generation.\n    - n (int): How many chat completion choices to generate for each input\n        message. Only support one here.\n    - stream: whether to stream the results or not. Default to false.\n    - max_tokens (int): output token nums\n    - repetition_penalty (float): The parameter for repetition penalty.\n        1.0 means no penalty\n\n    Additional arguments supported by LMDeploy:\n    - renew_session (bool): Whether renew the session. Can be used when the\n        session length is exceeded.\n    - ignore_eos (bool): indicator for ignoring eos\n\n    Currently we do not support the following features:\n    - function_call (Users should implement this by themselves)\n    - logit_bias (not supported yet)\n    - presence_penalty (replaced with repetition_penalty)\n    - frequency_penalty (replaced with repetition_penalty)\n    \"\"\"\n    instance_id = int(raw_request.client.host.replace('.', ''))\n\n    error_check_ret = await check_request(request)\n    if error_check_ret is not None:\n        return error_check_ret\n\n    model_name = request.model\n    request_id = str(instance_id)\n    created_time = int(time.time())\n\n    result_generator = VariableInterface.async_engine.generate_openai(\n        request.messages,\n        instance_id,\n        request.stream,\n        request.renew_session,\n        request_output_len=request.max_tokens if request.max_tokens else 512,\n        stop=request.stop,\n        top_p=request.top_p,\n        temperature=request.temperature,\n        repetition_penalty=request.repetition_penalty,\n        ignore_eos=request.ignore_eos)\n\n    async def abort_request() -> None:\n        async for _ in VariableInterface.async_engine.generate_openai(\n                request.messages,\n                instance_id,\n                request.stream,\n                request.renew_session,\n                stop=True):\n            pass\n\n    def create_stream_response_json(\n        index: int,\n        text: str,\n        finish_reason: Optional[str] = None,\n    ) -> str:\n        choice_data = ChatCompletionResponseStreamChoice(\n            index=index,\n            delta=DeltaMessage(role='assistant', content=text),\n            finish_reason=finish_reason,\n        )\n        response = ChatCompletionStreamResponse(\n            id=request_id,\n            created=created_time,\n            model=model_name,\n            choices=[choice_data],\n        )\n        response_json = response.json(ensure_ascii=False)\n\n        return response_json\n\n    async def completion_stream_generator() -> AsyncGenerator[str, None]:\n        # First chunk with role\n        for i in range(request.n):\n            choice_data = ChatCompletionResponseStreamChoice(\n                index=i,\n                delta=DeltaMessage(role='assistant'),\n                finish_reason=None,\n            )\n            chunk = ChatCompletionStreamResponse(id=request_id,\n                                                 choices=[choice_data],\n                                                 model=model_name)\n            data = chunk.json(exclude_unset=True, ensure_ascii=False)\n            yield f'data: {data}\\n\\n'\n\n        async for res in result_generator:\n            response_json = create_stream_response_json(\n                index=0,\n                text=res.response,\n            )\n            yield f'data: {response_json}\\n\\n'\n        yield 'data: [DONE]\\n\\n'\n\n    # Streaming response\n    if request.stream:\n        background_tasks = BackgroundTasks()\n        # Abort the request if the client disconnects.\n        background_tasks.add_task(abort_request)\n        return StreamingResponse(completion_stream_generator(),\n                                 media_type='text/event-stream',\n                                 background=background_tasks)\n\n    # Non-streaming response\n    final_res = None\n    async for res in result_generator:\n        if await raw_request.is_disconnected():\n            # Abort the request if the client disconnects.\n            await abort_request()\n            return create_error_response(HTTPStatus.BAD_REQUEST,\n                                         'Client disconnected')\n        final_res = res\n    assert final_res is not None\n    choices = []\n    choice_data = ChatCompletionResponseChoice(\n        index=0,\n        message=ChatMessage(role='assistant', content=final_res.response),\n        finish_reason=final_res.finish_reason,\n    )\n    choices.append(choice_data)\n\n    total_tokens = sum([\n        final_res.history_token_len, final_res.input_token_len,\n        final_res.generate_token_len\n    ])\n    usage = UsageInfo(\n        prompt_tokens=final_res.input_token_len,\n        completion_tokens=final_res.generate_token_len,\n        total_tokens=total_tokens,\n    )\n    response = ChatCompletionResponse(\n        id=request_id,\n        created=created_time,\n        model=model_name,\n        choices=choices,\n        usage=usage,\n    )\n\n    return response\n\n\n@app.post('/v1/embeddings')\nasync def create_embeddings(request: EmbeddingsRequest,\n                            raw_request: Request = None):\n    \"\"\"Creates embeddings for the text.\"\"\"\n    error_check_ret = await check_request(request)\n    if error_check_ret is not None:\n        return error_check_ret\n\n    embedding = await VariableInterface.async_engine.get_embeddings(\n        request.input)\n    data = [{'object': 'embedding', 'embedding': embedding, 'index': 0}]\n    token_num = len(embedding)\n    return EmbeddingsResponse(\n        data=data,\n        model=request.model,\n        usage=UsageInfo(\n            prompt_tokens=token_num,\n            total_tokens=token_num,\n            completion_tokens=None,\n        ),\n    ).dict(exclude_none=True)\n\n\n@app.post('/generate')\nasync def generate(request: GenerateRequest, raw_request: Request = None):\n    \"\"\"Generate completion for the request.\n\n    The request should be a JSON object with the following fields:\n    - prompt: the prompt to use for the generation.\n    - instance_id: determine which instance will be called. If not specified\n        with a value other than -1, using host ip directly.\n    - sequence_start (bool): indicator for starting a sequence.\n    - sequence_end (bool): indicator for ending a sequence\n    - stream: whether to stream the results or not.\n    - stop: whether to stop the session response or not.\n    - request_output_len (int): output token nums\n    - step (int): the offset of the k/v cache\n    - top_p (float): If set to float < 1, only the smallest set of most\n        probable tokens with probabilities that add up to top_p or higher\n        are kept for generation.\n    - top_k (int): The number of the highest probability vocabulary\n        tokens to keep for top-k-filtering\n    - temperature (float): to modulate the next token probability\n    - repetition_penalty (float): The parameter for repetition penalty.\n        1.0 means no penalty\n    - ignore_eos (bool): indicator for ignoring eos\n    \"\"\"\n    if request.instance_id == -1:\n        instance_id = int(raw_request.client.host.replace('.', ''))\n        request.instance_id = instance_id\n\n    generation = VariableInterface.async_engine.generate(\n        request.prompt,\n        request.instance_id,\n        stream_response=True,  # always use stream to enable batching\n        sequence_start=request.sequence_start,\n        sequence_end=request.sequence_end,\n        request_output_len=request.request_output_len,\n        top_p=request.top_p,\n        top_k=request.top_k,\n        stop=request.stop,\n        temperature=request.temperature,\n        repetition_penalty=request.repetition_penalty,\n        ignore_eos=request.ignore_eos)\n\n    # Streaming case\n    async def stream_results() -> AsyncGenerator[bytes, None]:\n        async for out in generation:\n            ret = {\n                'text': out.response,\n                'tokens': out.generate_token_len,\n                'finish_reason': out.finish_reason\n            }\n            yield (json.dumps(ret) + '\\0').encode('utf-8')\n\n    if request.stream:\n        return StreamingResponse(stream_results())\n    else:\n        ret = {}\n        text = ''\n        tokens = 0\n        finish_reason = None\n        async for out in generation:\n            text += out.response\n            tokens += out.generate_token_len\n            finish_reason = out.finish_reason\n        ret = {'text': text, 'tokens': tokens, 'finish_reason': finish_reason}\n        return JSONResponse(ret)\n\n\ndef main(model_path: str,\n         server_name: str = 'localhost',\n         server_port: int = 23333,\n         instance_num: int = 32,\n         tp: int = 1):\n    \"\"\"An example to perform model inference through the command line\n    interface.\n\n    Args:\n        model_path (str): the path of the deployed model\n        server_name (str): host ip for serving\n        server_port (int): server port\n        instance_num (int): number of instances of turbomind model\n        tp (int): tensor parallel\n    \"\"\"\n    VariableInterface.async_engine = AsyncEngine(model_path=model_path,\n                                                 instance_num=instance_num,\n                                                 tp=tp)\n    uvicorn.run(app=app, host=server_name, port=server_port, log_level='info')\n\n\nif __name__ == '__main__':\n    fire.Fire(main)\n", "path": "lmdeploy/serve/openai/api_server.py"}], "after_files": [{"content": "# Copyright (c) OpenMMLab. All rights reserved.\nimport json\nimport os\nimport time\nfrom http import HTTPStatus\nfrom typing import AsyncGenerator, Optional\n\nimport fire\nimport uvicorn\nfrom fastapi import BackgroundTasks, FastAPI, Request\nfrom fastapi.responses import JSONResponse, StreamingResponse\n\nfrom lmdeploy.serve.async_engine import AsyncEngine\nfrom lmdeploy.serve.openai.protocol import (  # noqa: E501\n    ChatCompletionRequest, ChatCompletionResponse,\n    ChatCompletionResponseChoice, ChatCompletionResponseStreamChoice,\n    ChatCompletionStreamResponse, ChatMessage, DeltaMessage, EmbeddingsRequest,\n    EmbeddingsResponse, ErrorResponse, GenerateRequest, ModelCard, ModelList,\n    ModelPermission, UsageInfo)\n\nos.environ['TM_LOG_LEVEL'] = 'ERROR'\n\n\nclass VariableInterface:\n    \"\"\"A IO interface maintaining variables.\"\"\"\n    async_engine: AsyncEngine = None\n    request_hosts = []\n\n\napp = FastAPI(docs_url='/')\n\n\ndef get_model_list():\n    \"\"\"Available models.\n\n    Only provided one now.\n    \"\"\"\n    return [VariableInterface.async_engine.tm_model.model_name]\n\n\n@app.get('/v1/models')\ndef available_models():\n    \"\"\"Show available models.\"\"\"\n    model_cards = []\n    for model_name in get_model_list():\n        model_cards.append(\n            ModelCard(id=model_name,\n                      root=model_name,\n                      permission=[ModelPermission()]))\n    return ModelList(data=model_cards)\n\n\ndef create_error_response(status: HTTPStatus, message: str):\n    \"\"\"Create error response according to http status and message.\n\n    Args:\n        status (HTTPStatus): HTTP status codes and reason phrases\n        message (str): error message\n    \"\"\"\n    return JSONResponse(\n        ErrorResponse(message=message,\n                      type='invalid_request_error',\n                      code=status.value).dict())\n\n\nasync def check_request(request) -> Optional[JSONResponse]:\n    \"\"\"Check if a request is valid.\"\"\"\n    if request.model in get_model_list():\n        return\n    ret = create_error_response(\n        HTTPStatus.NOT_FOUND, f'The model `{request.model}` does not exist.')\n    return ret\n\n\n@app.post('/v1/chat/completions')\nasync def chat_completions_v1(request: ChatCompletionRequest,\n                              raw_request: Request = None):\n    \"\"\"Completion API similar to OpenAI's API.\n\n    Refer to  `https://platform.openai.com/docs/api-reference/chat/create`\n    for the API specification.\n\n    The request should be a JSON object with the following fields:\n    - model: model name. Available from /v1/models.\n    - messages: string prompt or chat history in OpenAI format.\n    - temperature (float): to modulate the next token probability\n    - top_p (float): If set to float < 1, only the smallest set of most\n        probable tokens with probabilities that add up to top_p or higher\n        are kept for generation.\n    - n (int): How many chat completion choices to generate for each input\n        message. Only support one here.\n    - stream: whether to stream the results or not. Default to false.\n    - max_tokens (int): output token nums\n    - repetition_penalty (float): The parameter for repetition penalty.\n        1.0 means no penalty\n\n    Additional arguments supported by LMDeploy:\n    - renew_session (bool): Whether renew the session. Can be used when the\n        session length is exceeded.\n    - ignore_eos (bool): indicator for ignoring eos\n\n    Currently we do not support the following features:\n    - function_call (Users should implement this by themselves)\n    - logit_bias (not supported yet)\n    - presence_penalty (replaced with repetition_penalty)\n    - frequency_penalty (replaced with repetition_penalty)\n    \"\"\"\n    instance_id = int(raw_request.client.host.replace('.', ''))\n\n    error_check_ret = await check_request(request)\n    if error_check_ret is not None:\n        return error_check_ret\n\n    model_name = request.model\n    request_id = str(instance_id)\n    created_time = int(time.time())\n\n    result_generator = VariableInterface.async_engine.generate_openai(\n        request.messages,\n        instance_id,\n        True,  # always use stream to enable batching\n        request.renew_session,\n        request_output_len=request.max_tokens if request.max_tokens else 512,\n        stop=request.stop,\n        top_p=request.top_p,\n        temperature=request.temperature,\n        repetition_penalty=request.repetition_penalty,\n        ignore_eos=request.ignore_eos)\n\n    async def abort_request() -> None:\n        async for _ in VariableInterface.async_engine.generate_openai(\n                request.messages,\n                instance_id,\n                True,\n                request.renew_session,\n                stop=True):\n            pass\n\n    def create_stream_response_json(\n        index: int,\n        text: str,\n        finish_reason: Optional[str] = None,\n    ) -> str:\n        choice_data = ChatCompletionResponseStreamChoice(\n            index=index,\n            delta=DeltaMessage(role='assistant', content=text),\n            finish_reason=finish_reason,\n        )\n        response = ChatCompletionStreamResponse(\n            id=request_id,\n            created=created_time,\n            model=model_name,\n            choices=[choice_data],\n        )\n        response_json = response.json(ensure_ascii=False)\n\n        return response_json\n\n    async def completion_stream_generator() -> AsyncGenerator[str, None]:\n        # First chunk with role\n        for i in range(request.n):\n            choice_data = ChatCompletionResponseStreamChoice(\n                index=i,\n                delta=DeltaMessage(role='assistant'),\n                finish_reason=None,\n            )\n            chunk = ChatCompletionStreamResponse(id=request_id,\n                                                 choices=[choice_data],\n                                                 model=model_name)\n            data = chunk.json(exclude_unset=True, ensure_ascii=False)\n            yield f'data: {data}\\n\\n'\n\n        async for res in result_generator:\n            response_json = create_stream_response_json(\n                index=0,\n                text=res.response,\n            )\n            yield f'data: {response_json}\\n\\n'\n        yield 'data: [DONE]\\n\\n'\n\n    # Streaming response\n    if request.stream:\n        background_tasks = BackgroundTasks()\n        # Abort the request if the client disconnects.\n        background_tasks.add_task(abort_request)\n        return StreamingResponse(completion_stream_generator(),\n                                 media_type='text/event-stream',\n                                 background=background_tasks)\n\n    # Non-streaming response\n    final_res = None\n    text = ''\n    async for res in result_generator:\n        if await raw_request.is_disconnected():\n            # Abort the request if the client disconnects.\n            await abort_request()\n            return create_error_response(HTTPStatus.BAD_REQUEST,\n                                         'Client disconnected')\n        final_res = res\n        text += res.response\n    assert final_res is not None\n    choices = []\n    choice_data = ChatCompletionResponseChoice(\n        index=0,\n        message=ChatMessage(role='assistant', content=text),\n        finish_reason=final_res.finish_reason,\n    )\n    choices.append(choice_data)\n\n    total_tokens = sum([\n        final_res.history_token_len, final_res.input_token_len,\n        final_res.generate_token_len\n    ])\n    usage = UsageInfo(\n        prompt_tokens=final_res.input_token_len,\n        completion_tokens=final_res.generate_token_len,\n        total_tokens=total_tokens,\n    )\n    response = ChatCompletionResponse(\n        id=request_id,\n        created=created_time,\n        model=model_name,\n        choices=choices,\n        usage=usage,\n    )\n\n    return response\n\n\n@app.post('/v1/embeddings')\nasync def create_embeddings(request: EmbeddingsRequest,\n                            raw_request: Request = None):\n    \"\"\"Creates embeddings for the text.\"\"\"\n    error_check_ret = await check_request(request)\n    if error_check_ret is not None:\n        return error_check_ret\n\n    embedding = await VariableInterface.async_engine.get_embeddings(\n        request.input)\n    data = [{'object': 'embedding', 'embedding': embedding, 'index': 0}]\n    token_num = len(embedding)\n    return EmbeddingsResponse(\n        data=data,\n        model=request.model,\n        usage=UsageInfo(\n            prompt_tokens=token_num,\n            total_tokens=token_num,\n            completion_tokens=None,\n        ),\n    ).dict(exclude_none=True)\n\n\n@app.post('/generate')\nasync def generate(request: GenerateRequest, raw_request: Request = None):\n    \"\"\"Generate completion for the request.\n\n    The request should be a JSON object with the following fields:\n    - prompt: the prompt to use for the generation.\n    - instance_id: determine which instance will be called. If not specified\n        with a value other than -1, using host ip directly.\n    - sequence_start (bool): indicator for starting a sequence.\n    - sequence_end (bool): indicator for ending a sequence\n    - stream: whether to stream the results or not.\n    - stop: whether to stop the session response or not.\n    - request_output_len (int): output token nums\n    - step (int): the offset of the k/v cache\n    - top_p (float): If set to float < 1, only the smallest set of most\n        probable tokens with probabilities that add up to top_p or higher\n        are kept for generation.\n    - top_k (int): The number of the highest probability vocabulary\n        tokens to keep for top-k-filtering\n    - temperature (float): to modulate the next token probability\n    - repetition_penalty (float): The parameter for repetition penalty.\n        1.0 means no penalty\n    - ignore_eos (bool): indicator for ignoring eos\n    \"\"\"\n    if request.instance_id == -1:\n        instance_id = int(raw_request.client.host.replace('.', ''))\n        request.instance_id = instance_id\n\n    generation = VariableInterface.async_engine.generate(\n        request.prompt,\n        request.instance_id,\n        stream_response=True,  # always use stream to enable batching\n        sequence_start=request.sequence_start,\n        sequence_end=request.sequence_end,\n        request_output_len=request.request_output_len,\n        top_p=request.top_p,\n        top_k=request.top_k,\n        stop=request.stop,\n        temperature=request.temperature,\n        repetition_penalty=request.repetition_penalty,\n        ignore_eos=request.ignore_eos)\n\n    # Streaming case\n    async def stream_results() -> AsyncGenerator[bytes, None]:\n        async for out in generation:\n            ret = {\n                'text': out.response,\n                'tokens': out.generate_token_len,\n                'finish_reason': out.finish_reason\n            }\n            yield (json.dumps(ret) + '\\0').encode('utf-8')\n\n    if request.stream:\n        return StreamingResponse(stream_results())\n    else:\n        ret = {}\n        text = ''\n        tokens = 0\n        finish_reason = None\n        async for out in generation:\n            text += out.response\n            tokens = out.generate_token_len\n            finish_reason = out.finish_reason\n        ret = {'text': text, 'tokens': tokens, 'finish_reason': finish_reason}\n        return JSONResponse(ret)\n\n\ndef main(model_path: str,\n         server_name: str = 'localhost',\n         server_port: int = 23333,\n         instance_num: int = 32,\n         tp: int = 1):\n    \"\"\"An example to perform model inference through the command line\n    interface.\n\n    Args:\n        model_path (str): the path of the deployed model\n        server_name (str): host ip for serving\n        server_port (int): server port\n        instance_num (int): number of instances of turbomind model\n        tp (int): tensor parallel\n    \"\"\"\n    VariableInterface.async_engine = AsyncEngine(model_path=model_path,\n                                                 instance_num=instance_num,\n                                                 tp=tp)\n    uvicorn.run(app=app, host=server_name, port=server_port, log_level='info')\n\n\nif __name__ == '__main__':\n    fire.Fire(main)\n", "path": "lmdeploy/serve/openai/api_server.py"}]}