We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
the syntax in serving_completion.py is not compatible in python3.8
in #2529 @simon-mo introduce some python typing syntax, which is not compatible in python3.8

like  TypeTokenIDs = list[int] in https://github.com/vllm-project/vllm/blob/93b38bea5dd03e1b140ca997dfaadef86f8f1855/vllm/entrypoints/openai/serving_completion.py#L22C1-L22C25

which should be TypeTokenIDs=List[int] in python3.8.

could you please fix it?

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `vllm/entrypoints/openai/serving_completion.py`
Content:
```
1 import asyncio
2 import time
3 from fastapi import Request
4 from typing import AsyncGenerator, AsyncIterator, Callable, List, Optional
5 from vllm.logger import init_logger
6 from vllm.utils import random_uuid
7 from vllm.engine.async_llm_engine import AsyncLLMEngine
8 from .protocol import (
9     CompletionRequest,
10     CompletionResponse,
11     CompletionResponseChoice,
12     CompletionResponseStreamChoice,
13     CompletionStreamResponse,
14     LogProbs,
15     UsageInfo,
16 )
17 from vllm.outputs import RequestOutput
18 from vllm.entrypoints.openai.serving_engine import OpenAIServing
19 
20 logger = init_logger(__name__)
21 
22 TypeTokenIDs = list[int]
23 TypeTopLogProbs = List[Optional[dict[int, float]]]
24 TypeCreateLogProbsFn = Callable[
25     [TypeTokenIDs, TypeTopLogProbs, Optional[int], int], LogProbs]
26 
27 
28 async def completion_stream_generator(
29     request: CompletionRequest,
30     raw_request: Request,
31     on_abort,
32     result_generator: AsyncIterator[tuple[int, RequestOutput]],
33     create_logprobs_fn: TypeCreateLogProbsFn,
34     request_id: str,
35     created_time: int,
36     model_name: str,
37     num_prompts: int,
38 ) -> AsyncGenerator[str, None]:
39     previous_texts = [""] * request.n * num_prompts
40     previous_num_tokens = [0] * request.n * num_prompts
41     has_echoed = [False] * request.n * num_prompts
42 
43     async for prompt_idx, res in result_generator:
44 
45         # Abort the request if the client disconnects.
46         if await raw_request.is_disconnected():
47             await on_abort(f"{request_id}-{prompt_idx}")
48             raise StopAsyncIteration()
49 
50         for output in res.outputs:
51             i = output.index + prompt_idx * request.n
52             # TODO(simon): optimize the performance by avoiding full text O(n^2) sending.
53 
54             if request.echo and request.max_tokens == 0:
55                 # only return the prompt
56                 delta_text = res.prompt
57                 delta_token_ids = res.prompt_token_ids
58                 top_logprobs = res.prompt_logprobs
59                 has_echoed[i] = True
60             elif request.echo and request.max_tokens > 0 and not has_echoed[i]:
61                 # echo the prompt and first token
62                 delta_text = res.prompt + output.text
63                 delta_token_ids = res.prompt_token_ids + output.token_ids
64                 top_logprobs = res.prompt_logprobs + (output.logprobs or [])
65                 has_echoed[i] = True
66             else:
67                 # return just the delta
68                 delta_text = output.text[len(previous_texts[i]):]
69                 delta_token_ids = output.token_ids[previous_num_tokens[i]:]
70                 top_logprobs = output.logprobs[
71                     previous_num_tokens[i]:] if output.logprobs else None
72 
73             if request.logprobs is not None:
74                 assert top_logprobs is not None, "top_logprobs must be provided when logprobs is requested"
75                 logprobs = create_logprobs_fn(
76                     token_ids=delta_token_ids,
77                     top_logprobs=top_logprobs,
78                     num_output_top_logprobs=request.logprobs,
79                     initial_text_offset=len(previous_texts[i]),
80                 )
81             else:
82                 logprobs = None
83 
84             previous_texts[i] = output.text
85             previous_num_tokens[i] = len(output.token_ids)
86             finish_reason = output.finish_reason
87             response_json = CompletionStreamResponse(
88                 id=request_id,
89                 created=created_time,
90                 model=model_name,
91                 choices=[
92                     CompletionResponseStreamChoice(
93                         index=i,
94                         text=delta_text,
95                         logprobs=logprobs,
96                         finish_reason=finish_reason,
97                     )
98                 ]).model_dump_json(exclude_unset=True)
99             yield f"data: {response_json}\n\n"
100 
101             if output.finish_reason is not None:  # return final usage
102                 logprobs = LogProbs() if request.logprobs is not None else None
103                 prompt_tokens = len(res.prompt_token_ids)
104                 completion_tokens = len(output.token_ids)
105                 final_usage = UsageInfo(
106                     prompt_tokens=prompt_tokens,
107                     completion_tokens=completion_tokens,
108                     total_tokens=prompt_tokens + completion_tokens,
109                 )
110                 response_json = CompletionStreamResponse(
111                     id=request_id,
112                     created=created_time,
113                     model=model_name,
114                     choices=[
115                         CompletionResponseStreamChoice(
116                             index=i,
117                             text="",
118                             logprobs=logprobs,
119                             finish_reason=output.finish_reason,
120                         )
121                     ],
122                     usage=final_usage,
123                 ).model_dump_json(exclude_unset=True)
124                 yield f"data: {response_json}\n\n"
125 
126     yield "data: [DONE]\n\n"
127 
128 
129 def parse_prompt_format(prompt) -> tuple[bool, list]:
130     # get the prompt, openai supports the following
131     # "a string, array of strings, array of tokens, or array of token arrays."
132     prompt_is_tokens = False
133     prompts = [prompt]  # case 1: a string
134     if isinstance(prompt, list):
135         if len(prompt) == 0:
136             raise ValueError("please provide at least one prompt")
137         elif isinstance(prompt[0], str):
138             prompt_is_tokens = False
139             prompts = prompt  # case 2: array of strings
140         elif isinstance(prompt[0], int):
141             prompt_is_tokens = True
142             prompts = [prompt]  # case 3: array of tokens
143         elif isinstance(prompt[0], list) and isinstance(prompt[0][0], int):
144             prompt_is_tokens = True
145             prompts = prompt  # case 4: array of token arrays
146         else:
147             raise ValueError(
148                 "prompt must be a string, array of strings, array of tokens, or array of token arrays"
149             )
150     return prompt_is_tokens, prompts
151 
152 
153 def request_output_to_completion_response(
154     final_res_batch: list[RequestOutput],
155     request: CompletionRequest,
156     create_logprobs_fn: TypeCreateLogProbsFn,
157     request_id: str,
158     created_time: int,
159     model_name: str,
160 ) -> CompletionResponse:
161     choices = []
162     num_prompt_tokens = 0
163     num_generated_tokens = 0
164     for final_res in final_res_batch:
165         assert final_res is not None
166         prompt_token_ids = final_res.prompt_token_ids
167         prompt_logprobs = final_res.prompt_logprobs
168         prompt_text = final_res.prompt
169 
170         for output in final_res.outputs:
171             if request.echo and request.max_tokens == 0:
172                 token_ids = prompt_token_ids
173                 top_logprobs = prompt_logprobs
174                 output_text = prompt_text
175             elif request.echo and request.max_tokens > 0:
176                 token_ids = prompt_token_ids + output.token_ids
177                 top_logprobs = prompt_logprobs + output.logprobs
178                 output_text = prompt_text + output.text
179             else:
180                 token_ids = output.token_ids
181                 top_logprobs = output.logprobs
182                 output_text = output.text
183 
184             if request.logprobs is not None:
185                 logprobs = create_logprobs_fn(
186                     token_ids=token_ids,
187                     top_logprobs=top_logprobs,
188                     num_output_top_logprobs=request.logprobs,
189                 )
190             else:
191                 logprobs = None
192 
193             choice_data = CompletionResponseChoice(
194                 index=len(choices),
195                 text=output_text,
196                 logprobs=logprobs,
197                 finish_reason=output.finish_reason,
198             )
199             choices.append(choice_data)
200 
201         num_prompt_tokens += len(prompt_token_ids)
202         num_generated_tokens += sum(
203             len(output.token_ids) for output in final_res.outputs)
204 
205     usage = UsageInfo(
206         prompt_tokens=num_prompt_tokens,
207         completion_tokens=num_generated_tokens,
208         total_tokens=num_prompt_tokens + num_generated_tokens,
209     )
210 
211     return CompletionResponse(
212         id=request_id,
213         created=created_time,
214         model=model_name,
215         choices=choices,
216         usage=usage,
217     )
218 
219 
220 def merge_async_iterators(*iterators):
221     """Merge multiple asynchronous iterators into a single iterator.
222 
223     This method handle the case where some iterators finish before others.
224     When it yields, it yields a tuple (i, item) where i is the index of the
225     iterator that yields the item.
226     """
227     queue = asyncio.Queue()
228 
229     finished = [False] * len(iterators)
230 
231     async def producer(i, iterator):
232         async for item in iterator:
233             await queue.put((i, item))
234         finished[i] = True
235 
236     _tasks = [
237         asyncio.create_task(producer(i, iterator))
238         for i, iterator in enumerate(iterators)
239     ]
240 
241     async def consumer():
242         while not all(finished) or not queue.empty():
243             item = await queue.get()
244             yield item
245         await asyncio.gather(*_tasks)
246 
247     return consumer()
248 
249 
250 class OpenAIServingCompletion(OpenAIServing):
251 
252     def __init__(self, engine: AsyncLLMEngine, served_model: str):
253         super().__init__(engine=engine, served_model=served_model)
254 
255     async def create_completion(self, request: CompletionRequest,
256                                 raw_request: Request):
257         """Completion API similar to OpenAI's API.
258 
259         See https://platform.openai.com/docs/api-reference/completions/create
260         for the API specification. This API mimics the OpenAI Completion API.
261 
262         NOTE: Currently we do not support the following features:
263             - suffix (the language models we currently support do not support
264             suffix)
265             - logit_bias (to be supported by vLLM engine)
266         """
267         error_check_ret = await self._check_model(request)
268         if error_check_ret is not None:
269             return error_check_ret
270 
271         # Return error for unsupported features.
272         if request.suffix is not None:
273             return self.create_error_response(
274                 "suffix is not currently supported")
275         if request.logit_bias is not None and len(request.logit_bias) > 0:
276             return self.create_error_response(
277                 "logit_bias is not currently supported")
278 
279         model_name = request.model
280         request_id = f"cmpl-{random_uuid()}"
281         created_time = int(time.monotonic())
282 
283         # Schedule the request and get the result generator.
284         generators = []
285         try:
286             sampling_params = request.to_sampling_params()
287             prompt_is_tokens, prompts = parse_prompt_format(request.prompt)
288 
289             for i, prompt in enumerate(prompts):
290                 if prompt_is_tokens:
291                     input_ids = self._validate_prompt_and_tokenize(
292                         request, prompt_ids=prompt)
293                 else:
294                     input_ids = self._validate_prompt_and_tokenize(
295                         request, prompt=prompt)
296 
297                 generators.append(
298                     self.engine.generate(None,
299                                          sampling_params,
300                                          f"{request_id}-{i}",
301                                          prompt_token_ids=input_ids))
302         except ValueError as e:
303             return self.create_error_response(str(e))
304 
305         result_generator: AsyncIterator[tuple[
306             int, RequestOutput]] = merge_async_iterators(*generators)
307 
308         # Similar to the OpenAI API, when n != best_of, we do not stream the
309         # results. In addition, we do not stream the results when use beam search.
310         stream = (request.stream
311                   and (request.best_of is None or request.n == request.best_of)
312                   and not request.use_beam_search)
313 
314         # Streaming response
315         if stream:
316             return completion_stream_generator(request,
317                                                raw_request,
318                                                self.engine.abort,
319                                                result_generator,
320                                                self._create_logprobs,
321                                                request_id,
322                                                created_time,
323                                                model_name,
324                                                num_prompts=len(prompts))
325 
326         # Non-streaming response
327         final_res_batch: RequestOutput = [None] * len(prompts)
328         async for i, res in result_generator:
329             if await raw_request.is_disconnected():
330                 # Abort the request if the client disconnects.
331                 await self.engine.abort(f"{request_id}-{i}")
332                 return self.create_error_response("Client disconnected")
333             final_res_batch[i] = res
334         response = request_output_to_completion_response(
335             final_res_batch, request, self._create_logprobs, request_id,
336             created_time, model_name)
337 
338         # When user requests streaming but we don't stream, we still need to
339         # return a streaming response with a single event.
340         if request.stream:
341             response_json = response.model_dump_json()
342 
343             async def fake_stream_generator() -> AsyncGenerator[str, None]:
344                 yield f"data: {response_json}\n\n"
345                 yield "data: [DONE]\n\n"
346 
347             return fake_stream_generator()
348 
349         return response
350
```
--- 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/vllm/entrypoints/openai/serving_completion.py b/vllm/entrypoints/openai/serving_completion.py
--- a/vllm/entrypoints/openai/serving_completion.py
+++ b/vllm/entrypoints/openai/serving_completion.py
@@ -1,7 +1,7 @@
 import asyncio
 import time
 from fastapi import Request
