[Feature]Add output fallback support for OpenAI serving

fastdeploy/engine/args_utils.py

@@ -209,6 +209,18 @@ class EngineArgs:
     """
     tool parser plugin used to register user defined tool parsers
     """
+    output_fallback: Optional[str] = None
+    """
+    output fallback strategies to apply, separated by commas
+    """
+    output_fallback_plugin: Optional[list[str]] = None
+    """
+    output fallback plugin used to register user defined fallback strategies
+    """
+    output_fallback_config: Optional[Dict[str, Any]] = None
+    """
+    output fallback config keyed by strategy name
+    """
     enable_mm: bool = False
     """
     Flags to enable multi-modal model
@@ -922,6 +934,25 @@ def add_cli_args(parser: FlexibleArgumentParser) -> FlexibleArgumentParser:
             default=EngineArgs.tool_parser_plugin,
             help="tool parser plugin used to register user defined tool parsers",
         )
+        model_group.add_argument(
+            "--output-fallback",
+            type=str,
+            default=EngineArgs.output_fallback,
+            help="Output fallback strategies to apply, separated by commas.",
+        )
+        model_group.add_argument(
+            "--output-fallback-plugin",
+            type=str,
+            action="append",
+            default=EngineArgs.output_fallback_plugin,
+            help="Output fallback plugin used to register user defined fallback strategies.",
+        )
+        model_group.add_argument(
+            "--output-fallback-config",
+            type=json.loads,
+            default=EngineArgs.output_fallback_config,
+            help="Output fallback config keyed by strategy name.",
+        )
         model_group.add_argument(
             "--speculative-config",
             type=json.loads,

fastdeploy/engine/request.py

@@ -913,6 +913,8 @@ class CompletionOutput:
     delta_message: Optional[DeltaMessage] = None
     multipart: Optional[list[Any]] = None
     num_image_tokens: Optional[int] = None
+    fallback_truncated: Optional[bool] = False
+    skipped: Optional[bool] = False
 
     def to_dict(self):
         """
@@ -931,6 +933,14 @@ def to_dict(self):
             "text": self.text,
             "reasoning_content": self.reasoning_content,
             "reasoning_token_num": self.reasoning_token_num,
+            "tool_calls": self.tool_calls,
+            "speculate_metrics": self.speculate_metrics,
+            "completion_tokens": self.completion_tokens,
+            "delta_message": self.delta_message,
+            "multipart": self.multipart,
+            "num_image_tokens": self.num_image_tokens,
+            "fallback_truncated": self.fallback_truncated,
+            "skipped": self.skipped,
         }
 
     @classmethod

fastdeploy/entrypoints/openai/api_server.py

@@ -79,6 +79,7 @@
     log_request_error,
 )
 from fastdeploy.metrics.metrics import get_filtered_metrics
+from fastdeploy.output.fallback import OutputFallbackManager
 from fastdeploy.utils import (
     ExceptionHandler,
     FlexibleArgumentParser,
@@ -102,6 +103,18 @@
 chat_template = load_chat_template(args.chat_template, args.model)
 if args.tool_parser_plugin:
     ToolParserManager.import_tool_parser(args.tool_parser_plugin)
+if args.output_fallback_plugin:
+    for output_fallback_plugin in args.output_fallback_plugin:
+        OutputFallbackManager.import_fallback_plugin(output_fallback_plugin)
+
+output_fallback_manager = None
+if args.output_fallback:
+    output_fallback_names = [name.strip() for name in args.output_fallback.split(",") if name.strip()]
+    output_fallback_manager = OutputFallbackManager(
+        strategies=output_fallback_names,
+        config=args.output_fallback_config,
+    )
+
 llm_engine = None
 
 MAX_CONCURRENT_CONNECTIONS = (args.max_concurrency + args.workers - 1) // args.workers
@@ -224,6 +237,12 @@ async def lifespan(app: FastAPI):
         max_logprobs=args.max_logprobs,
     )
     await engine_client.connection_manager.initialize()
