Ianmacleod/completion sync error throws 4xx (#234)

* changing 5xx error to 4xx error

* .

* .

* adding completion stream changes

* parsing error dictionary

* .

* .

* fixing error handling for 400

* .

* hacky way of fixing completion stream w error message

* cleanup

* cleanup, add docs

* .

* fixing indentation on docs

Author: ian-scale

docs/getting_started.md

@@ -74,7 +74,11 @@ stream = Completion.create(
 )
 
 for response in stream:
-    if response.output:
-        print(response.output.text, end="")
-        sys.stdout.flush()
+    try:
+        if response.output:
+            print(response.output.text, end="")
+            sys.stdout.flush()
+    except: # an error occurred
+        print(stream.text) # print the error message out 
+        break
 ```

docs/guides/completions.md

@@ -67,6 +67,8 @@ applications. When streaming, tokens will be sent as data-only
 
 To enable token streaming, pass `stream=True` to either [Completion.create](../../api/python_client/#llmengine.completion.Completion.create) or [Completion.acreate](../../api/python_client/#llmengine.completion.Completion.acreate).
 
+Note that errors from streaming calls are returned back to the user as plain-text messages and currently need to be handled by the client.
+
 An example of token streaming using the synchronous Completions API looks as follows:
 
 === "Token streaming with synchronous API in python"
@@ -85,9 +87,13 @@ stream = Completion.create(
 )
 
 for response in stream:
-    if response.output:
-        print(response.output.text, end="")
-        sys.stdout.flush()
+    try:
+        if response.output:
+            print(response.output.text, end="")
+            sys.stdout.flush()
+    except: # an error occurred
+        print(stream.text) # print the error message out 
+        break
 ```
 
 ## Async requests

model-engine/model_engine_server/api/llms_v1.py

@@ -199,6 +199,8 @@ async def create_completion_sync_task(
         ) from exc
     except ObjectHasInvalidValueException as exc:
         raise HTTPException(status_code=400, detail=str(exc))
+    except InvalidRequestException as exc:
+        raise HTTPException(status_code=400, detail=str(exc))
     except EndpointUnsupportedInferenceTypeException as exc:
         raise HTTPException(
             status_code=400,
@@ -230,8 +232,12 @@ async def create_completion_stream_task(
         )
 
         async def event_generator():
-            async for message in response:
-                yield {"data": message.json()}
+            try:
+                async for message in response:
+                    yield {"data": message.json()}
+            except InvalidRequestException as exc:
+                yield {"data": {"error": {"status_code": 400, "detail": str(exc)}}}
+                return
 
         return EventSourceResponse(event_generator())
     except UpstreamServiceError:

model-engine/model_engine_server/domain/use_cases/llm_model_endpoint_use_cases.py

@@ -53,6 +53,8 @@
 from model_engine_server.domain.exceptions import (
     EndpointLabelsException,
     EndpointUnsupportedInferenceTypeException,
+    InvalidRequestException,
+    UpstreamServiceError,
 )
 from model_engine_server.domain.gateways.llm_artifact_gateway import LLMArtifactGateway
 from model_engine_server.domain.repositories import ModelBundleRepository
@@ -741,9 +743,15 @@ def model_output_to_completion_output(
                     num_completion_tokens=model_output["details"]["generated_tokens"],
                     tokens=tokens,
                 )
-            except Exception as e:
-                logger.exception(f"Error parsing text-generation-inference output {model_output}")
-                raise e
+            except Exception:
+                logger.exception(f"Error parsing text-generation-inference output {model_output}.")
+                if model_output.get("error_type") == "validation":
+                    raise InvalidRequestException(model_output.get("error"))  # trigger a 400
+                else:
+                    raise UpstreamServiceError(
+                        status_code=500, content=bytes(model_output["error"])
+                    )
+
         elif model_content.inference_framework == LLMInferenceFramework.VLLM:
             tokens = None
             if with_token_probs:
@@ -924,7 +932,6 @@ async def execute(
                 )
 
             output = json.loads(predict_result.result["result"])
-
             return CompletionSyncV1Response(
                 request_id=request_id,
                 output=self.model_output_to_completion_output(
@@ -1106,15 +1113,29 @@ async def execute(
                             token=result["result"]["token"]["text"],
                             log_prob=result["result"]["token"]["logprob"],
                         )
-                    yield CompletionStreamV1Response(
-                        request_id=request_id,
-                        output=CompletionStreamOutput(
-                            text=result["result"]["token"]["text"],
-                            finished=finished,
-                            num_completion_tokens=num_completion_tokens,
-                            token=token,
-                        ),
-                    )
+                    try:
+                        yield CompletionStreamV1Response(
+                            request_id=request_id,
+                            output=CompletionStreamOutput(
+                                text=result["result"]["token"]["text"],
+                                finished=finished,
+                                num_completion_tokens=num_completion_tokens,
+                                token=token,
+                            ),
+                        )
+                    except Exception:
+                        logger.exception(
+                            f"Error parsing text-generation-inference output. Result: {result['result']}"
+                        )
+                        if result["result"].get("error_type") == "validation":
+                            raise InvalidRequestException(
+                                result["result"].get("error")
+                            )  # trigger a 400
+                        else:
+                            raise UpstreamServiceError(
+                                status_code=500, content=result.get("error")
+                            )  # also change llms_v1.py that will return a 500 HTTPException so user can retry
+
                 else:
                     yield CompletionStreamV1Response(
                         request_id=request_id,