Repository: lm-sys/FastChat (language: python)

Issue:
Add a simple chatbot
Download the model ckpt from GCS bucket and specify the correct `CKPT_PATH`.
Run `python chatbot.py`.
<img width="626" alt="image" src="https://user-images.githubusercontent.com/1206058/226166520-f397e6bc-42ed-4754-9ca1-15bf47c354eb.png">


Discussion / hints:

Fix patch:
diff --git a/.gitignore b/.gitignore
index 14fad0560..06818db1d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,6 +7,7 @@ dist
 
 # Log
 *.log
+*.log.*
 *.json
 
 # Editor
diff --git a/docs/openai_api.md b/docs/openai_api.md
index 5d771b10f..7b9643856 100644
--- a/docs/openai_api.md
+++ b/docs/openai_api.md
@@ -25,14 +25,45 @@ Finally, launch the RESTful API server
 python3 -m fastchat.serve.openai_api_server --host localhost --port 8000
 ```
 
-Test the API server
+Now, let us test the API server...
 
-### List Models
+### OpenAI Official SDK
+The final goal of `openai_api_server.py` is to implement a fully OpenAI-Compatible API server, so the models can be used directly with [openai-python](https://github.com/openai/openai-python) library.
+
+First, install openai-python:
+```bash
+pip install --upgrade openai
+```
+
+Then, interact with model vicuna:
+```python
+import openai
+openai.api_key = "EMPTY" # Not support yet
+openai.api_base = "http://localhost:8000/v1"
+
+# create a completion
+completion = openai.Completion.create(model="vicuna-7b-v1.1", prompt="Hello world", max_tokens=64)
+# print the completion
+print(completion.choices[0].text)
+
+# create a chat completion
+completion = openai.ChatCompletion.create(
+  model="vicuna-7b-v1.1",
+  messages=[{"role": "user", "content": "Hello world!"}]
+)
+# print the completion
+print(completion.choices[0].message.content)
+```
+
+### cURL
+cURL is another good tool for observing the output of the api.
+
+List Models:
 ```bash
 curl http://localhost:8000/v1/models
 ```
 
-### Chat Completions
+Chat Completions:
 ```bash
 curl http://localhost:8000/v1/chat/completions \
   -H "Content-Type: application/json" \
@@ -42,7 +73,7 @@ curl http://localhost:8000/v1/chat/completions \
   }'
 ```
 
-### Text Completions
+Text Completions:
 ```bash
 curl http://localhost:8000/v1/completions \
   -H "Content-Type: application/json" \
@@ -54,9 +85,9 @@ curl http://localhost:8000/v1/completions \
   }'
 ```
 
-### Embeddings
+Embeddings:
 ```bash
-curl http://localhost:8000/v1/create_embeddings \
+curl http://localhost:8000/v1/embeddings \
   -H "Content-Type: application/json" \
   -d '{
     "model": "vicuna-7b-v1.1",
@@ -64,7 +95,8 @@ curl http://localhost:8000/v1/create_embeddings \
   }'
 ```
 
-## Client SDK
+### FastChat Client SDK
+FastChat also includes its own client SDK for the API.
 
 Assuming environment variable `FASTCHAT_BASEURL` is set to the API server URL (e.g., `http://localhost:8000`), you can use the following code to send a request to the API server:
 