+    # The data_processor owns output-fallback application: strategies run on
+    # the raw decoded stream BEFORE reasoning/tool parsing, so all sub-streams
+    # (content / reasoning / tool calls) benefit. Serving handlers no longer
+    # invoke the manager themselves.
+    if output_fallback_manager is not None and getattr(engine_client, "data_processor", None) is not None:
+        engine_client.data_processor.output_fallback_manager = output_fallback_manager
     app.state.dynamic_load_weight = args.dynamic_load_weight
     model_handler = OpenAIServingModels(
         model_paths,
@@ -827,6 +846,9 @@ def process_object(obj):
         "enable_mm_output": args.enable_mm_output,
         "tool_call_parser": args.tool_call_parser,
         "tool_parser_plugin": args.tool_parser_plugin,
+        "output_fallback": args.output_fallback,
+        "output_fallback_plugin": args.output_fallback_plugin,
+        "output_fallback_config": args.output_fallback_config,
     }
 
     # GPU info

fastdeploy/entrypoints/openai/response_processors.py

@@ -209,8 +209,8 @@ async def process_response_chat(
                                 )
                             else:
                                 self.data_processor.process_response_dict(
-                                    response_dict=request_output,
-                                    stream=stream,
+                                    response_dict=part["request_output"],
+                                    stream=False,
                                     include_stop_str_in_output=include_stop_str_in_output,
                                     request=request,
                                     prompt_tokens=prompt_tokens,

fastdeploy/entrypoints/openai/serving_chat.py

@@ -246,6 +246,7 @@ async def chat_completion_stream_generator(
         num_cached_tokens = 0
         num_image_tokens = [0] * num_choices
         tool_called = [False] * num_choices
+        fallback_truncated_choices = set()
         inference_start_time = [0] * num_choices
         max_streaming_response_tokens = (
             request.max_streaming_response_tokens
@@ -403,6 +404,8 @@ async def chat_completion_stream_generator(
                         first_iteration = False
 
                     output = res["outputs"]
+                    if idx in fallback_truncated_choices:
+                        continue
                     output_top_logprobs = output["top_logprobs"]
                     output_draft_top_logprobs = output["draft_top_logprobs"]
                     previous_num_tokens[idx] += len(output["token_ids"])
@@ -439,6 +442,11 @@ async def chat_completion_stream_generator(
                     if output["skipped"] and not request.return_token_ids:
                         continue
 
+                    delta_text = "" if output["skipped"] else (output["text"] or "")
+                    fallback_truncated = bool(output.get("fallback_truncated"))
+                    if fallback_truncated:
+                        res["finished"] = True
+
                     delta_message = DeltaMessage(
                         reasoning_content=output["reasoning_content"],
                         tool_calls=output["tool_calls"],
@@ -452,7 +460,7 @@ async def chat_completion_stream_generator(
                             [{"type": "text", "text": ""}] if output["skipped"] else output["multipart"]
                         )
                     else:
-                        delta_message.content = "" if output["skipped"] else (output["text"] or "")
+                        delta_message.content = delta_text
 
                     if output.get("audio_content", None) is not None:
                         delta_message.audio_content = output["audio_content"]
@@ -496,6 +504,11 @@ async def chat_completion_stream_generator(
                         if res.get("error_msg") is not None and "Aborted" in res["error_msg"]:
                             choice.finish_reason = "abort"
 
+                        if fallback_truncated:
+                            choice.finish_reason = "length"
+                            fallback_truncated_choices.add(idx)
+                            await self.engine_client.abort(make_choice_id(request_id, idx), 1)
+
                         inference_start_time[idx] = 0
 
                     if request.collect_metrics:

fastdeploy/entrypoints/openai/serving_completion.py

@@ -478,6 +478,7 @@ async def completion_stream_generator(
             reasoning_tokens = [0] * num_choices
             first_iteration = [True] * num_choices
             tool_called = [False] * num_choices
+            fallback_truncated_choices = set()
             max_streaming_response_tokens = (
                 request.max_streaming_response_tokens
                 if request.max_streaming_response_tokens is not None
@@ -516,6 +517,8 @@ async def completion_stream_generator(
 
                 for res in response:
                     idx = get_choice_index(res["request_id"])
+                    if idx in fallback_truncated_choices:
+                        continue
                     if res.get("error_code", 200) != 200:
                         raise ValueError("{}".format(res["error_msg"]))
                     prompt_logprobs_res: Optional[PromptLogprobs] = None
@@ -597,9 +600,14 @@ async def completion_stream_generator(
                     if output["skipped"] and not request.return_token_ids:
                         continue
 
+                    delta_text = "" if output["skipped"] else (output["text"] or "")
+                    fallback_truncated = bool(output.get("fallback_truncated"))
+                    if fallback_truncated:
+                        res["finished"] = True
+
                     delta_message = CompletionResponseStreamChoice(
                         index=idx,
-                        text="" if output["skipped"] else (output["text"] or ""),
+                        text=delta_text,
                         prompt_token_ids=None,
                         completion_token_ids=output.get("token_ids") if request.return_token_ids else None,
                         tool_calls=output["tool_calls"],
@@ -625,6 +633,10 @@ async def completion_stream_generator(
                         )
                         if res.get("error_msg") is not None and "Aborted" in res["error_msg"]:
                             choices[-1].finish_reason = "abort"
+                        if fallback_truncated:
+                            choices[-1].finish_reason = "length"
+                            fallback_truncated_choices.add(idx)
+                            await self.engine_client.abort(make_choice_id(request_id, idx), 1)
                         inference_start_time[idx] = 0
 
                     send_idx = output.get("send_idx")
@@ -767,13 +779,14 @@ def request_output_to_completion_response(
                 prompt_logprobs_res = self._build_prompt_logprobs(
                     prompt_logprobs_tensors, num_prompt_logprobs, request.include_logprobs_decode_token
                 )
+            generated_text = output["text"]
             if request.echo:
                 prompt_text = self._echo_back_prompt(request, idx // (1 if request.n is None else request.n))
                 token_ids = [*prompt_token_ids, *output["token_ids"]]
-                output_text = prompt_text + output["text"]
+                output_text = prompt_text + generated_text
             else:
                 token_ids = output["token_ids"]
-                output_text = output["text"]
+                output_text = generated_text
             finish_reason = self.calc_finish_reason(
                 max_tokens_list[idx // (1 if request.n is None else request.n)],
                 final_res["output_token_ids"],

fastdeploy/input/base_processor.py

@@ -52,6 +52,7 @@
 from fastdeploy import envs
 from fastdeploy.input.utils import process_stop_token_ids
 from fastdeploy.logger.request_logger import RequestLogLevel, log_request
+from fastdeploy.output.fallback.base import OutputFallbackContext
 from fastdeploy.utils import data_processor_logger, make_choice_id, parse_choice_id
 
 _SAMPLING_EPS = 1e-5
@@ -74,6 +75,15 @@ def __init__(self, model_name_or_path, tokenizer_type="auto", reasoning_parser_o
         self.decode_status: Dict[str, list] = {}
         self.model_status_dict: Dict[str, dict] = {}
         self.tool_parser_dict: Dict = {}
+        # Per-request "fallback-corrected" stream history. Maintained
+        # separately from decode_status so that reasoning/tool parsers can
+        # be fed a coherent text stream after on_delta rewrites/holds it.
+        self.fallback_decode_status: Dict[str, dict] = {}
+        # Output fallback manager. Set externally (e.g. by api_server) once
+        # the processor instance is wired up. When set, fallback strategies
+        # are applied to the raw decoded stream BEFORE reasoning/tool
+        # parsing inside ``process_response_dict_*``. ``None`` disables.
+        self.output_fallback_manager = None
         # Token-encode cache shared by all subclasses.
         self._tokenize_cache: OrderedDict = OrderedDict()
         self._tokenize_cache_capacity: int = 128
@@ -363,6 +373,7 @@ def process_response_dict_normal(self, response_dict, **kwargs):
         req_id = response_dict["request_id"]
         request = kwargs.get("request", None)
         direct_decode = kwargs.get("direct_decode", False)
+        output_fallback_manager = self.output_fallback_manager if not kwargs.get("disable_output_fallback") else None
 
         if is_end and len(token_ids) > 0 and not kwargs.get("include_stop_str_in_output"):
             if token_ids[-1] in self.eos_token_ids:
@@ -376,6 +387,20 @@ def process_response_dict_normal(self, response_dict, **kwargs):
 
         if is_end:
             full_text = previous_texts + delta_text
+
+            # Apply output fallback to the full raw text BEFORE reasoning /
+            # tool parsing so all sub-streams (content, reasoning, tools)
+            # benefit from the rewrite.
+            if output_fallback_manager is not None and full_text:
+                context = OutputFallbackContext(
+                    request=request,
+                    request_id=req_id,
+                    stream=False,
+                    output=response_dict["outputs"],
+                    full_text=full_text,
+                )
+                full_text = output_fallback_manager.apply(full_text, context)
+
             response_dict["outputs"]["completion_tokens"] = full_text
             response_dict["outputs"]["text"] = full_text
 
@@ -411,19 +436,67 @@ def process_response_dict_streaming(self, response_dict, **kwargs):
         req_id = response_dict["request_id"]
         token_ids = response_dict["outputs"]["token_ids"]
         request = kwargs.get("request", None)
+        # Caller may opt out via ``disable_output_fallback`` (e.g. the
+        # response-processor's multimodal streaming branch where the raw text
+        # is later wrapped into a ``multipart`` payload).
+        output_fallback_manager = self.output_fallback_manager if not kwargs.get("disable_output_fallback") else None
 
         if is_end and len(token_ids) > 0 and not kwargs.get("include_stop_str_in_output"):
             if token_ids[-1] in self.eos_token_ids:
                 token_ids = token_ids[:-1]
 
         delta_text, previous_token_ids, previous_texts = self.ids2tokens(token_ids, req_id)
 
+        # ------------------------------------------------------------------
+        # Output fallback (applied BEFORE reasoning/tool parsing so that the
+        # whole raw stream — content + thinking + tool-call text — gets a
+        # chance to be repaired/rewritten).
+        # ------------------------------------------------------------------
+        fallback_held = False
+        if output_fallback_manager is not None:
+            state = self.fallback_decode_status.setdefault(req_id, {"previous_corrected": ""})
+            context = OutputFallbackContext(
+                request=request,
+                request_id=req_id,
+                stream=True,
+                output=response_dict["outputs"],
+                delta_text=delta_text,
+            )
+            decision = output_fallback_manager.on_delta(req_id, delta_text, context)
+            if decision.action == "truncate":
+                delta_text = decision.text
+                response_dict["finished"] = True
+                response_dict["outputs"]["fallback_truncated"] = True
+                is_end = True
+            elif decision.action == "hold":
+                fallback_held = True
+                delta_text = ""
+            else:  # send
+                delta_text = decision.text
+
+            if is_end and decision.action != "truncate":
+                finish_decision = output_fallback_manager.on_finish(req_id, context)
+                if finish_decision.text:
+                    delta_text = (delta_text or "") + finish_decision.text
+
+            # Override the parser-visible history with the corrected stream
+            # (held deltas contribute nothing; sent/flushed text is appended
+            # in arrival order). token_ids remain raw — parsers tolerate the
+            # minor text/token drift introduced by text-rewriting strategies.
+            corrected_previous = state["previous_corrected"]
+            state["previous_corrected"] = corrected_previous + (delta_text or "")
+            previous_texts = corrected_previous
+
         response_dict["outputs"]["text"] = delta_text
         response_dict["outputs"]["completion_tokens"] = delta_text
         response_dict["outputs"]["skipped"] = False
         response_dict["outputs"]["tool_calls"] = None
         response_dict["outputs"]["reasoning_content"] = ""
 
+        if fallback_held and not is_end:
+            response_dict["outputs"]["skipped"] = True
+            return response_dict
+
         if self.reasoning_parser:
             reasoning_delta_message = self.reasoning_parser.extract_reasoning_content_streaming(
                 previous_texts,
@@ -497,6 +570,10 @@ def process_response_dict_streaming(self, response_dict, **kwargs):
                 del self.tool_parser_dict[req_id]
             if req_id in self.model_status_dict:
                 del self.model_status_dict[req_id]
+            if req_id in self.fallback_decode_status:
+                del self.fallback_decode_status[req_id]
+            if output_fallback_manager is not None:
+                output_fallback_manager.cleanup(req_id)
 
         return response_dict