-from typing import AsyncGenerator, AsyncIterator, Callable, List, Optional
+from typing import AsyncGenerator, AsyncIterator, Callable, List, Optional, Dict, Tuple
 from vllm.logger import init_logger
 from vllm.utils import random_uuid
 from vllm.engine.async_llm_engine import AsyncLLMEngine
@@ -19,8 +19,8 @@
 
 logger = init_logger(__name__)
 
-TypeTokenIDs = list[int]
-TypeTopLogProbs = List[Optional[dict[int, float]]]
+TypeTokenIDs = List[int]
+TypeTopLogProbs = List[Optional[Dict[int, float]]]
 TypeCreateLogProbsFn = Callable[
     [TypeTokenIDs, TypeTopLogProbs, Optional[int], int], LogProbs]
 
@@ -29,7 +29,7 @@
     request: CompletionRequest,
     raw_request: Request,
     on_abort,
-    result_generator: AsyncIterator[tuple[int, RequestOutput]],
+    result_generator: AsyncIterator[Tuple[int, RequestOutput]],
     create_logprobs_fn: TypeCreateLogProbsFn,
     request_id: str,
     created_time: int,
@@ -126,7 +126,7 @@
     yield "data: [DONE]\n\n"
 
 
-def parse_prompt_format(prompt) -> tuple[bool, list]:
+def parse_prompt_format(prompt) -> Tuple[bool, list]:
     # get the prompt, openai supports the following
     # "a string, array of strings, array of tokens, or array of token arrays."
     prompt_is_tokens = False