@@ -105,7 +137,12 @@ The script will train classifiers based on `vicuna-7b`, `text-similarity-ada-001
 ## Todos
 Some features to be implemented:
 
-- [ ] Support more parameters like `top_p`, `presence_penalty`
-- [ ] Report token usage for chat completion
-- [ ] Proper error handling (e.g., model not found)
+- [ ] Support more parameters like `logprobs`, `logit_bias`, `user`, `presence_penalty` and `frequency_penalty`
 - [ ] The return value in the client SDK could be used like a dict
+- [ ] Model details (permissions, owner and create time)
+- [ ] Edits API
+- [ ] Authentication and API key
+- [ ] Rate Limitation Settings
+- [x] Parameter `top_p` support
+- [x] Report token usage for chat completion
+- [x] Proper error handling (e.g., model not found)
\ No newline at end of file
diff --git a/fastchat/client/openai_api_client.py b/fastchat/client/openai_api_client.py
index bdcdf859a..e68a8d4a9 100644
--- a/fastchat/client/openai_api_client.py
+++ b/fastchat/client/openai_api_client.py
@@ -48,10 +48,13 @@ async def request_completion_stream(
     ) -> AsyncGenerator:
         """
         Create chat completion as a stream
+        Parse the Event stream format: 
+        https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event_stream_format
         :param request: The request data
         :param timeout: The timeout of the request
         :returns: Compleation stream
         """
+        VALID_EVENT_STREAM_FIELD = ["id", "data", "event", "retry"]
         async with httpx.AsyncClient() as client:
             async with client.stream(
                 "POST",
@@ -62,10 +65,25 @@ async def request_completion_stream(
                 async for chunk in response.aiter_text():
                     if not chunk:
                         continue
-                    for line in chunk.split("\n"):
-                        if not line:
+                    for message in chunk.split("\n\n"):
+                        if not message:
                             continue
-                        yield ChatCompletionStreamResponse.parse_obj(json.loads(line))
+                        lines = message.split("\n")
+
+                        message_dict = {}
+                        for line in lines:
+                            colon_index = line.find(":")
+                            if colon_index == 0 or colon_index == -1:
+                                continue
+                            message_key = line[:colon_index].strip()
+                            message_value = line[colon_index + 1:]
+                            if message_key in VALID_EVENT_STREAM_FIELD:
+                                message_dict[message_key] = message_value
+                        
+                        data_field = message_dict["data"]
+                        if data_field.strip() == "[DONE]":
+                            break
+                        yield ChatCompletionStreamResponse.parse_obj(json.loads(data_field))
 
 
 class ChatCompletion:
diff --git a/fastchat/constants.py b/fastchat/constants.py
index a0cc7917c..6435c0cfe 100644
--- a/fastchat/constants.py
+++ b/fastchat/constants.py
@@ -1,5 +1,29 @@
+from enum import IntEnum
+
+
 CONTROLLER_HEART_BEAT_EXPIRATION = 90
 WORKER_HEART_BEAT_INTERVAL = 30
 WORKER_API_TIMEOUT = 20
 
 LOGDIR = "."
+
+class ErrorCode(IntEnum):
+    '''
+    https://platform.openai.com/docs/guides/error-codes/api-errors
+    '''
+    VALIDATION_TYPE_ERROR = 40001
+
+    INVALID_AUTH_KEY = 40101
+    INCORRECT_AUTH_KEY = 40102
+    NO_PERMISSION = 40103
+
+    INVALID_MODEL = 40301
+    PARAM_OUT_OF_RANGE = 40302
+    CONTEXT_OVERFLOW = 40303
+
+    RATE_LIMIT = 42901
+    QUOTA_EXCEEDED = 42902
+    ENGINE_OVERLOADED = 42903
+
+    INTERNAL_ERROR = 50001
+    CUDA_OUT_OF_MEMORY = 50002
diff --git a/fastchat/model/chatglm_model.py b/fastchat/model/chatglm_model.py
index 923065966..2a79b53fb 100644
--- a/fastchat/model/chatglm_model.py
+++ b/fastchat/model/chatglm_model.py
@@ -1,6 +1,19 @@
 import torch
 from typing import List, Tuple
 
+@torch.no_grad()
+def stream_chat_token_num(tokenizer, query: str, history: List[Tuple[str, str]] = None):
+    if history is None:
+        history = []
+    if not history:
+        prompt = query
+    else:
+        prompt = ""
+        for i, (old_query, response) in enumerate(history):
+            prompt += "[Round {}]\n问：{}\n答：{}\n".format(i, old_query, response)
+        prompt += "[Round {}]\n问：{}\n答：".format(len(history), query)
+    inputs = tokenizer([prompt], return_tensors="pt")
+    return torch.numel(inputs['input_ids'])
 
 @torch.inference_mode()
 def chatglm_generate_stream(
@@ -29,10 +42,33 @@ def chatglm_generate_stream(
         hist.append((messages[i][1], messages[i + 1][1]))
     query = messages[-2][1]
 
-    for response, new_hist in model.stream_chat(tokenizer, query, hist, **gen_kwargs):
+    input_echo_len = stream_chat_token_num(tokenizer, query, hist)
+
+    for i, (response, new_hist) in enumerate(model.stream_chat(tokenizer, query, hist, **gen_kwargs)):
         if echo:
             output = query + " " + response
         else:
             output = response
 
-        yield output
+        yield {
+            "text": output,
+            "usage": {
+                "prompt_tokens": input_echo_len,
+                "completion_tokens": i,
+                "total_tokens": input_echo_len + i,
+            },
+            "finish_reason": None
+        }
+
+    # TODO: ChatGLM stop when it reach max length
+    # Only last stream result contains finish_reason, we set finish_reason as stop
+    ret = {
+        "text": output,
+        "usage": {
+            "prompt_tokens": input_echo_len,
+            "completion_tokens": i,
+            "total_tokens": input_echo_len + i,
+        },
+        "finish_reason": "stop"
+    }
+    yield ret
diff --git a/fastchat/protocol/openai_api_protocol.py b/fastchat/protocol/openai_api_protocol.py
index 63471737b..646f0cb2e 100644
--- a/fastchat/protocol/openai_api_protocol.py
+++ b/fastchat/protocol/openai_api_protocol.py
@@ -1,20 +1,61 @@
-from typing import Optional, List, Dict, Any
+from typing import Literal, Optional, List, Dict, Any, Union
 
 import time
 
 import shortuuid
 from pydantic import BaseModel, Field
 
+class ErrorResponse(BaseModel):
+    object: str = "error"
+    message: str
+    code: int
+
+
+class ModelPermission(BaseModel):
+    id: str = Field(default_factory=lambda: f"modelperm-{shortuuid.random()}")
+    object: str = "model_permission"
+    created: int = Field(default_factory=lambda: int(time.time()))
+    allow_create_engine: bool = False
+    allow_sampling: bool = True
+    allow_logprobs: bool = True
+    allow_search_indices: bool = True
+    allow_view: bool = True
+    allow_fine_tuning: bool = False
+    organization: str = "*"
+    group: Optional[str] = None
+    is_blocking: str = False
+
+
+class ModelCard(BaseModel):
+    id: str
+    object: str = "model"
+    created: int = Field(default_factory=lambda: int(time.time()))
+    owned_by: str = "fastchat"
+    root: Optional[str] = None
+    parent: Optional[str] = None
+    permission: List[ModelPermission] = []
+
+class ModelList(BaseModel):
+    object: str = "list"
+    data: List[ModelCard] = []
+
+class UsageInfo(BaseModel):
+    prompt_tokens: int = 0
+    total_tokens: int = 0
+    completion_tokens: Optional[int] = 0
 
 class ChatCompletionRequest(BaseModel):
-    # TODO: support stop with a list of text etc.
     model: str
     messages: List[Dict[str, str]]
     temperature: Optional[float] = 0.7
-    n: int = 1
+    top_p: Optional[float] = 1.0
+    n: Optional[int] = 1
     max_tokens: Optional[int] = None
-    stop: Optional[str] = None
-    stream: Optional[bool] = None
+    stop: Optional[Union[str, List[str]]] = None
+    stream: Optional[bool] = False
+    presence_penalty: Optional[float] = 0.0
+    frequency_penalty: Optional[float] = 0.0
+    user: Optional[str] = None
 
 
 class ChatMessage(BaseModel):
@@ -25,30 +66,30 @@ class ChatMessage(BaseModel):
 class ChatCompletionResponseChoice(BaseModel):
     index: int
     message: ChatMessage
-    finish_reason: str
-
+    finish_reason: Optional[Literal["stop", "length"]]
 
 class ChatCompletionResponse(BaseModel):
-    id: str = Field(default_factory=shortuuid.random)
+    id: str = Field(default_factory=lambda: f"chatcmpl-{shortuuid.random()}")
     object: str = "chat.completion"
     created: int = Field(default_factory=lambda: int(time.time()))
     model: str
     choices: List[ChatCompletionResponseChoice]
-    usage: Dict[str, int]
+    usage: UsageInfo
 
 
 class DeltaMessage(BaseModel):
-    content: str
+    role: Optional[str] = None
+    content: Optional[str] = None
 
 
 class ChatCompletionResponseStreamChoice(BaseModel):
     index: int
     delta: DeltaMessage
-    finish_reason: Optional[str]
+    finish_reason: Optional[Literal["stop", "length"]]
 
 
 class ChatCompletionStreamResponse(BaseModel):
-    id: str = Field(default_factory=shortuuid.random)
+    id: str = Field(default_factory=lambda: f"chatcmpl-{shortuuid.random()}")
     object: str = "chat.completion.chunk"
     created: int = Field(default_factory=lambda: int(time.time()))
     model: str
@@ -58,13 +99,14 @@ class ChatCompletionStreamResponse(BaseModel):
 class EmbeddingsRequest(BaseModel):
     model: str
     input: str
+    user: Optional[str] = None
 
 
 class EmbeddingsResponse(BaseModel):
     object: str = "list"
     data: List[Dict[str, Any]]
     model: str
-    usage: Dict[str, int]
+    usage: UsageInfo
 
 
 class CompletionRequest(BaseModel):
@@ -72,18 +114,44 @@ class CompletionRequest(BaseModel):
     prompt: str
     suffix: Optional[str] = None
     temperature: Optional[float] = 0.7
-    n: int = 1
-    max_tokens: int
-    stop: Optional[str] = None
-    stream: bool = False
+    n: Optional[int] = 1
+    max_tokens: Optional[int] = None
+    stop: Optional[Union[str, List[str]]] = None
+    stream: Optional[bool] = False
+    top_p: Optional[float] = 1.0
+    logprobs: Optional[int] = None
+    echo: Optional[bool] = False
+    presence_penalty: Optional[float] = 0.0
+    frequency_penalty: Optional[float] = 0.0
+    user: Optional[str] = None
+
+
+class CompletionResponseChoice(BaseModel):
+    index: int
+    text: str
     logprobs: Optional[int] = None
-    echo: bool = False
+    finish_reason: Optional[Literal["stop", "length"]]
 
 
 class CompletionResponse(BaseModel):
-    id: str = Field(default_factory=shortuuid.random)
+    id: str = Field(default_factory=lambda: f"cmpl-{shortuuid.random()}")
+    object: str = "text_completion"
+    created: int = Field(default_factory=lambda: int(time.time()))
+    model: str
+    choices: List[CompletionResponseChoice]
+    usage: UsageInfo
+
+
+class CompletionResponseStreamChoice(BaseModel):
+    index: int
+    text: str
+    logprobs: Optional[float] = None
+    finish_reason: Optional[Literal["stop", "length"]] = None
+
+
+class CompletionStreamResponse(BaseModel):
+    id: str = Field(default_factory=lambda: f"cmpl-{shortuuid.random()}")
     object: str = "text_completion"
     created: int = Field(default_factory=lambda: int(time.time()))
     model: str
-    choices: List[Dict[str, Any]]
-    usage: Dict[str, int]
+    choices: List[CompletionResponseStreamChoice]
diff --git a/fastchat/serve/inference.py b/fastchat/serve/inference.py
index 1924ded12..cec598224 100644
--- a/fastchat/serve/inference.py
+++ b/fastchat/serve/inference.py
@@ -2,7 +2,7 @@
 import abc
 import gc
 import math
-from typing import Optional
+from typing import Iterable, Optional
 import sys
 import warnings
 
@@ -70,6 +70,13 @@ def generate_stream(
     input_ids = tokenizer(prompt).input_ids
     input_echo_len = len(input_ids)
     output_ids = list(input_ids)
+    
+    if len(input_ids) + max_new_tokens > context_len:
+        raise ValueError(f"This model's maximum context length is {context_len} tokens. "
+                         f"However, you requested {max_new_tokens + len(input_ids)} tokens "
+                         f"({len(input_ids)} in the messages, "
+                         f"{max_new_tokens} in the completion). "
+                         f"Please reduce the length of the messages or completion.")
 
     if model.config.is_encoder_decoder:
         max_src_len = context_len
@@ -160,15 +167,53 @@ def generate_stream(
                 spaces_between_special_tokens=False,
             )
             if stop_str:
-                pos = output.rfind(stop_str, rfind_start)
-                if pos != -1:
-                    output = output[:pos]
-                    stopped = True
-            yield output
+                if isinstance(stop_str, str):
+                    pos = output.rfind(stop_str, rfind_start)
+                    if pos != -1:
+                        output = output[:pos]
+                        stopped = True
+                elif isinstance(stop_str, Iterable):
+                    for each_stop in stop_str:
+                        pos = output.rfind(each_stop, rfind_start)
+                        if pos != -1:
+                            output = output[:pos]
+                            stopped = True
+                            break
+                else:
+                    raise ValueError("Invalid stop field type.")
+
+            yield {
+                "text": output,
+                "usage": {
+                    "prompt_tokens": input_echo_len,
+                    "completion_tokens": i,
+                    "total_tokens": input_echo_len + i,
+                },
+                "finish_reason": None
+            }
 
         if stopped:
             break
-
+    
+    # finish stream event, which contains finish reason
+    if i == max_new_tokens - 1:
+        finish_reason = "length"
+    elif stopped:
+        finish_reason = "stop"
+    else:
+        finish_reason = None
+    
+    yield {
+        "text": output,
+        "usage": {
+            "prompt_tokens": input_echo_len,
+            "completion_tokens": i,
+            "total_tokens": input_echo_len + i,
+        },
+        "finish_reason": finish_reason
+    }
+
+    # clean
     del past_key_values, out
     gc.collect()
     torch.cuda.empty_cache()
diff --git a/fastchat/serve/model_worker.py b/fastchat/serve/model_worker.py
index a6dca495a..9c3a27bce 100644
--- a/fastchat/serve/model_worker.py
+++ b/fastchat/serve/model_worker.py
@@ -33,7 +33,7 @@
 import torch
 import uvicorn
 
-from fastchat.constants import WORKER_HEART_BEAT_INTERVAL
+from fastchat.constants import WORKER_HEART_BEAT_INTERVAL, ErrorCode
 from fastchat.model.model_adapter import load_model, add_model_args
 from fastchat.model.chatglm_model import chatglm_generate_stream
 from fastchat.serve.inference import generate_stream
@@ -89,12 +89,13 @@ def __init__(
         else:
             self.context_len = 2048
 
+        # generate_stream
         is_chatglm = "chatglm" in str(type(self.model)).lower()
         if is_chatglm:
             self.generate_stream_func = chatglm_generate_stream
         else:
             self.generate_stream_func = generate_stream
-
+        
         if not no_register:
             self.register_to_controller()
             self.heart_beat_thread = threading.Thread(
@@ -174,57 +175,61 @@ def generate_stream_gate(self, params):
                 args.stream_interval,
             ):
                 ret = {
-                    "text": output,
+                    "text": output["text"],
                     "error_code": 0,
                 }
+                if "usage" in output:
+                    ret["usage"] = output["usage"]
+                if "finish_reason" in output:
+                    ret["finish_reason"] = output["finish_reason"]
+                if "logprobs" in output:
+                    ret["logprobs"] = output["logprobs"]
                 yield json.dumps(ret).encode() + b"\0"
         except torch.cuda.OutOfMemoryError:
             ret = {
                 "text": server_error_msg,
-                "error_code": 1,
+                "error_code": ErrorCode.CUDA_OUT_OF_MEMORY,
             }
-            yield json.dumps(ret).encode() + b"\0"
-
-    def generate_completion(self, params):
+        except ValueError as e:
+            ret = {
+                "text": str(e),
+                "error_code": ErrorCode.INTERNAL_ERROR,
+            }
+        yield json.dumps(ret).encode() + b"\0"
+    
+    def generate_gate(self, params):
         try:
-            input_ids = self.tokenizer([params["prompt"]]).input_ids
-            output_ids = self.model.generate(
-                torch.as_tensor(input_ids).cuda(),
-                do_sample=True,
-                temperature=params["temperature"],
-                max_new_tokens=params["max_tokens"]
-                - 1,  # generate max_new_tokens + 1 tokens
-            )
-            if self.model.config.is_encoder_decoder:
-                output_ids = output_ids[0]
-            else:
-                output_ids = output_ids[0][len(input_ids[0]) :]
-            outputs = self.tokenizer.decode(
-                output_ids,
-                skip_special_tokens=True,
-                spaces_between_special_tokens=False,
-            )
-            completion_tokens = len(self.tokenizer(outputs).input_ids)
-            if completion_tokens >= params["max_tokens"]:
-                finish_reason = "length"
-            else:
-                finish_reason = "stop"
-            return json.dumps(
-                {
-                    "text": outputs,
-                    "finish_reason": finish_reason,
-                    "completion_tokens": completion_tokens,
-                    "prompt_tokens": len(input_ids[0]),
-                }
-            )
-
+            ret = {
+                "text": "",
+                "error_code": 0
+            }
+            for output in self.generate_stream_func(
+                self.model,
+                self.tokenizer,
+                params,
+                self.device,
+                self.context_len,
+                args.stream_interval,
+            ):
+                ret["text"] = output["text"]
+            if "usage" in output:
+                ret["usage"] = output["usage"]
+            if "finish_reason" in output:
+                ret["finish_reason"] = output["finish_reason"]
+            if "logprobs" in output:
+                ret["logprobs"] = output["logprobs"]
         except torch.cuda.OutOfMemoryError:
             ret = {
                 "text": server_error_msg,
-                "error_code": 1,
+                "error_code": ErrorCode.CUDA_OUT_OF_MEMORY,
             }
-            return json.dumps(ret).encode() + b"\0"
-
+        except ValueError as e:
+            ret = {
+                "text": str(e),
+                "error_code": ErrorCode.INTERNAL_ERROR,
+            }
+        return ret
+    
     def get_embeddings(self, params):
         try:
             tokenizer = self.tokenizer
@@ -247,7 +252,7 @@ def get_embeddings(self, params):
         except torch.cuda.OutOfMemoryError:
             ret = {
                 "text": server_error_msg,
-                "error_code": 1,
+                "error_code": ErrorCode.CUDA_OUT_OF_MEMORY,
             }
             return json.dumps(ret).encode() + b"\0"
 
@@ -281,12 +286,28 @@ async def api_generate_stream(request: Request):
     background_tasks = create_background_tasks()
     return StreamingResponse(generator, background=background_tasks)
 
+@app.post("/worker_generate")
+async def api_generate(request: Request):
+    params = await request.json()
+    await acquire_model_semaphore()
+    output = worker.generate_gate(params)
+    release_model_semaphore()
+    return JSONResponse(output)
+
+@app.post("/worker_generate_completion_stream")
+async def api_generate_completion_stream(request: Request):
+    params = await request.json()
+    await acquire_model_semaphore()
+    generator = worker.generate_stream_gate(params)
+    background_tasks = create_background_tasks()
+    return StreamingResponse(generator, background=background_tasks)
+
 
 @app.post("/worker_generate_completion")
 async def api_generate_completion(request: Request):
     params = await request.json()
     await acquire_model_semaphore()
-    completion = worker.generate_completion(params)
+    completion = worker.generate_gate(params)
     background_tasks = create_background_tasks()
     return JSONResponse(content=completion, background=background_tasks)
 
diff --git a/fastchat/serve/openai_api_server.py b/fastchat/serve/openai_api_server.py
index 53ffab142..2c0e1ede8 100644
--- a/fastchat/serve/openai_api_server.py
+++ b/fastchat/serve/openai_api_server.py
@@ -8,34 +8,46 @@
 python3 -m fastchat.serve.openai_api_server
 """
 import asyncio
+
 import argparse
 import asyncio
 import json
 import logging
+
 import os
-from typing import Union, Dict, List, Any
+from typing import Generator, Optional, Union, Dict, List, Any
 
 import fastapi
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import StreamingResponse
+from fastapi.responses import StreamingResponse, JSONResponse
 import httpx
+import shortuuid
 import uvicorn
 from pydantic import BaseSettings
 
-from fastchat.constants import WORKER_API_TIMEOUT
+from fastchat.constants import WORKER_API_TIMEOUT, ErrorCode
 from fastchat.model.model_adapter import get_conversation_template
+from fastapi.exceptions import RequestValidationError
 from fastchat.protocol.openai_api_protocol import (
     ChatCompletionRequest,
     ChatCompletionResponse,
+    ChatCompletionResponseStreamChoice,
     ChatCompletionStreamResponse,
     ChatMessage,
     ChatCompletionResponseChoice,
-    ChatCompletionResponseStreamChoice,
-    DeltaMessage,
     CompletionRequest,
     CompletionResponse,
+    CompletionResponseChoice,
+    DeltaMessage,
+    CompletionResponseStreamChoice,
+    CompletionStreamResponse,
     EmbeddingsRequest,
     EmbeddingsResponse,
+    ErrorResponse,
+    ModelCard,
+    ModelList,
+    ModelPermission,
+    UsageInfo,
 )
 
 logger = logging.getLogger(__name__)
@@ -48,95 +60,110 @@ class AppSettings(BaseSettings):
 
 app_settings = AppSettings()
 
-app = fastapi.FastAPI()
+app = fastapi.FastAPI(debug=True)
 headers = {"User-Agent": "FastChat API Server"}
 
 
-@app.get("/v1/models")
-async def show_available_models():
-    controller_address = app_settings.controller_address
-    async with httpx.AsyncClient() as client:
-        ret = await client.post(controller_address + "/refresh_all_workers")
-        ret = await client.post(controller_address + "/list_models")
-    models = ret.json()["models"]
-    models.sort()
-    return {"data": [{"id": m, "object": "model"} for m in models],
-            "object": "list"}
-
+def create_error_response(code: int, message: str) -> JSONResponse:
+    return JSONResponse(ErrorResponse(message=message, code=code).dict(), status_code=500) 
 
-@app.post("/v1/chat/completions")
-async def create_chat_completion(request: ChatCompletionRequest):
-    """Creates a completion for the chat message"""
-    gen_params = get_gen_params(
-        request.model,
-        request.messages,
-        temperature=request.temperature,
-        max_tokens=request.max_tokens,
-        echo=False,
-        stop=request.stop,
-    )
-    if request.stream:
-        response_stream = chat_completion_stream(
-            request.model, gen_params, request.n)
-        return StreamingResponse(response_stream, media_type="text/event-stream")
 
-    # TODO: batch the requests
-    choices = []
-    chat_completions = []
-    for i in range(request.n):
-        content = asyncio.create_task(chat_completion(request.model, gen_params))
-        chat_completions.append(content)
+@app.exception_handler(RequestValidationError)
+async def validation_exception_handler(request, exc):
+    return create_error_response(ErrorCode.VALIDATION_TYPE_ERROR, str(exc))
 
-    for i, content_task in enumerate(chat_completions):
-        content = await content_task
-        choices.append(
-            ChatCompletionResponseChoice(
-                index=i,
-                message=ChatMessage(role="assistant", content=content),
-                # TODO: support other finish_reason
-                finish_reason="stop",
+async def check_model(request) -> Optional[JSONResponse]:
+    controller_address = app_settings.controller_address
+    ret = None
+    async with httpx.AsyncClient() as client:
+        try:
+            _worker_addr = await _get_worker_address(request.model, client)
+        except:
+            models_ret = await client.post(controller_address + "/list_models")
+            models = models_ret.json()["models"]
+            ret = create_error_response(
+                ErrorCode.INVALID_MODEL,
+                f"Only {'&&'.join(models)} allowed now, your model {request.model}"
             )
+    return ret
+
+def check_requests(request) -> Optional[JSONResponse]:
+    # Check all params
+    if request.max_tokens is not None and request.max_tokens <= 0:
+        return create_error_response(
+            ErrorCode.PARAM_OUT_OF_RANGE,
+            f"{request.max_tokens} is less than the minimum of 1 - 'max_tokens'"
         )
-
-    # TODO: support usage field
-    usage = {
-        "prompt_tokens": -1,
-        "completion_tokens": -1,
-        "total_tokens": -1,
-    }
-    return ChatCompletionResponse(model=request.model, choices=choices, usage=usage)
+    if request.n is not None and request.n <= 0:
+        return create_error_response(
+            ErrorCode.PARAM_OUT_OF_RANGE,
+            f"{request.n} is less than the minimum of 1 - 'n'"
+        )
+    if request.temperature is not None and request.temperature < 0:
+        return create_error_response(
+            ErrorCode.PARAM_OUT_OF_RANGE,
+            f"{request.temperature} is less than the minimum of 0 - 'temperature'"
+        )
+    if request.temperature is not None and request.temperature > 2:
+        return create_error_response(
+            ErrorCode.PARAM_OUT_OF_RANGE,
+            f"{request.temperature} is greater than the maximum of 2 - 'temperature'"
+        )
+    if request.top_p is not None and request.top_p < 0:
+        return create_error_response(
+            ErrorCode.PARAM_OUT_OF_RANGE,
+            f"{request.top_p} is less than the minimum of 0 - 'top_p'"
+        )
+    if request.top_p is not None and request.top_p > 1:
+        return create_error_response(
+            ErrorCode.PARAM_OUT_OF_RANGE,
+            f"{request.top_p} is greater than the maximum of 1 - 'temperature'"
+        )
+    if request.stop is not None and \
+        (not isinstance(request.stop, str) and not isinstance(request.stop, list)):
+        return create_error_response(
+            ErrorCode.PARAM_OUT_OF_RANGE,
+            f"{request.stop} is not valid under any of the given schemas - 'stop'"
+        )
+        
+    return None
 
 
 def get_gen_params(
     model_name: str,
-    messages: List[Dict[str, str]],
+    messages: Union[str, List[Dict[str, str]]],
     *,
     temperature: float,
-    max_tokens: int,
-    echo: bool,
-    stop: Union[str, None],
-):
+    top_p: float,
+    max_tokens: Optional[int],
+    echo: Optional[bool],
+    stream: Optional[bool],
+    stop: Optional[Union[str, List[str]]],
+) -> Dict[str, Any]:
     is_chatglm = "chatglm" in model_name.lower()
     conv = get_conversation_template(model_name)
 
-    for message in messages:
-        msg_role = message["role"]
-        if msg_role == "system":
-            conv.system = message["content"]
-        elif msg_role == "user":
-            conv.append_message(conv.roles[0], message["content"])
-        elif msg_role == "assistant":
-            conv.append_message(conv.roles[1], message["content"])
-        else:
-            raise ValueError(f"Unknown role: {msg_role}")
-
-    # Add a blank message for the assistant.
-    conv.append_message(conv.roles[1], None)
-
-    if is_chatglm:
-        prompt = conv.messages[conv.offset :]
+    if isinstance(messages, str):
+        prompt = messages
     else:
-        prompt = conv.get_prompt()
+        for message in messages:
+            msg_role = message["role"]
+            if msg_role == "system":
+                conv.system = message["content"]
+            elif msg_role == "user":
+                conv.append_message(conv.roles[0], message["content"])
+            elif msg_role == "assistant":
+                conv.append_message(conv.roles[1], message["content"])
+            else:
+                raise ValueError(f"Unknown role: {msg_role}")
+
+        # Add a blank message for the assistant.
+        conv.append_message(conv.roles[1], None)
+
+        if is_chatglm:
+            prompt = conv.messages[conv.offset :]
+        else:
+            prompt = conv.get_prompt()
 
     if max_tokens is None:
         max_tokens = 512
@@ -145,11 +172,22 @@ def get_gen_params(
         "model": model_name,
         "prompt": prompt,
         "temperature": temperature,
+        "top_p": top_p,
         "max_new_tokens": max_tokens,
         "echo": echo,
-        "stop": conv.stop_str,
-        "stop_token_ids": conv.stop_token_ids,
+        "stream": stream
     }
+
+    if stop is None:
+        gen_params.update({
+            "stop": conv.stop_str,
+            "stop_token_ids": conv.stop_token_ids
+        })
+    else:
+        gen_params.update({
+            "stop": stop
+        })
+
     logger.debug(f"==== request ====\n{gen_params}")
     return gen_params
 
@@ -177,59 +215,171 @@ async def _get_worker_address(model_name: str, client: httpx.AsyncClient) -> str
     return worker_addr
 
 
-async def chat_completion_stream(model_name: str, gen_params: Dict[str, Any], n: int):
+@app.get("/v1/models")
+async def show_available_models():
+    controller_address = app_settings.controller_address
     async with httpx.AsyncClient() as client:
-        worker_addr = await _get_worker_address(model_name, client)
-        delimiter = b"\0"
- 
-        for idx in range(n):
-            delta_position = 0
-            response = ChatCompletionStreamResponse(
-                model=model_name,
-                choices=[
-                    ChatCompletionResponseStreamChoice(
-                        index=idx,
-                        delta=DeltaMessage(content=""),
-                        finish_reason=None,
-                    )
-                ]
+        ret = await client.post(controller_address + "/refresh_all_workers")
+        ret = await client.post(controller_address + "/list_models")
+    models = ret.json()["models"]
+    models.sort()
+    # TODO: return real model permission details
+    model_cards = []
+    for m in models:
+        model_cards.append(
+            ModelCard(
+                id=m,
+                root=m,
+                permission=[ModelPermission()]
             )
+        )
+    return ModelList(data=model_cards)
+
+
+@app.post("/v1/chat/completions")
+async def create_chat_completion(request: ChatCompletionRequest):
+    """Creates a completion for the chat message"""
+    error_check_ret = await check_model(request)
+    if error_check_ret is not None:
+        return error_check_ret
+    error_check_ret = check_requests(request)
+    if error_check_ret is not None:
+        return error_check_ret
 
-            async with client.stream(
-                "POST",
-                worker_addr + "/worker_generate_stream",
-                headers=headers,
-                json=gen_params,
-                timeout=WORKER_API_TIMEOUT,
-            ) as post_response:
-                # TODO: begins with a single delta containing the role and a null finish reason
-                async for raw_chunk in post_response.aiter_raw():
-                    for chunk in raw_chunk.split(delimiter):
-                        if not chunk:
-                            continue
-                        data = json.loads(chunk.decode())
-                        if data["error_code"] == 0:
-                            last_delta_position = delta_position
-                            output = data["text"].strip()
-                            delta_position = len(output)
-                            delta_diff = delta_position - last_delta_position
-                            if delta_diff > 0:
-                                delta = output[last_delta_position:]
-                                response.choices[0].delta.content = delta
-                                yield json.dumps(response.dict())
-
-                # Streaming the finishing token
-                response.choices[0].delta.content = ""
-                response.choices[0].finish_reason = "stop"
-                yield json.dumps(response.dict())
-
-
-async def chat_completion(model_name: str, gen_params: Dict[str, Any]):
+    gen_params = get_gen_params(
+        request.model,
+        request.messages,
+        temperature=request.temperature,
+        top_p=request.top_p,
+        max_tokens=request.max_tokens,
+        echo=False,
+        stream=request.stream,
+        stop=request.stop,
+    )
+    if request.stream:
+        generator = chat_completion_stream_generator(request.model, gen_params, request.n)
+        return StreamingResponse(generator, media_type="text/event-stream")
+
+    choices = []
+    # TODO: batch the requests. maybe not necessary if using CacheFlow worker
+    chat_completions = []
+    for i in range(request.n):
+        content = asyncio.create_task(chat_completion(request.model, gen_params))
+        chat_completions.append(content)
+    try:
+        all_tasks = await asyncio.gather(*chat_completions)
+    except Exception as e:
+        return create_error_response(
+            ErrorCode.INTERNAL_ERROR,
+            str(e)
+        )
+    usage = UsageInfo()
+    for i, content in enumerate(all_tasks):
+        if content["error_code"] != 0:
+            return create_error_response(content["error_code"], content["text"])
+        choices.append(
+            ChatCompletionResponseChoice(
+                index=i,
+                message=ChatMessage(role="assistant", content=content["text"]),
+                finish_reason=content.get("finish_reason", "stop"),
+            )
+        )
+        task_usage = UsageInfo.parse_obj(content["usage"])
+        for usage_key, usage_value in task_usage.dict().items():
+            setattr(usage, usage_key, getattr(usage, usage_key) + usage_value)
+
+    return ChatCompletionResponse(
+        model=request.model,
+        choices=choices,
+        usage=usage
+    )
+
+
+async def chat_completion_stream_generator(
+        model_name: str,
+        gen_params: Dict[str, Any],
+        n: int
+    ) -> Generator[str, Any, None]:
+    """
+    Event stream format:
+    https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event_stream_format
+    """
+    id = f"chatcmpl-{shortuuid.random()}"
+    finish_stream_events = []
+    for i in range(n):
+        # First chunk with role
+        choice_data = ChatCompletionResponseStreamChoice(
+            index=i,
+            delta=DeltaMessage(role="assistant"),
+            finish_reason=None,
+        )
+        chunk = ChatCompletionStreamResponse(
+            id=id,
+            choices=[choice_data],
+            model=model_name
+        )
+        yield f"data: {chunk.json(exclude_unset=True, ensure_ascii=False)}\n\n"
+
+        previous_text = ""
+        async for content in chat_completion_stream(model_name, gen_params):
+            if content["error_code"] != 0:
+                yield f"data: {json.dumps(content, ensure_ascii=False)}\n\n"
+                yield "data: [DONE]\n\n"
+                return
+            decoded_unicode = content["text"].replace('\ufffd', '')
+            delta_text = decoded_unicode[len(previous_text):]
+            previous_text = decoded_unicode
+
+            if len(delta_text) == 0:
+                delta_text = None
+            choice_data = ChatCompletionResponseStreamChoice(
+                index=i,
+                delta=DeltaMessage(content=delta_text),
+                finish_reason=content.get("finish_reason", None),
+            )
+            chunk = ChatCompletionStreamResponse(
+                id=id,
+                choices=[choice_data],
+                model=model_name
+            )
+            if delta_text is None:
+                if content.get("finish_reason", None) is not None:
+                    finish_stream_events.append(chunk)
+                continue
+            yield f"data: {chunk.json(exclude_unset=True, ensure_ascii=False)}\n\n"
+    # There is not "content" field in the last delta message, so exclude_none to exclude field "content".
+    for finish_chunk in finish_stream_events:
+        yield f"data: {finish_chunk.json(exclude_none=True, ensure_ascii=False)}\n\n"
+    yield "data: [DONE]\n\n"
+
+
+async def chat_completion_stream(model_name: str, gen_params: Dict[str, Any]):
+    controller_url = app_settings.controller_address
+    async with httpx.AsyncClient() as client:
+        worker_addr = await _get_worker_address(model_name, client)
+        delimiter = b"\0"
+        async with client.stream(
+            "POST",
+            worker_addr + "/worker_generate_stream",
+            headers=headers,
+            json=gen_params,
+            timeout=WORKER_API_TIMEOUT,
+        ) as response:
+            # content = await response.aread()
+            async for raw_chunk in response.aiter_raw():
+                for chunk in raw_chunk.split(delimiter):
+                    if not chunk:
+                        continue
+                    data = json.loads(chunk.decode())
+                    yield data
+
+async def chat_completion(model_name: str, gen_params: Dict[str, Any]) -> Optional[Dict[str, Any]]:
     async with httpx.AsyncClient() as client:
         worker_addr = await _get_worker_address(model_name, client)
 
-        output = ""
+        output = None
         delimiter = b"\0"
+
         async with client.stream(
             "POST",
             worker_addr + "/worker_generate_stream",
@@ -243,48 +393,128 @@ async def chat_completion(model_name: str, gen_params: Dict[str, Any]):
             if not chunk:
                 continue
             data = json.loads(chunk.decode())
-            if data["error_code"] == 0:
-                output = data["text"].strip()
+            output = data
 
         return output
 
-
 @app.post("/v1/completions")
 async def create_completion(request: CompletionRequest):
-    payload = {
-        "model": request.model,
-        "prompt": request.prompt,
-        "temperature": request.temperature,
-        "max_tokens": request.max_tokens,
-        "logprobs": request.logprobs,
-    }
+    error_check_ret = await check_model(request)
+    if error_check_ret is not None:
+        return error_check_ret
+    error_check_ret = check_requests(request)
+    if error_check_ret is not None:
+        return error_check_ret
+
+    payload = get_gen_params(
+        request.model,
+        request.prompt,
+        temperature=request.temperature,
+        top_p=request.top_p,
+        max_tokens=request.max_tokens,
+        echo=request.echo,
+        stream=request.stream,
+        stop=request.stop,
+    )
 
     if request.stream:
-        raise NotImplementedError("streaming is not supported yet")
+        generator = generate_completion_stream_generator(payload, request.n)
+        return StreamingResponse(generator, media_type="text/event-stream")
     else:
-        completions = []
-        prompt_tokens = 0
-        completion_tokens = 0
+        text_completions = []
         for i in range(request.n):
-            content = await generate_completion(payload)
-            content = json.loads(content)
-            content["index"] = i
-            completion_tokens += content["completion_tokens"]
-            prompt_tokens = content["prompt_tokens"]
-            content.pop("completion_tokens")
-            content.pop("prompt_tokens")
-            if request.echo:
-                content["text"] = request.prompt + content["text"]
-            completions.append(content)
-    return CompletionResponse(
-        model=request.model,
-        choices=completions,
-        usage={
-            "prompt_tokens": prompt_tokens,
-            "completion_tokens": completion_tokens,
-            "total_tokens": prompt_tokens + completion_tokens,
-        },
-    )
+            content = asyncio.create_task(generate_completion(payload))
+            text_completions.append(content)
+        
+        try:
+            all_tasks = await asyncio.gather(*text_completions)
+        except Exception as e:
+            return create_error_response(
+                ErrorCode.INTERNAL_ERROR,
+                str(e)
+            )
+
+        choices = []
+        usage = UsageInfo()
+        for i, content in enumerate(all_tasks):
+            if content["error_code"] != 0:
+                return create_error_response(content["error_code"], content["text"])
+            choices.append(
+                CompletionResponseChoice(
+                    index=i,
+                    text=content["text"],
+                    logprobs=content.get("logprobs", None),
+                    finish_reason=content.get("finish_reason", "stop"),
+                )
+            )
+            task_usage = UsageInfo.parse_obj(content["usage"])
+            for usage_key, usage_value in task_usage.dict().items():
+                setattr(usage, usage_key, getattr(usage, usage_key) + usage_value)
+
+        return CompletionResponse(
+            model=request.model,
+            choices=choices,
+            usage=UsageInfo.parse_obj(usage)
+        )
+
+async def generate_completion_stream_generator(payload: Dict[str, Any], n: int):
+    model_name = payload["model"]
+    id = f"cmpl-{shortuuid.random()}"
+    finish_stream_events = []
+    for i in range(n):
+        previous_text = ""
+        async for content in generate_completion_stream(payload):
+            if content["error_code"] != 0:
+                yield f"data: {json.dumps(content, ensure_ascii=False)}\n\n"
+                yield "data: [DONE]\n\n"
+                return
+            decoded_unicode = content["text"].replace('\ufffd', '')
+            delta_text = decoded_unicode[len(previous_text):]
+            previous_text = decoded_unicode
+
+            choice_data = CompletionResponseStreamChoice(
+                index=i,
+                text=delta_text,
+                logprobs=content.get("logprobs", None),
+                finish_reason=content.get("finish_reason", None),
+            )
+            chunk = CompletionStreamResponse(
+                id=id,
+                object="text_completion",
+                choices=[choice_data],
+                model=model_name
+            )
+            if len(delta_text) == 0:
+                if content.get("finish_reason", None) is not None:
+                    finish_stream_events.append(chunk)
+                continue
+            yield f"data: {chunk.json(exclude_unset=True, ensure_ascii=False)}\n\n"
+    # There is not "content" field in the last delta message, so exclude_none to exclude field "content".
+    for finish_chunk in finish_stream_events:
+        yield f"data: {finish_chunk.json(exclude_unset=True, ensure_ascii=False)}\n\n"
+    yield "data: [DONE]\n\n"
+
+
+async def generate_completion_stream(payload: Dict[str, Any]):
+    controller_address = app_settings.controller_address
+    async with httpx.AsyncClient() as client:
+        worker_addr = await _get_worker_address(payload["model"], client)
+
+        delimiter = b"\0"
+        async with client.stream(
+            "POST",
+            worker_addr + "/worker_generate_completion_stream",
+            headers=headers,
+            json=payload,
+            timeout=WORKER_API_TIMEOUT,
+        ) as response:
+            # content = await response.aread()
+            async for raw_chunk in response.aiter_raw():
+                for chunk in raw_chunk.split(delimiter):
+                    if not chunk:
+                        continue
+                    data = json.loads(chunk.decode())
+                    yield data
 
 
 async def generate_completion(payload: Dict[str, Any]):
@@ -302,9 +532,12 @@ async def generate_completion(payload: Dict[str, Any]):
         return completion
 
 
-@app.post("/v1/create_embeddings")
+@app.post("/v1/embeddings")
 async def create_embeddings(request: EmbeddingsRequest):
     """Creates embeddings for the text"""
+    error_check_ret = await check_model(request)
+    if error_check_ret is not None:
+        return error_check_ret
     payload = {
         "model": request.model,
         "input": request.input,
@@ -316,11 +549,12 @@ async def create_embeddings(request: EmbeddingsRequest):
     return EmbeddingsResponse(
         data=data,
         model=request.model,
-        usage={
-            "prompt_tokens": embedding["token_num"],
-            "total_tokens": embedding["token_num"],
-        },
-    )
+        usage=UsageInfo(
+            prompt_tokens = embedding["token_num"],
+            total_tokens = embedding["token_num"],
+            completion_tokens = None,
+        ),
+    ).dict(exclude_none=True)
 
 
 async def get_embedding(payload: Dict[str, Any]):
@@ -373,4 +607,4 @@ async def get_embedding(payload: Dict[str, Any]):
 
     logger.debug(f"==== args ====\n{args}")
 
-    uvicorn.run(app, host=args.host, port=args.port, log_level="info")
+    uvicorn.run("fastchat.serve.openai_api_server:app", host=args.host, port=args.port, log_level="info", reload=True)


Test patch:
diff --git a/playground/test_embedding/test_classification.py b/playground/test_embedding/test_classification.py
index 9f0154e20..393827bb4 100644
--- a/playground/test_embedding/test_classification.py
+++ b/playground/test_embedding/test_classification.py
@@ -22,7 +22,7 @@ def get_embedding_from_api(word, model="vicuna-7b-v1.1"):
         embedding = np.array(resp["data"][0]["embedding"])
         return embedding
 
-    url = "http://localhost:8000/v1/create_embeddings"
+    url = "http://localhost:8000/v1/embeddings"
     headers = {"Content-Type": "application/json"}
     data = json.dumps({"model": model, "input": word})
 
diff --git a/playground/test_embedding/test_semantic_search.py b/playground/test_embedding/test_semantic_search.py
index 5c8b344b0..879b240b6 100644
--- a/playground/test_embedding/test_semantic_search.py
+++ b/playground/test_embedding/test_semantic_search.py
@@ -24,7 +24,7 @@ def get_embedding_from_api(word, model="vicuna-7b-v1.1"):
         embedding = np.array(resp["data"][0]["embedding"])
         return embedding
 
-    url = "http://localhost:8000/v1/create_embeddings"
+    url = "http://localhost:8000/v1/embeddings"
     headers = {"Content-Type": "application/json"}
     data = json.dumps({"model": model, "input": word})
 
diff --git a/playground/test_embedding/test_sentence_similarity.py b/playground/test_embedding/test_sentence_similarity.py
index 4a3838b8c..0b9a54081 100644
--- a/playground/test_embedding/test_sentence_similarity.py
+++ b/playground/test_embedding/test_sentence_similarity.py
@@ -16,7 +16,7 @@ def get_embedding_from_api(word, model="vicuna-7b-v1.1"):
         embedding = np.array(resp["data"][0]["embedding"])
         return embedding
 
-    url = "http://localhost:8000/v1/create_embeddings"
+    url = "http://localhost:8000/v1/embeddings"
     headers = {"Content-Type": "application/json"}
     data = json.dumps({"model": model, "input": word})
 
diff --git a/tests/test_openai_curl.sh b/tests/test_openai_curl.sh
index 19728c450..9b57b50f1 100644
--- a/tests/test_openai_curl.sh
+++ b/tests/test_openai_curl.sh
@@ -20,7 +20,7 @@ curl http://localhost:8000/v1/completions \
 
 echo
 
-curl http://localhost:8000/v1/create_embeddings \
+curl http://localhost:8000/v1/embeddings \
   -H "Content-Type: application/json" \
   -d '{
     "model": "vicuna-7b-v1.1",