@@ -151,7 +151,7 @@
 
 
 def request_output_to_completion_response(
-    final_res_batch: list[RequestOutput],
+    final_res_batch: List[RequestOutput],
     request: CompletionRequest,
     create_logprobs_fn: TypeCreateLogProbsFn,
     request_id: str,
@@ -302,7 +302,7 @@
         except ValueError as e:
             return self.create_error_response(str(e))
 
-        result_generator: AsyncIterator[tuple[
+        result_generator: AsyncIterator[Tuple[
             int, RequestOutput]] = merge_async_iterators(*generators)
 
         # Similar to the OpenAI API, when n != best_of, we do not stream the

{"golden_diff": "diff --git a/vllm/entrypoints/openai/serving_completion.py b/vllm/entrypoints/openai/serving_completion.py\n--- a/vllm/entrypoints/openai/serving_completion.py\n+++ b/vllm/entrypoints/openai/serving_completion.py\n@@ -1,7 +1,7 @@\n import asyncio\n import time\n from fastapi import Request\n-from typing import AsyncGenerator, AsyncIterator, Callable, List, Optional\n+from typing import AsyncGenerator, AsyncIterator, Callable, List, Optional, Dict, Tuple\n from vllm.logger import init_logger\n from vllm.utils import random_uuid\n from vllm.engine.async_llm_engine import AsyncLLMEngine\n@@ -19,8 +19,8 @@\n \n logger = init_logger(__name__)\n \n-TypeTokenIDs = list[int]\n-TypeTopLogProbs = List[Optional[dict[int, float]]]\n+TypeTokenIDs = List[int]\n+TypeTopLogProbs = List[Optional[Dict[int, float]]]\n TypeCreateLogProbsFn = Callable[\n     [TypeTokenIDs, TypeTopLogProbs, Optional[int], int], LogProbs]\n \n@@ -29,7 +29,7 @@\n     request: CompletionRequest,\n     raw_request: Request,\n     on_abort,\n-    result_generator: AsyncIterator[tuple[int, RequestOutput]],\n+    result_generator: AsyncIterator[Tuple[int, RequestOutput]],\n     create_logprobs_fn: TypeCreateLogProbsFn,\n     request_id: str,\n     created_time: int,\n@@ -126,7 +126,7 @@\n     yield \"data: [DONE]\\n\\n\"\n \n \n-def parse_prompt_format(prompt) -> tuple[bool, list]:\n+def parse_prompt_format(prompt) -> Tuple[bool, list]:\n     # get the prompt, openai supports the following\n     # \"a string, array of strings, array of tokens, or array of token arrays.\"\n     prompt_is_tokens = False\n@@ -151,7 +151,7 @@\n \n \n def request_output_to_completion_response(\n-    final_res_batch: list[RequestOutput],\n+    final_res_batch: List[RequestOutput],\n     request: CompletionRequest,\n     create_logprobs_fn: TypeCreateLogProbsFn,\n     request_id: str,\n@@ -302,7 +302,7 @@\n         except ValueError as e:\n             return self.create_error_response(str(e))\n \n-        result_generator: AsyncIterator[tuple[\n+        result_generator: AsyncIterator[Tuple[\n             int, RequestOutput]] = merge_async_iterators(*generators)\n \n         # Similar to the OpenAI API, when n != best_of, we do not stream the\n", "issue": "the syntax in serving_completion.py is not compatible in python3.8\nin #2529 @simon-mo introduce some python typing syntax, which is not compatible in python3.8\r\n\r\nlike  TypeTokenIDs = list[int] in https://github.com/vllm-project/vllm/blob/93b38bea5dd03e1b140ca997dfaadef86f8f1855/vllm/entrypoints/openai/serving_completion.py#L22C1-L22C25\r\n\r\nwhich should be TypeTokenIDs=List[int] in python3.8.\r\n\r\ncould you please fix it?\n", "before_files": [{"content": "import asyncio\nimport time\nfrom fastapi import Request\nfrom typing import AsyncGenerator, AsyncIterator, Callable, List, Optional\nfrom vllm.logger import init_logger\nfrom vllm.utils import random_uuid\nfrom vllm.engine.async_llm_engine import AsyncLLMEngine\nfrom .protocol import (\n    CompletionRequest,\n    CompletionResponse,\n    CompletionResponseChoice,\n    CompletionResponseStreamChoice,\n    CompletionStreamResponse,\n    LogProbs,\n    UsageInfo,\n)\nfrom vllm.outputs import RequestOutput\nfrom vllm.entrypoints.openai.serving_engine import OpenAIServing\n\nlogger = init_logger(__name__)\n\nTypeTokenIDs = list[int]\nTypeTopLogProbs = List[Optional[dict[int, float]]]\nTypeCreateLogProbsFn = Callable[\n    [TypeTokenIDs, TypeTopLogProbs, Optional[int], int], LogProbs]\n\n\nasync def completion_stream_generator(\n    request: CompletionRequest,\n    raw_request: Request,\n    on_abort,\n    result_generator: AsyncIterator[tuple[int, RequestOutput]],\n    create_logprobs_fn: TypeCreateLogProbsFn,\n    request_id: str,\n    created_time: int,\n    model_name: str,\n    num_prompts: int,\n) -> AsyncGenerator[str, None]:\n    previous_texts = [\"\"] * request.n * num_prompts\n    previous_num_tokens = [0] * request.n * num_prompts\n    has_echoed = [False] * request.n * num_prompts\n\n    async for prompt_idx, res in result_generator:\n\n        # Abort the request if the client disconnects.\n        if await raw_request.is_disconnected():\n            await on_abort(f\"{request_id}-{prompt_idx}\")\n            raise StopAsyncIteration()\n\n        for output in res.outputs:\n            i = output.index + prompt_idx * request.n\n            # TODO(simon): optimize the performance by avoiding full text O(n^2) sending.\n\n            if request.echo and request.max_tokens == 0:\n                # only return the prompt\n                delta_text = res.prompt\n                delta_token_ids = res.prompt_token_ids\n                top_logprobs = res.prompt_logprobs\n                has_echoed[i] = True\n            elif request.echo and request.max_tokens > 0 and not has_echoed[i]:\n                # echo the prompt and first token\n                delta_text = res.prompt + output.text\n                delta_token_ids = res.prompt_token_ids + output.token_ids\n                top_logprobs = res.prompt_logprobs + (output.logprobs or [])\n                has_echoed[i] = True\n            else:\n                # return just the delta\n                delta_text = output.text[len(previous_texts[i]):]\n                delta_token_ids = output.token_ids[previous_num_tokens[i]:]\n                top_logprobs = output.logprobs[\n                    previous_num_tokens[i]:] if output.logprobs else None\n\n            if request.logprobs is not None:\n                assert top_logprobs is not None, \"top_logprobs must be provided when logprobs is requested\"\n                logprobs = create_logprobs_fn(\n                    token_ids=delta_token_ids,\n                    top_logprobs=top_logprobs,\n                    num_output_top_logprobs=request.logprobs,\n                    initial_text_offset=len(previous_texts[i]),\n                )\n            else:\n                logprobs = None\n\n            previous_texts[i] = output.text\n            previous_num_tokens[i] = len(output.token_ids)\n            finish_reason = output.finish_reason\n            response_json = CompletionStreamResponse(\n                id=request_id,\n                created=created_time,\n                model=model_name,\n                choices=[\n                    CompletionResponseStreamChoice(\n                        index=i,\n                        text=delta_text,\n                        logprobs=logprobs,\n                        finish_reason=finish_reason,\n                    )\n                ]).model_dump_json(exclude_unset=True)\n            yield f\"data: {response_json}\\n\\n\"\n\n            if output.finish_reason is not None:  # return final usage\n                logprobs = LogProbs() if request.logprobs is not None else None\n                prompt_tokens = len(res.prompt_token_ids)\n                completion_tokens = len(output.token_ids)\n                final_usage = UsageInfo(\n                    prompt_tokens=prompt_tokens,\n                    completion_tokens=completion_tokens,\n                    total_tokens=prompt_tokens + completion_tokens,\n                )\n                response_json = CompletionStreamResponse(\n                    id=request_id,\n                    created=created_time,\n                    model=model_name,\n                    choices=[\n                        CompletionResponseStreamChoice(\n                            index=i,\n                            text=\"\",\n                            logprobs=logprobs,\n                            finish_reason=output.finish_reason,\n                        )\n                    ],\n                    usage=final_usage,\n                ).model_dump_json(exclude_unset=True)\n                yield f\"data: {response_json}\\n\\n\"\n\n    yield \"data: [DONE]\\n\\n\"\n\n\ndef parse_prompt_format(prompt) -> tuple[bool, list]:\n    # get the prompt, openai supports the following\n    # \"a string, array of strings, array of tokens, or array of token arrays.\"\n    prompt_is_tokens = False\n    prompts = [prompt]  # case 1: a string\n    if isinstance(prompt, list):\n        if len(prompt) == 0:\n            raise ValueError(\"please provide at least one prompt\")\n        elif isinstance(prompt[0], str):\n            prompt_is_tokens = False\n            prompts = prompt  # case 2: array of strings\n        elif isinstance(prompt[0], int):\n            prompt_is_tokens = True\n            prompts = [prompt]  # case 3: array of tokens\n        elif isinstance(prompt[0], list) and isinstance(prompt[0][0], int):\n            prompt_is_tokens = True\n            prompts = prompt  # case 4: array of token arrays\n        else:\n            raise ValueError(\n                \"prompt must be a string, array of strings, array of tokens, or array of token arrays\"\n            )\n    return prompt_is_tokens, prompts\n\n\ndef request_output_to_completion_response(\n    final_res_batch: list[RequestOutput],\n    request: CompletionRequest,\n    create_logprobs_fn: TypeCreateLogProbsFn,\n    request_id: str,\n    created_time: int,\n    model_name: str,\n) -> CompletionResponse:\n    choices = []\n    num_prompt_tokens = 0\n    num_generated_tokens = 0\n    for final_res in final_res_batch:\n        assert final_res is not None\n        prompt_token_ids = final_res.prompt_token_ids\n        prompt_logprobs = final_res.prompt_logprobs\n        prompt_text = final_res.prompt\n\n        for output in final_res.outputs:\n            if request.echo and request.max_tokens == 0:\n                token_ids = prompt_token_ids\n                top_logprobs = prompt_logprobs\n                output_text = prompt_text\n            elif request.echo and request.max_tokens > 0:\n                token_ids = prompt_token_ids + output.token_ids\n                top_logprobs = prompt_logprobs + output.logprobs\n                output_text = prompt_text + output.text\n            else:\n                token_ids = output.token_ids\n                top_logprobs = output.logprobs\n                output_text = output.text\n\n            if request.logprobs is not None:\n                logprobs = create_logprobs_fn(\n                    token_ids=token_ids,\n                    top_logprobs=top_logprobs,\n                    num_output_top_logprobs=request.logprobs,\n                )\n            else:\n                logprobs = None\n\n            choice_data = CompletionResponseChoice(\n                index=len(choices),\n                text=output_text,\n                logprobs=logprobs,\n                finish_reason=output.finish_reason,\n            )\n            choices.append(choice_data)\n\n        num_prompt_tokens += len(prompt_token_ids)\n        num_generated_tokens += sum(\n            len(output.token_ids) for output in final_res.outputs)\n\n    usage = UsageInfo(\n        prompt_tokens=num_prompt_tokens,\n        completion_tokens=num_generated_tokens,\n        total_tokens=num_prompt_tokens + num_generated_tokens,\n    )\n\n    return CompletionResponse(\n        id=request_id,\n        created=created_time,\n        model=model_name,\n        choices=choices,\n        usage=usage,\n    )\n\n\ndef merge_async_iterators(*iterators):\n    \"\"\"Merge multiple asynchronous iterators into a single iterator.\n\n    This method handle the case where some iterators finish before others.\n    When it yields, it yields a tuple (i, item) where i is the index of the\n    iterator that yields the item.\n    \"\"\"\n    queue = asyncio.Queue()\n\n    finished = [False] * len(iterators)\n\n    async def producer(i, iterator):\n        async for item in iterator:\n            await queue.put((i, item))\n        finished[i] = True\n\n    _tasks = [\n        asyncio.create_task(producer(i, iterator))\n        for i, iterator in enumerate(iterators)\n    ]\n\n    async def consumer():\n        while not all(finished) or not queue.empty():\n            item = await queue.get()\n            yield item\n        await asyncio.gather(*_tasks)\n\n    return consumer()\n\n\nclass OpenAIServingCompletion(OpenAIServing):\n\n    def __init__(self, engine: AsyncLLMEngine, served_model: str):\n        super().__init__(engine=engine, served_model=served_model)\n\n    async def create_completion(self, request: CompletionRequest,\n                                raw_request: Request):\n        \"\"\"Completion API similar to OpenAI's API.\n\n        See https://platform.openai.com/docs/api-reference/completions/create\n        for the API specification. This API mimics the OpenAI Completion API.\n\n        NOTE: Currently we do not support the following features:\n            - suffix (the language models we currently support do not support\n            suffix)\n            - logit_bias (to be supported by vLLM engine)\n        \"\"\"\n        error_check_ret = await self._check_model(request)\n        if error_check_ret is not None:\n            return error_check_ret\n\n        # Return error for unsupported features.\n        if request.suffix is not None:\n            return self.create_error_response(\n                \"suffix is not currently supported\")\n        if request.logit_bias is not None and len(request.logit_bias) > 0:\n            return self.create_error_response(\n                \"logit_bias is not currently supported\")\n\n        model_name = request.model\n        request_id = f\"cmpl-{random_uuid()}\"\n        created_time = int(time.monotonic())\n\n        # Schedule the request and get the result generator.\n        generators = []\n        try:\n            sampling_params = request.to_sampling_params()\n            prompt_is_tokens, prompts = parse_prompt_format(request.prompt)\n\n            for i, prompt in enumerate(prompts):\n                if prompt_is_tokens:\n                    input_ids = self._validate_prompt_and_tokenize(\n                        request, prompt_ids=prompt)\n                else:\n                    input_ids = self._validate_prompt_and_tokenize(\n                        request, prompt=prompt)\n\n                generators.append(\n                    self.engine.generate(None,\n                                         sampling_params,\n                                         f\"{request_id}-{i}\",\n                                         prompt_token_ids=input_ids))\n        except ValueError as e:\n            return self.create_error_response(str(e))\n\n        result_generator: AsyncIterator[tuple[\n            int, RequestOutput]] = merge_async_iterators(*generators)\n\n        # Similar to the OpenAI API, when n != best_of, we do not stream the\n        # results. In addition, we do not stream the results when use beam search.\n        stream = (request.stream\n                  and (request.best_of is None or request.n == request.best_of)\n                  and not request.use_beam_search)\n\n        # Streaming response\n        if stream:\n            return completion_stream_generator(request,\n                                               raw_request,\n                                               self.engine.abort,\n                                               result_generator,\n                                               self._create_logprobs,\n                                               request_id,\n                                               created_time,\n                                               model_name,\n                                               num_prompts=len(prompts))\n\n        # Non-streaming response\n        final_res_batch: RequestOutput = [None] * len(prompts)\n        async for i, res in result_generator:\n            if await raw_request.is_disconnected():\n                # Abort the request if the client disconnects.\n                await self.engine.abort(f\"{request_id}-{i}\")\n                return self.create_error_response(\"Client disconnected\")\n            final_res_batch[i] = res\n        response = request_output_to_completion_response(\n            final_res_batch, request, self._create_logprobs, request_id,\n            created_time, model_name)\n\n        # When user requests streaming but we don't stream, we still need to\n        # return a streaming response with a single event.\n        if request.stream:\n            response_json = response.model_dump_json()\n\n            async def fake_stream_generator() -> AsyncGenerator[str, None]:\n                yield f\"data: {response_json}\\n\\n\"\n                yield \"data: [DONE]\\n\\n\"\n\n            return fake_stream_generator()\n\n        return response\n", "path": "vllm/entrypoints/openai/serving_completion.py"}], "after_files": [{"content": "import asyncio\nimport time\nfrom fastapi import Request\nfrom typing import AsyncGenerator, AsyncIterator, Callable, List, Optional, Dict, Tuple\nfrom vllm.logger import init_logger\nfrom vllm.utils import random_uuid\nfrom vllm.engine.async_llm_engine import AsyncLLMEngine\nfrom .protocol import (\n    CompletionRequest,\n    CompletionResponse,\n    CompletionResponseChoice,\n    CompletionResponseStreamChoice,\n    CompletionStreamResponse,\n    LogProbs,\n    UsageInfo,\n)\nfrom vllm.outputs import RequestOutput\nfrom vllm.entrypoints.openai.serving_engine import OpenAIServing\n\nlogger = init_logger(__name__)\n\nTypeTokenIDs = List[int]\nTypeTopLogProbs = List[Optional[Dict[int, float]]]\nTypeCreateLogProbsFn = Callable[\n    [TypeTokenIDs, TypeTopLogProbs, Optional[int], int], LogProbs]\n\n\nasync def completion_stream_generator(\n    request: CompletionRequest,\n    raw_request: Request,\n    on_abort,\n    result_generator: AsyncIterator[Tuple[int, RequestOutput]],\n    create_logprobs_fn: TypeCreateLogProbsFn,\n    request_id: str,\n    created_time: int,\n    model_name: str,\n    num_prompts: int,\n) -> AsyncGenerator[str, None]:\n    previous_texts = [\"\"] * request.n * num_prompts\n    previous_num_tokens = [0] * request.n * num_prompts\n    has_echoed = [False] * request.n * num_prompts\n\n    async for prompt_idx, res in result_generator:\n\n        # Abort the request if the client disconnects.\n        if await raw_request.is_disconnected():\n            await on_abort(f\"{request_id}-{prompt_idx}\")\n            raise StopAsyncIteration()\n\n        for output in res.outputs:\n            i = output.index + prompt_idx * request.n\n            # TODO(simon): optimize the performance by avoiding full text O(n^2) sending.\n\n            if request.echo and request.max_tokens == 0:\n                # only return the prompt\n                delta_text = res.prompt\n                delta_token_ids = res.prompt_token_ids\n                top_logprobs = res.prompt_logprobs\n                has_echoed[i] = True\n            elif request.echo and request.max_tokens > 0 and not has_echoed[i]:\n                # echo the prompt and first token\n                delta_text = res.prompt + output.text\n                delta_token_ids = res.prompt_token_ids + output.token_ids\n                top_logprobs = res.prompt_logprobs + (output.logprobs or [])\n                has_echoed[i] = True\n            else:\n                # return just the delta\n                delta_text = output.text[len(previous_texts[i]):]\n                delta_token_ids = output.token_ids[previous_num_tokens[i]:]\n                top_logprobs = output.logprobs[\n                    previous_num_tokens[i]:] if output.logprobs else None\n\n            if request.logprobs is not None:\n                assert top_logprobs is not None, \"top_logprobs must be provided when logprobs is requested\"\n                logprobs = create_logprobs_fn(\n                    token_ids=delta_token_ids,\n                    top_logprobs=top_logprobs,\n                    num_output_top_logprobs=request.logprobs,\n                    initial_text_offset=len(previous_texts[i]),\n                )\n            else:\n                logprobs = None\n\n            previous_texts[i] = output.text\n            previous_num_tokens[i] = len(output.token_ids)\n            finish_reason = output.finish_reason\n            response_json = CompletionStreamResponse(\n                id=request_id,\n                created=created_time,\n                model=model_name,\n                choices=[\n                    CompletionResponseStreamChoice(\n                        index=i,\n                        text=delta_text,\n                        logprobs=logprobs,\n                        finish_reason=finish_reason,\n                    )\n                ]).model_dump_json(exclude_unset=True)\n            yield f\"data: {response_json}\\n\\n\"\n\n            if output.finish_reason is not None:  # return final usage\n                logprobs = LogProbs() if request.logprobs is not None else None\n                prompt_tokens = len(res.prompt_token_ids)\n                completion_tokens = len(output.token_ids)\n                final_usage = UsageInfo(\n                    prompt_tokens=prompt_tokens,\n                    completion_tokens=completion_tokens,\n                    total_tokens=prompt_tokens + completion_tokens,\n                )\n                response_json = CompletionStreamResponse(\n                    id=request_id,\n                    created=created_time,\n                    model=model_name,\n                    choices=[\n                        CompletionResponseStreamChoice(\n                            index=i,\n                            text=\"\",\n                            logprobs=logprobs,\n                            finish_reason=output.finish_reason,\n                        )\n                    ],\n                    usage=final_usage,\n                ).model_dump_json(exclude_unset=True)\n                yield f\"data: {response_json}\\n\\n\"\n\n    yield \"data: [DONE]\\n\\n\"\n\n\ndef parse_prompt_format(prompt) -> Tuple[bool, list]:\n    # get the prompt, openai supports the following\n    # \"a string, array of strings, array of tokens, or array of token arrays.\"\n    prompt_is_tokens = False\n    prompts = [prompt]  # case 1: a string\n    if isinstance(prompt, list):\n        if len(prompt) == 0:\n            raise ValueError(\"please provide at least one prompt\")\n        elif isinstance(prompt[0], str):\n            prompt_is_tokens = False\n            prompts = prompt  # case 2: array of strings\n        elif isinstance(prompt[0], int):\n            prompt_is_tokens = True\n            prompts = [prompt]  # case 3: array of tokens\n        elif isinstance(prompt[0], list) and isinstance(prompt[0][0], int):\n            prompt_is_tokens = True\n            prompts = prompt  # case 4: array of token arrays\n        else:\n            raise ValueError(\n                \"prompt must be a string, array of strings, array of tokens, or array of token arrays\"\n            )\n    return prompt_is_tokens, prompts\n\n\ndef request_output_to_completion_response(\n    final_res_batch: List[RequestOutput],\n    request: CompletionRequest,\n    create_logprobs_fn: TypeCreateLogProbsFn,\n    request_id: str,\n    created_time: int,\n    model_name: str,\n) -> CompletionResponse:\n    choices = []\n    num_prompt_tokens = 0\n    num_generated_tokens = 0\n    for final_res in final_res_batch:\n        assert final_res is not None\n        prompt_token_ids = final_res.prompt_token_ids\n        prompt_logprobs = final_res.prompt_logprobs\n        prompt_text = final_res.prompt\n\n        for output in final_res.outputs:\n            if request.echo and request.max_tokens == 0:\n                token_ids = prompt_token_ids\n                top_logprobs = prompt_logprobs\n                output_text = prompt_text\n            elif request.echo and request.max_tokens > 0:\n                token_ids = prompt_token_ids + output.token_ids\n                top_logprobs = prompt_logprobs + output.logprobs\n                output_text = prompt_text + output.text\n            else:\n                token_ids = output.token_ids\n                top_logprobs = output.logprobs\n                output_text = output.text\n\n            if request.logprobs is not None:\n                logprobs = create_logprobs_fn(\n                    token_ids=token_ids,\n                    top_logprobs=top_logprobs,\n                    num_output_top_logprobs=request.logprobs,\n                )\n            else:\n                logprobs = None\n\n            choice_data = CompletionResponseChoice(\n                index=len(choices),\n                text=output_text,\n                logprobs=logprobs,\n                finish_reason=output.finish_reason,\n            )\n            choices.append(choice_data)\n\n        num_prompt_tokens += len(prompt_token_ids)\n        num_generated_tokens += sum(\n            len(output.token_ids) for output in final_res.outputs)\n\n    usage = UsageInfo(\n        prompt_tokens=num_prompt_tokens,\n        completion_tokens=num_generated_tokens,\n        total_tokens=num_prompt_tokens + num_generated_tokens,\n    )\n\n    return CompletionResponse(\n        id=request_id,\n        created=created_time,\n        model=model_name,\n        choices=choices,\n        usage=usage,\n    )\n\n\ndef merge_async_iterators(*iterators):\n    \"\"\"Merge multiple asynchronous iterators into a single iterator.\n\n    This method handle the case where some iterators finish before others.\n    When it yields, it yields a tuple (i, item) where i is the index of the\n    iterator that yields the item.\n    \"\"\"\n    queue = asyncio.Queue()\n\n    finished = [False] * len(iterators)\n\n    async def producer(i, iterator):\n        async for item in iterator:\n            await queue.put((i, item))\n        finished[i] = True\n\n    _tasks = [\n        asyncio.create_task(producer(i, iterator))\n        for i, iterator in enumerate(iterators)\n    ]\n\n    async def consumer():\n        while not all(finished) or not queue.empty():\n            item = await queue.get()\n            yield item\n        await asyncio.gather(*_tasks)\n\n    return consumer()\n\n\nclass OpenAIServingCompletion(OpenAIServing):\n\n    def __init__(self, engine: AsyncLLMEngine, served_model: str):\n        super().__init__(engine=engine, served_model=served_model)\n\n    async def create_completion(self, request: CompletionRequest,\n                                raw_request: Request):\n        \"\"\"Completion API similar to OpenAI's API.\n\n        See https://platform.openai.com/docs/api-reference/completions/create\n        for the API specification. This API mimics the OpenAI Completion API.\n\n        NOTE: Currently we do not support the following features:\n            - suffix (the language models we currently support do not support\n            suffix)\n            - logit_bias (to be supported by vLLM engine)\n        \"\"\"\n        error_check_ret = await self._check_model(request)\n        if error_check_ret is not None:\n            return error_check_ret\n\n        # Return error for unsupported features.\n        if request.suffix is not None:\n            return self.create_error_response(\n                \"suffix is not currently supported\")\n        if request.logit_bias is not None and len(request.logit_bias) > 0:\n            return self.create_error_response(\n                \"logit_bias is not currently supported\")\n\n        model_name = request.model\n        request_id = f\"cmpl-{random_uuid()}\"\n        created_time = int(time.monotonic())\n\n        # Schedule the request and get the result generator.\n        generators = []\n        try:\n            sampling_params = request.to_sampling_params()\n            prompt_is_tokens, prompts = parse_prompt_format(request.prompt)\n\n            for i, prompt in enumerate(prompts):\n                if prompt_is_tokens:\n                    input_ids = self._validate_prompt_and_tokenize(\n                        request, prompt_ids=prompt)\n                else:\n                    input_ids = self._validate_prompt_and_tokenize(\n                        request, prompt=prompt)\n\n                generators.append(\n                    self.engine.generate(None,\n                                         sampling_params,\n                                         f\"{request_id}-{i}\",\n                                         prompt_token_ids=input_ids))\n        except ValueError as e:\n            return self.create_error_response(str(e))\n\n        result_generator: AsyncIterator[Tuple[\n            int, RequestOutput]] = merge_async_iterators(*generators)\n\n        # Similar to the OpenAI API, when n != best_of, we do not stream the\n        # results. In addition, we do not stream the results when use beam search.\n        stream = (request.stream\n                  and (request.best_of is None or request.n == request.best_of)\n                  and not request.use_beam_search)\n\n        # Streaming response\n        if stream:\n            return completion_stream_generator(request,\n                                               raw_request,\n                                               self.engine.abort,\n                                               result_generator,\n                                               self._create_logprobs,\n                                               request_id,\n                                               created_time,\n                                               model_name,\n                                               num_prompts=len(prompts))\n\n        # Non-streaming response\n        final_res_batch: RequestOutput = [None] * len(prompts)\n        async for i, res in result_generator:\n            if await raw_request.is_disconnected():\n                # Abort the request if the client disconnects.\n                await self.engine.abort(f\"{request_id}-{i}\")\n                return self.create_error_response(\"Client disconnected\")\n            final_res_batch[i] = res\n        response = request_output_to_completion_response(\n            final_res_batch, request, self._create_logprobs, request_id,\n            created_time, model_name)\n\n        # When user requests streaming but we don't stream, we still need to\n        # return a streaming response with a single event.\n        if request.stream:\n            response_json = response.model_dump_json()\n\n            async def fake_stream_generator() -> AsyncGenerator[str, None]:\n                yield f\"data: {response_json}\\n\\n\"\n                yield \"data: [DONE]\\n\\n\"\n\n            return fake_stream_generator()\n\n        return response\n", "path": "vllm/entrypoints/openai/serving_completion.py"}]}