diff --git a/fastdeploy/engine/args_utils.py b/fastdeploy/engine/args_utils.py index 9c3f3c0f10e..99a7a85b103 100644 --- a/fastdeploy/engine/args_utils.py +++ b/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, diff --git a/fastdeploy/engine/request.py b/fastdeploy/engine/request.py index c17b8821ce2..076e35ef1f8 100644 --- a/fastdeploy/engine/request.py +++ b/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 diff --git a/fastdeploy/entrypoints/openai/api_server.py b/fastdeploy/entrypoints/openai/api_server.py index a4de9cf9375..3f5714c0f89 100644 --- a/fastdeploy/entrypoints/openai/api_server.py +++ b/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 diff --git a/fastdeploy/entrypoints/openai/response_processors.py b/fastdeploy/entrypoints/openai/response_processors.py index 2cfef290201..add9a260083 100644 --- a/fastdeploy/entrypoints/openai/response_processors.py +++ b/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, diff --git a/fastdeploy/entrypoints/openai/serving_chat.py b/fastdeploy/entrypoints/openai/serving_chat.py index c9f7c6764e5..b3313dfee1f 100644 --- a/fastdeploy/entrypoints/openai/serving_chat.py +++ b/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: diff --git a/fastdeploy/entrypoints/openai/serving_completion.py b/fastdeploy/entrypoints/openai/serving_completion.py index e8cb5d99fe4..6366563849b 100644 --- a/fastdeploy/entrypoints/openai/serving_completion.py +++ b/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"], diff --git a/fastdeploy/input/base_processor.py b/fastdeploy/input/base_processor.py index 3dd6ad664af..9c6b5beb3ce 100644 --- a/fastdeploy/input/base_processor.py +++ b/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,6 +436,10 @@ 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: @@ -418,12 +447,56 @@ def process_response_dict_streaming(self, response_dict, **kwargs): 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 diff --git a/fastdeploy/output/fallback/__init__.py b/fastdeploy/output/fallback/__init__.py new file mode 100644 index 00000000000..9ded92fffaa --- /dev/null +++ b/fastdeploy/output/fallback/__init__.py @@ -0,0 +1,32 @@ +""" +# Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License" +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +""" + +from fastdeploy.output.fallback.base import ( + OutputFallbackContext, + OutputFallbackStrategy, + StreamFallbackDecision, +) +from fastdeploy.output.fallback.manager import OutputFallbackManager +from fastdeploy.plugins import load_output_fallback_plugins + +__all__ = [ + "OutputFallbackContext", + "OutputFallbackManager", + "OutputFallbackStrategy", + "StreamFallbackDecision", +] + +load_output_fallback_plugins() diff --git a/fastdeploy/output/fallback/base.py b/fastdeploy/output/fallback/base.py new file mode 100644 index 00000000000..140c1de4b03 --- /dev/null +++ b/fastdeploy/output/fallback/base.py @@ -0,0 +1,97 @@ +""" +# Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License" +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Any, Literal, Optional + + +@dataclass +class OutputFallbackContext: + request: Optional[Any] + request_id: str + stream: bool + output: dict + full_text: Optional[str] = None + delta_text: Optional[str] = None + + +@dataclass +class StreamFallbackDecision: + """Per-delta decision returned by streaming strategy hooks. + + ``action`` semantics (consumed by ``OutputFallbackManager``): + + - ``send`` : emit ``text`` downstream as the current delta. + - ``hold`` : keep this delta buffered by the manager; no later strategies + run for this round, and the manager emits nothing. + - ``flush`` : used by ``on_finish`` to emit any remaining buffered text + after the stream ends. + - ``truncate``: send ``text`` as the final delta and stop further generation. + """ + + action: Literal["send", "hold", "flush", "truncate"] + text: str = "" + + +class OutputFallbackStrategy(ABC): + """Base class for output fallback strategies. + + Subclasses must implement the two primitives ``should_apply`` and ``apply``, + which operate on the **full** text. ``on_delta`` / ``on_finish`` are optional + hooks for streaming scenarios; the default implementations treat each delta + as an independent piece of text and are suitable for stateless strategies. + Strategies that need cross-chunk state (e.g. matching patterns spanning + multiple deltas) should override these hooks and persist state via the + per-request ``state`` dict managed by ``OutputFallbackManager``. + + The ``context`` argument carries request-level metadata; see + :class:`OutputFallbackContext` for the available fields and how to use them. + Built-in implementations may not read ``context`` today, but custom + strategies are free to. + """ + + name: str = "" + + def __init__(self, config: Optional[dict] = None): + self.config = config or {} + + @abstractmethod + def should_apply(self, text: str, context: OutputFallbackContext) -> bool: + """Return True if ``apply`` should run on ``text`` for this request.""" + + @abstractmethod + def apply(self, text: str, context: OutputFallbackContext) -> str: + """Transform the **full** text. Must be a pure function of its inputs; + do not store cross-call state on ``self``.""" + + def on_delta(self, delta_text: str, context: OutputFallbackContext, state: dict) -> StreamFallbackDecision: + """Streaming hook called for each delta. Default: stateless per-delta apply. + + Override when a strategy must buffer / look across multiple deltas. + Use the per-request ``state`` dict for any cross-delta state — never + store it on ``self`` (instances are shared across requests). + """ + if not self.should_apply(delta_text, context): + return StreamFallbackDecision(action="send", text=delta_text) + return StreamFallbackDecision(action="send", text=self.apply(delta_text, context)) + + def on_finish(self, context: OutputFallbackContext, state: dict) -> StreamFallbackDecision: + """Streaming hook called once after the last delta. Override to flush + any text buffered in ``state`` during ``on_delta``.""" + return StreamFallbackDecision(action="flush") diff --git a/fastdeploy/output/fallback/manager.py b/fastdeploy/output/fallback/manager.py new file mode 100644 index 00000000000..de845b317f1 --- /dev/null +++ b/fastdeploy/output/fallback/manager.py @@ -0,0 +1,194 @@ +""" +# Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License" +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +""" + +from __future__ import annotations + +import copy +import os +from typing import Callable, Optional, Union + +from fastdeploy.output.fallback.base import ( + OutputFallbackContext, + OutputFallbackStrategy, + StreamFallbackDecision, +) +from fastdeploy.utils import data_processor_logger, import_from_path, is_list_of + + +class OutputFallbackManager: + fallback_strategies: dict[str, type[OutputFallbackStrategy]] = {} + + @classmethod + def get_strategy(cls, name: str) -> type[OutputFallbackStrategy]: + name = name.replace("_", "-") + if name in cls.fallback_strategies: + return cls.fallback_strategies[name] + raise KeyError(f"output fallback strategy: '{name}' not found") + + @classmethod + def _register_strategy( + cls, + module: type[OutputFallbackStrategy], + strategy_name: Optional[Union[str, list[str]]] = None, + force: bool = True, + ) -> None: + if not issubclass(module, OutputFallbackStrategy): + raise TypeError(f"module must be subclass of OutputFallbackStrategy, but got {type(module)}") + if strategy_name is None: + strategy_name = module.name or module.__name__ + if isinstance(strategy_name, str): + strategy_name = [strategy_name] + for name in strategy_name: + name = name.replace("_", "-") + if not force and name in cls.fallback_strategies: + existed_module = cls.fallback_strategies[name] + raise KeyError(f"{name} is already registered at {existed_module.__module__}") + cls.fallback_strategies[name] = module + + @classmethod + def register( + cls, name: Optional[Union[str, list[str]]] = None, force: bool = True, module: Union[type, None] = None + ) -> Union[type, Callable]: + if not isinstance(force, bool): + raise TypeError(f"force must be a boolean, but got {type(force)}") + if not (name is None or isinstance(name, str) or is_list_of(name, str)): + raise TypeError("name must be None, an instance of str, or a sequence of str, " f"but got {type(name)}") + if module is not None: + cls._register_strategy(module=module, strategy_name=name, force=force) + return module + + def _register(module): + cls._register_strategy(module=module, strategy_name=name, force=force) + return module + + return _register + + @classmethod + def import_fallback_plugin(cls, plugin_path: str) -> None: + module_name = os.path.splitext(os.path.basename(plugin_path))[0] + try: + import_from_path(module_name, plugin_path) + except Exception: + data_processor_logger.exception( + "Failed to load output fallback module '%s' from %s.", module_name, plugin_path + ) + + def __init__(self, strategies: list[str], config: Optional[dict] = None): + self.strategies = [name.replace("_", "-") for name in strategies] + self.config = {name.replace("_", "-"): value for name, value in (config or {}).items()} + self.instances = [self.get_strategy(name)(self.config.get(name, {})) for name in self.strategies] + self.states: dict[str, dict[str, dict]] = {} + self._buffers: dict[str, str] = {} + + def apply(self, text: str, context: OutputFallbackContext) -> str: + result_text = text + for strategy in self.instances: + try: + if strategy.should_apply(result_text, context): + result_text = strategy.apply(result_text, context) + except Exception: + data_processor_logger.exception("Failed to apply output fallback strategy '%s'.", strategy.name) + return result_text + + def on_delta(self, request_id: str, delta_text: str, context: OutputFallbackContext) -> StreamFallbackDecision: + self._buffers[request_id] = self._buffers.get(request_id, "") + delta_text + buffer = self._buffers[request_id] + + current_text = buffer + trial_states: dict[str, dict] = {} + truncated = False + + for strategy in self.instances: + original_state = self._get_state(request_id, strategy.name) + trial_state = copy.deepcopy(original_state) + try: + decision = strategy.on_delta(current_text, context, trial_state) + except Exception: + data_processor_logger.exception( + "Failed to apply streaming output fallback strategy '%s'.", strategy.name + ) + continue + + trial_states[strategy.name] = trial_state + + if decision.action == "hold": + if not truncated: + return StreamFallbackDecision(action="hold", text="") + continue + + if decision.action == "truncate": + truncated = True + current_text = decision.text or current_text + continue + + current_text = decision.text + + for name, ts in trial_states.items(): + state = self._get_state(request_id, name) + state.clear() + state.update(ts) + self._buffers.pop(request_id, None) + + if truncated: + return StreamFallbackDecision(action="truncate", text=current_text) + return StreamFallbackDecision(action="send", text=current_text) + + def on_finish(self, request_id: str, context: OutputFallbackContext) -> StreamFallbackDecision: + buffer = self._buffers.pop(request_id, "") + if buffer: + current_text = buffer + for strategy in self.instances: + state = self._get_state(request_id, strategy.name) + try: + decision = strategy.on_delta(current_text, context, state) + except Exception: + data_processor_logger.exception( + "Failed to apply streaming output fallback strategy '%s'.", strategy.name + ) + continue + if decision.action == "hold": + try: + finish_decision = strategy.on_finish(context, state) + except Exception: + data_processor_logger.exception( + "Failed to finish streaming output fallback strategy '%s'.", strategy.name + ) + continue + current_text = finish_decision.text + else: + current_text = decision.text + return StreamFallbackDecision(action="flush", text=current_text) + + pending = "" + for strategy in self.instances: + state = self._get_state(request_id, strategy.name) + try: + decision = strategy.on_finish(context, state) + except Exception: + data_processor_logger.exception( + "Failed to finish streaming output fallback strategy '%s'.", strategy.name + ) + continue + if decision.text: + pending += decision.text + return StreamFallbackDecision(action="flush", text=pending) + + def cleanup(self, request_id: str) -> None: + self.states.pop(request_id, None) + self._buffers.pop(request_id, None) + + def _get_state(self, request_id: str, strategy_name: str) -> dict: + return self.states.setdefault(request_id, {}).setdefault(strategy_name, {}) diff --git a/fastdeploy/plugins/__init__.py b/fastdeploy/plugins/__init__.py index 96e30e5a56b..170344c9485 100644 --- a/fastdeploy/plugins/__init__.py +++ b/fastdeploy/plugins/__init__.py @@ -17,6 +17,7 @@ from .input_processor import load_input_processor_plugins from .model_register import load_model_register_plugins from .model_runner import load_model_runner_plugins +from .output_fallback import load_output_fallback_plugins from .reasoning_parser import load_reasoning_parser_plugins from .token_processor import load_token_processor_plugins from .tool_parser import load_tool_parser_plugins @@ -25,6 +26,7 @@ "load_model_register_plugins", "load_model_runner_plugins", "load_input_processor_plugins", + "load_output_fallback_plugins", "load_reasoning_parser_plugins", "load_token_processor_plugins", "load_tool_parser_plugins", diff --git a/fastdeploy/plugins/output_fallback/__init__.py b/fastdeploy/plugins/output_fallback/__init__.py new file mode 100644 index 00000000000..8ca72c95342 --- /dev/null +++ b/fastdeploy/plugins/output_fallback/__init__.py @@ -0,0 +1,34 @@ +""" +# Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License" +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +""" + +from fastdeploy.plugins.utils import load_plugins_by_group + +# make sure one process only loads plugins once +plugins_loaded = False +PLUGINS_GROUP = "fastdeploy.output_fallback_plugins" + + +def load_output_fallback_plugins(): + """load_output_fallback_plugins""" + global plugins_loaded + if plugins_loaded: + return + plugins_loaded = True + + plugins = load_plugins_by_group(group=PLUGINS_GROUP) + # general plugins, we only need to execute the loaded functions + for func in plugins.values(): + func() diff --git a/tests/engine/test_request_output.py b/tests/engine/test_request_output.py index d3494fea502..35cbb7d1a09 100644 --- a/tests/engine/test_request_output.py +++ b/tests/engine/test_request_output.py @@ -403,6 +403,31 @@ def test_from_dict_without_outputs_and_metrics(self): self.assertIsNone(request_output.outputs) self.assertIsNone(request_output.metrics) + def test_completion_output_to_dict_preserves_serving_fields(self): + output = CompletionOutput( + index=0, + send_idx=1, + token_ids=[100, 200], + text="delta", + completion_tokens="delta", + multipart=[{"type": "text", "text": "delta"}], + num_image_tokens=3, + fallback_truncated=True, + skipped=True, + ) + + output_dict = output.to_dict() + restored = CompletionOutput.from_dict(output_dict) + + self.assertIn("tool_calls", output_dict) + self.assertEqual(output_dict["completion_tokens"], "delta") + self.assertEqual(output_dict["multipart"], [{"type": "text", "text": "delta"}]) + self.assertEqual(output_dict["num_image_tokens"], 3) + self.assertTrue(output_dict["fallback_truncated"]) + self.assertTrue(output_dict["skipped"]) + self.assertTrue(restored.fallback_truncated) + self.assertTrue(restored.skipped) + if __name__ == "__main__": unittest.main() diff --git a/tests/entrypoints/openai/test_api_server.py b/tests/entrypoints/openai/test_api_server.py index 94f9d641bc2..d1881242a3f 100644 --- a/tests/entrypoints/openai/test_api_server.py +++ b/tests/entrypoints/openai/test_api_server.py @@ -42,6 +42,9 @@ def _build_args(**overrides): revision=None, chat_template=None, tool_parser_plugin=None, + output_fallback=None, + output_fallback_plugin=None, + output_fallback_config=None, host="0.0.0.0", port=9000, metrics_port=None, diff --git a/tests/entrypoints/openai/test_metrics_routes.py b/tests/entrypoints/openai/test_metrics_routes.py index eecdef22188..ef454b1c5c2 100644 --- a/tests/entrypoints/openai/test_metrics_routes.py +++ b/tests/entrypoints/openai/test_metrics_routes.py @@ -37,6 +37,9 @@ def _build_mock_args(): revision=None, chat_template=None, tool_parser_plugin=None, + output_fallback=None, + output_fallback_plugin=None, + output_fallback_config=None, # server/network host="0.0.0.0", port=8000, diff --git a/tests/entrypoints/openai/test_response_processors.py b/tests/entrypoints/openai/test_response_processors.py index 6c3a153f0fe..4ef8dd58039 100644 --- a/tests/entrypoints/openai/test_response_processors.py +++ b/tests/entrypoints/openai/test_response_processors.py @@ -123,6 +123,37 @@ async def test_streaming_buffer_accumulation(self): self.assertEqual(results, []) self.assertEqual(self.processor_mm._mm_buffer, [[[33, 44]]]) + async def test_streaming_text_mm_output_keeps_output_fallback_enabled(self): + request_outputs = [{"request_id": "req4", "outputs": {"decode_type": 0, "token_ids": [1], "text": "raw"}}] + + results = [ + r + async for r in self.processor_mm.process_response_chat( + request_outputs, stream=True, include_stop_str_in_output=False, request=None + ) + ] + + self.assertEqual(results[0]["outputs"]["multipart"][0]["text"], "raw") + call_kwargs = self.mock_data_processor.process_response_dict.call_args.kwargs + self.assertNotIn("disable_output_fallback", call_kwargs) + + async def test_non_streaming_text_mm_output_keeps_output_fallback_enabled(self): + request_outputs = [ + {"request_id": "req5", "outputs": {"decode_type": 0, "token_ids": [10], "text": "raw"}}, + {"request_id": "req5", "outputs": {"decode_type": 0, "token_ids": [2], "text": "done"}}, + ] + + results = [ + r + async for r in self.processor_mm.process_response_chat( + request_outputs, stream=False, include_stop_str_in_output=False, request=None + ) + ] + + self.assertEqual(results[0]["outputs"]["multipart"][0]["text"], "done") + call_kwargs = self.mock_data_processor.process_response_dict.call_args.kwargs + self.assertNotIn("disable_output_fallback", call_kwargs) + async def test_non_streaming_accumulate_and_emit(self): """非流式模式：等 eos_token_id 才输出 multipart（text+image）""" request_outputs = [ diff --git a/tests/entrypoints/openai/test_serving_chat.py b/tests/entrypoints/openai/test_serving_chat.py index 26f3f1c49b9..7a90bbc0106 100644 --- a/tests/entrypoints/openai/test_serving_chat.py +++ b/tests/entrypoints/openai/test_serving_chat.py @@ -952,6 +952,97 @@ async def mock_async_generator(): # logprobs should be None when not requested self.assertIsNone(choice.get("logprobs")) + async def test_chat_completion_stream_generator_with_fallback_truncate(self): + """Processor-level fallback truncate sets finish_reason and aborts generation.""" + request = ChatCompletionRequest( + messages=[{"role": "user", "content": "Hello"}], + stream=True, + n=2, + ) + request_id = "test_request_fallback_truncate" + model_name = "test_model" + prompt_token_ids = [1, 2, 3] + prompt_tokens = "Hello world" + + def make_response(choice_index, text, finished, fallback_truncated=False, skipped=False): + return { + "request_id": f"{request_id}::n::{choice_index}", + "error_code": 200, + "metrics": { + "first_token_time": 1234567890, + "inference_start_time": 1234567880, + "arrival_time": 1234567890, + "request_start_time": 1234567870, + }, + "prompt_logprobs": None, + "outputs": { + "send_idx": 0, + "index": choice_index, + "token_ids": [5], + "text": text, + "top_logprobs": None, + "draft_top_logprobs": None, + "tool_calls": None, + "reasoning_content": "", + "multipart": [{"type": "text", "text": text}], + "skipped": skipped, + "fallback_truncated": fallback_truncated, + }, + "finished": finished, + "num_cached_tokens": 0, + "num_input_image_tokens": 0, + "num_input_video_tokens": 0, + } + + # Processor signals truncate on choice 0 (text already corrected upstream). + truncated_response = make_response(0, "corrected", True, fallback_truncated=True) + normal_response = make_response(1, "normal", True) + + mock_dealer = MagicMock() + mock_response_queue = AsyncMock() + mock_response_queue.get.side_effect = [truncated_response, normal_response] + self.chat_completion_handler.engine_client.connection_manager.get_connection = AsyncMock( + return_value=(mock_dealer, mock_response_queue) + ) + self.chat_completion_handler.engine_client.connection_manager.cleanup_request = AsyncMock() + self.chat_completion_handler.engine_client.semaphore = MagicMock() + self.chat_completion_handler.engine_client.semaphore.acquire = AsyncMock(return_value=True) + self.chat_completion_handler.engine_client.semaphore.release = MagicMock() + self.chat_completion_handler.engine_client.check_model_weight_status = Mock(return_value=False) + self.chat_completion_handler.engine_client.abort = AsyncMock() + + mock_response_processor = MagicMock() + mock_response_processor.enable_multimodal_content.return_value = False + + def process_response_chat(response, **kwargs): + async def mock_async_generator(): + yield response + + return mock_async_generator() + + mock_response_processor.process_response_chat.side_effect = process_response_chat + + with patch( + "fastdeploy.entrypoints.openai.serving_chat.ChatResponseProcessor", return_value=mock_response_processor + ): + results = [] + async for chunk in self.chat_completion_handler.chat_completion_stream_generator( + request, request_id, model_name, prompt_token_ids, prompt_tokens, max_tokens=100 + ): + results.append(chunk) + + choices = [] + for result in results: + if result.strip() == "data: [DONE]": + continue + chunk_data = json.loads(result.replace("data: ", "").strip()) + choices.extend(chunk_data.get("choices", [])) + + truncated_choices = [choice for choice in choices if choice.get("finish_reason") == "length"] + self.assertEqual(len(truncated_choices), 1) + self.assertEqual(truncated_choices[0]["index"], 0) + self.chat_completion_handler.engine_client.abort.assert_awaited_once() + async def test_chat_completion_full_generator_with_prompt_logprobs(self): """Test chat_completion_full_generator with prompt_logprobs enabled""" # Create mock request with prompt_logprobs enabled diff --git a/tests/entrypoints/openai/test_serving_completion.py b/tests/entrypoints/openai/test_serving_completion.py index 28e7ca0bc76..901ce389be1 100644 --- a/tests/entrypoints/openai/test_serving_completion.py +++ b/tests/entrypoints/openai/test_serving_completion.py @@ -15,6 +15,7 @@ """ import asyncio +import json import unittest from typing import List from unittest.mock import AsyncMock, Mock, patch @@ -859,6 +860,114 @@ async def test_completion_stream_generator_without_logprobs(self): self.assertTrue(prompt_logprobs_null, "prompt_logprobs should be null when not requested") self.assertTrue(logprobs_null, "logprobs should be null when not requested") + async def test_completion_stream_generator_with_fallback_truncate(self): + """Processor-level fallback truncate sets finish_reason and aborts generation.""" + mock_engine_client = Mock() + mock_engine_client.semaphore = Mock() + mock_engine_client.semaphore.acquire = AsyncMock() + mock_engine_client.semaphore.release = Mock() + mock_engine_client.connection_manager = AsyncMock() + mock_engine_client.data_processor = Mock() + mock_engine_client.ori_vocab_size = 1000 + mock_engine_client.check_model_weight_status.return_value = False + mock_engine_client.check_health.return_value = (True, "Healthy") + mock_engine_client.abort = AsyncMock() + mock_engine_client.data_processor.process_response_dict = Mock() + + request_id = "test_request_fallback_truncate" + + def make_response(choice_index, text, finished, fallback_truncated=False, skipped=False): + return { + "request_id": f"{request_id}::n::{choice_index}", + "error_code": 200, + "metrics": { + "arrival_time": 1234567890, + "inference_start_time": 1234567880, + "first_token_time": 1234567890, + }, + "prompt_logprobs": None, + "outputs": { + "text": text, + "token_ids": [100], + "top_logprobs": None, + "draft_top_logprobs": None, + "send_idx": 0, + "completion_tokens": "1", + "num_cache_tokens": 0, + "num_image_tokens": 0, + "reasoning_token_num": 0, + "tool_calls": None, + "reasoning_content": "", + "skipped": skipped, + "fallback_truncated": fallback_truncated, + }, + "finished": finished, + } + + # Processor signals truncate on choice 0 (text already corrected upstream). + truncated_response = make_response(0, "corrected", True, fallback_truncated=True) + normal_response = make_response(1, "normal", True) + + mock_dealer = Mock() + mock_dealer.write = Mock() + mock_response_queue = AsyncMock() + mock_response_queue.get.side_effect = [ + [truncated_response], + [normal_response], + ] + mock_engine_client.connection_manager.get_connection.return_value = (mock_dealer, mock_response_queue) + mock_engine_client.connection_manager.cleanup_request = AsyncMock() + + serving_completion = OpenAIServingCompletion( + mock_engine_client, + None, + "pid", + None, + 360, + ) + + mock_request = Mock() + mock_request.prompt_logprobs = None + mock_request.logprobs = None + mock_request.include_draft_logprobs = False + mock_request.return_token_ids = False + mock_request.include_stop_str_in_output = False + mock_request.max_streaming_response_tokens = 1 + mock_request.max_tokens = None + mock_request.stream_options = Mock() + mock_request.stream_options.include_usage = False + mock_request.n = 2 + mock_request.echo = False + mock_request.collect_metrics = False + mock_request.suffix = None + + result_generator = serving_completion.completion_stream_generator( + request=mock_request, + num_choices=2, + request_id=request_id, + created_time=1234567890, + model_name="test_model", + prompt_batched_token_ids=[[1, 2, 3]], + prompt_tokens_list=["hello"], + max_tokens_list=[100], + ) + + results = [] + async for result in result_generator: + results.append(result) + + choices = [] + for result in results: + if result.strip() == "data: [DONE]": + continue + chunk_data = json.loads(result.replace("data: ", "").strip()) + choices.extend(chunk_data.get("choices", [])) + + truncated_choices = [choice for choice in choices if choice.get("finish_reason") == "length"] + self.assertEqual(len(truncated_choices), 1) + self.assertEqual(truncated_choices[0]["index"], 0) + mock_engine_client.abort.assert_awaited_once() + async def test_completion_full_generator_with_prompt_logprobs(self): """Test completion_full_generator with prompt_logprobs enabled""" # Mock the engine client and its dependencies diff --git a/tests/entrypoints/openai/test_wrap_streaming_generator.py b/tests/entrypoints/openai/test_wrap_streaming_generator.py index bc8fca03c82..8c68a1e49d4 100644 --- a/tests/entrypoints/openai/test_wrap_streaming_generator.py +++ b/tests/entrypoints/openai/test_wrap_streaming_generator.py @@ -32,6 +32,9 @@ revision=None, chat_template=None, tool_parser_plugin=None, + output_fallback=None, + output_fallback_plugin=None, + output_fallback_config=None, max_concurrency=100, # Add required attribute max_num_seqs=100, tensor_parallel_size=1, diff --git a/tests/input/test_multimodal_processor.py b/tests/input/test_multimodal_processor.py index d7bd6e770ca..ea5c91bf9b4 100644 --- a/tests/input/test_multimodal_processor.py +++ b/tests/input/test_multimodal_processor.py @@ -59,6 +59,8 @@ def _make_processor(model_type, **overrides): proc.model_status_dict = {} proc.decode_status = {} proc.tool_parser_dict = {} + proc.fallback_decode_status = {} + proc.output_fallback_manager = None proc.input_max_tokens = None proc.max_completion_tokens = None proc.reasoning_max_tokens = None diff --git a/tests/input/test_preprocess.py b/tests/input/test_preprocess.py index d940d2800d4..be79b329e4b 100644 --- a/tests/input/test_preprocess.py +++ b/tests/input/test_preprocess.py @@ -62,7 +62,10 @@ def test_create_processor_text_normal_path(self): mock_dp = MagicMock() with ( - patch.dict("sys.modules", {"fastdeploy.plugins": None, "fastdeploy.plugins.input_processor": None}), + patch( + "fastdeploy.plugins.input_processor.load_input_processor_plugins", + side_effect=ImportError("mock plugin unavailable"), + ), patch("fastdeploy.input.text_processor.TextProcessor", return_value=mock_dp), ): pp.create_processor() @@ -76,7 +79,10 @@ def test_unsupported_mm_arch_raises(self): config = _make_model_config("UnknownMMArch", enable_mm=True) pp = InputPreprocessor(model_config=config) - with patch.dict("sys.modules", {"fastdeploy.plugins": None, "fastdeploy.plugins.input_processor": None}): + with patch( + "fastdeploy.plugins.input_processor.load_input_processor_plugins", + side_effect=ImportError("mock plugin unavailable"), + ): with self.assertRaises(ValueError): pp.create_processor() diff --git a/tests/input/test_text_processor.py b/tests/input/test_text_processor.py index dee0fd02a31..87665ef00c0 100644 --- a/tests/input/test_text_processor.py +++ b/tests/input/test_text_processor.py @@ -155,6 +155,24 @@ def get_choice_index(compound_id: str) -> int: return int(compound_id.rsplit(CHOICE_SEPARATOR, 1)[1]) raise ValueError(f"No choice index in request_id: {compound_id}") + def is_list_of(value, typ, *, check="first"): + if not isinstance(value, list): + return False + if check == "first": + return len(value) == 0 or isinstance(value[0], typ) + if check == "all": + return all(isinstance(v, typ) for v in value) + raise AssertionError(check) + + def import_from_path(module_name, file_path): + spec = importlib.util.spec_from_file_location(module_name, file_path) + if spec is None: + raise ModuleNotFoundError(f"No module named '{module_name}'") + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + spec.loader.exec_module(module) + return module + utils_module = types.ModuleType("fastdeploy.utils") utils_module.data_processor_logger = dummy_logger utils_module.CHOICE_SEPARATOR = CHOICE_SEPARATOR @@ -162,6 +180,8 @@ def get_choice_index(compound_id: str) -> int: utils_module.parse_choice_id = parse_choice_id utils_module.get_base_request_id = get_base_request_id utils_module.get_choice_index = get_choice_index + utils_module.is_list_of = is_list_of + utils_module.import_from_path = import_from_path envs_module = types.ModuleType("fastdeploy.envs") envs_module.FD_USE_HF_TOKENIZER = False @@ -796,6 +816,45 @@ def test_process_response_streaming_clears_state(self): self.assertEqual(result["outputs"]["text"], "7") self.assertNotIn(req_id, processor.decode_status) + def test_process_response_streaming_uses_compound_request_id_for_fallback_state(self): + processor = self.processor + manager = mock.Mock() + manager.on_delta.return_value = SimpleNamespace(action="send", text="7") + manager.on_finish.return_value = SimpleNamespace(text="") + processor.output_fallback_manager = manager + + req_id = "stream::n::1" + processor.decode_status[req_id] = [0, 0, [], ""] + response = {"finished": True, "request_id": req_id, "outputs": {"token_ids": [7]}} + + result = processor.process_response_dict_streaming(response, enable_thinking=False) + self.assertEqual(result["outputs"]["text"], "7") + manager.on_delta.assert_called_once() + self.assertEqual(manager.on_delta.call_args.args[0:2], (req_id, "7")) + self.assertEqual(manager.on_delta.call_args.args[2].request_id, req_id) + manager.on_finish.assert_called_once() + self.assertEqual(manager.on_finish.call_args.args[0], req_id) + manager.cleanup.assert_called_once_with(req_id) + self.assertNotIn(req_id, processor.fallback_decode_status) + + def test_process_response_streaming_final_hold_finishes_without_flush_text(self): + processor = self.processor + manager = mock.Mock() + manager.on_delta.return_value = SimpleNamespace(action="hold", text="") + manager.on_finish.return_value = SimpleNamespace(text="") + processor.output_fallback_manager = manager + + req_id = "stream::n::1" + processor.decode_status[req_id] = [0, 0, [], ""] + response = {"finished": True, "request_id": req_id, "outputs": {"token_ids": [7]}} + + result = processor.process_response_dict_streaming(response, enable_thinking=False) + self.assertFalse(result["outputs"]["skipped"]) + self.assertEqual(result["outputs"]["text"], "") + manager.on_finish.assert_called_once() + manager.cleanup.assert_called_once_with(req_id) + self.assertNotIn(req_id, processor.fallback_decode_status) + def test_process_response_streaming_with_reasoning_and_tools(self): processor = self.processor self.processor.model_status_dict["normal"] = "think_start" diff --git a/tests/output/test_fallback.py b/tests/output/test_fallback.py new file mode 100644 index 00000000000..cac6cf4d8bc --- /dev/null +++ b/tests/output/test_fallback.py @@ -0,0 +1,432 @@ +""" +# Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License" +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +""" + +from pathlib import Path + +from fastdeploy.output.fallback import ( + OutputFallbackContext, + OutputFallbackManager, + OutputFallbackStrategy, + StreamFallbackDecision, +) + + +@OutputFallbackManager.register("test-replace", force=True) +class ReplaceStrategy(OutputFallbackStrategy): + name = "test-replace" + + def should_apply(self, text: str, context: OutputFallbackContext) -> bool: + return "bad" in text + + def apply(self, text: str, context: OutputFallbackContext) -> str: + return text.replace("bad", "good") + + +@OutputFallbackManager.register("test-suffix", force=True) +class SuffixStrategy(OutputFallbackStrategy): + name = "test-suffix" + + def should_apply(self, text: str, context: OutputFallbackContext) -> bool: + return True + + def apply(self, text: str, context: OutputFallbackContext) -> str: + return text + "-suffix" + + def on_delta(self, delta_text: str, context: OutputFallbackContext, state: dict) -> StreamFallbackDecision: + return StreamFallbackDecision(action="send", text=self.apply(delta_text, context)) + + +@OutputFallbackManager.register("test-hold", force=True) +class HoldStrategy(OutputFallbackStrategy): + name = "test-hold" + + def should_apply(self, text: str, context: OutputFallbackContext) -> bool: + return True + + def apply(self, text: str, context: OutputFallbackContext) -> str: + return text + + def on_delta(self, delta_text: str, context: OutputFallbackContext, state: dict) -> StreamFallbackDecision: + state["held"] = state.get("held", "") + delta_text + return StreamFallbackDecision(action="hold") + + def on_finish(self, context: OutputFallbackContext, state: dict) -> StreamFallbackDecision: + return StreamFallbackDecision(action="flush", text=state.get("held", "")) + + +@OutputFallbackManager.register("test-hold-drop", force=True) +class HoldDropStrategy(OutputFallbackStrategy): + name = "test-hold-drop" + + def should_apply(self, text: str, context: OutputFallbackContext) -> bool: + return True + + def apply(self, text: str, context: OutputFallbackContext) -> str: + return text + + def on_delta(self, delta_text: str, context: OutputFallbackContext, state: dict) -> StreamFallbackDecision: + state["held"] = state.get("held", "") + delta_text + return StreamFallbackDecision(action="hold") + + def on_finish(self, context: OutputFallbackContext, state: dict) -> StreamFallbackDecision: + return StreamFallbackDecision(action="flush", text="") + + +@OutputFallbackManager.register("test-truncate", force=True) +class TruncateStrategy(OutputFallbackStrategy): + name = "test-truncate" + + def __init__(self, config: dict | None = None): + super().__init__(config) + self.trigger = self.config.get("trigger", "truncate") + + def should_apply(self, text: str, context: OutputFallbackContext) -> bool: + return True + + def apply(self, text: str, context: OutputFallbackContext) -> str: + return text + + def on_delta(self, delta_text: str, context: OutputFallbackContext, state: dict) -> StreamFallbackDecision: + state["seen"] = delta_text + if self.trigger in delta_text: + return StreamFallbackDecision(action="truncate", text=delta_text) + return StreamFallbackDecision(action="send", text=delta_text) + + +@OutputFallbackManager.register("test-text-observer", force=True) +class TextObserverStrategy(OutputFallbackStrategy): + name = "test-text-observer" + + def should_apply(self, text: str, context: OutputFallbackContext) -> bool: + return True + + def apply(self, text: str, context: OutputFallbackContext) -> str: + return text + + def on_delta(self, delta_text: str, context: OutputFallbackContext, state: dict) -> StreamFallbackDecision: + state["seen"] = delta_text + return StreamFallbackDecision(action="send", text=delta_text) + + +@OutputFallbackManager.register("test-mutable-state", force=True) +class MutableStateStrategy(OutputFallbackStrategy): + name = "test-mutable-state" + + def should_apply(self, text: str, context: OutputFallbackContext) -> bool: + return True + + def apply(self, text: str, context: OutputFallbackContext) -> str: + return text + + def on_delta(self, delta_text: str, context: OutputFallbackContext, state: dict) -> StreamFallbackDecision: + state.setdefault("items", []).append(delta_text) + return StreamFallbackDecision(action="send", text=delta_text) + + +@OutputFallbackManager.register("test-raising-state", force=True) +class RaisingStateStrategy(OutputFallbackStrategy): + name = "test-raising-state" + + def should_apply(self, text: str, context: OutputFallbackContext) -> bool: + return True + + def apply(self, text: str, context: OutputFallbackContext) -> str: + return text + + def on_delta(self, delta_text: str, context: OutputFallbackContext, state: dict) -> StreamFallbackDecision: + state["phase"] = "partial" + raise RuntimeError("boom") + + +@OutputFallbackManager.register("test-token-observer", force=True) +class TokenObserverStrategy(OutputFallbackStrategy): + name = "test-token-observer" + + def should_apply(self, text: str, context: OutputFallbackContext) -> bool: + return True + + def apply(self, text: str, context: OutputFallbackContext) -> str: + return text + + def on_delta(self, delta_text: str, context: OutputFallbackContext, state: dict) -> StreamFallbackDecision: + state["token_count"] = state.get("token_count", 0) + len(context.output.get("token_ids") or []) + return StreamFallbackDecision(action="send", text=delta_text) + + +def make_context(text: str, stream: bool = False, token_ids: list[int] | None = None) -> OutputFallbackContext: + output = {"text": text} + if token_ids is not None: + output["token_ids"] = token_ids + return OutputFallbackContext( + request=None, + request_id="test-request::n::0", + stream=stream, + output=output, + full_text=None if stream else text, + delta_text=text if stream else None, + ) + + +class TestOutputFallbackStrategy: + def test_default_on_delta_applies_full_text_strategy(self): + strategy = ReplaceStrategy() + decision = strategy.on_delta("bad output", make_context("bad output", stream=True), {}) + assert decision.action == "send" + assert decision.text == "good output" + + def test_default_on_finish_flushes_nothing(self): + strategy = ReplaceStrategy() + decision = strategy.on_finish(make_context("", stream=True), {}) + assert decision.action == "flush" + assert decision.text == "" + + +class TestOutputFallbackManager: + def test_apply_runs_enabled_strategies_in_order(self): + manager = OutputFallbackManager(strategies=["test-replace", "test-suffix"]) + assert manager.apply("bad output", make_context("bad output")) == "good output-suffix" + + def test_on_delta_send(self): + manager = OutputFallbackManager(strategies=["test-suffix"]) + decision = manager.on_delta("request-1", "hello", make_context("hello", stream=True)) + assert decision.action == "send" + assert decision.text == "hello-suffix" + + def test_runs_later_strategy_after_hold(self): + manager = OutputFallbackManager(strategies=["test-hold", "test-token-observer"]) + decision = manager.on_delta("request-1", "hello", make_context("hello", stream=True, token_ids=[1, 2])) + assert decision.action == "hold" + assert decision.text == "" + assert "test-token-observer" not in manager.states.get("request-1", {}) + + def test_later_strategy_observes_text_when_hold_returns_empty_text(self): + manager = OutputFallbackManager(strategies=["test-hold", "test-text-observer"]) + decision = manager.on_delta("request-1", "hello", make_context("hello", stream=True)) + assert decision.action == "hold" + assert decision.text == "" + assert "test-text-observer" not in manager.states.get("request-1", {}) + + def test_hold_does_not_commit_mutable_trial_state(self): + manager = OutputFallbackManager(strategies=["test-mutable-state", "test-hold"]) + manager._get_state("request-1", "test-mutable-state")["items"] = [] + decision = manager.on_delta("request-1", "hello", make_context("hello", stream=True)) + assert decision.action == "hold" + assert manager.states["request-1"]["test-mutable-state"]["items"] == [] + + def test_delta_exception_does_not_commit_partial_trial_state(self): + manager = OutputFallbackManager(strategies=["test-raising-state", "test-suffix"]) + manager._get_state("request-1", "test-raising-state")["phase"] = "original" + decision = manager.on_delta("request-1", "hello", make_context("hello", stream=True)) + assert decision.action == "send" + assert decision.text == "hello-suffix" + assert manager.states["request-1"]["test-raising-state"]["phase"] == "original" + + def test_hold_is_preserved_after_later_send_strategy(self): + manager = OutputFallbackManager(strategies=["test-hold", "test-suffix"]) + decision = manager.on_delta("request-1", "hello", make_context("hello", stream=True)) + assert decision.action == "hold" + assert decision.text == "" + + def test_truncate_continues_later_strategy(self): + manager = OutputFallbackManager( + strategies=["test-truncate", "test-suffix"], + config={"test-truncate": {"trigger": "stop"}}, + ) + decision = manager.on_delta("request-1", "please stop", make_context("please stop", stream=True)) + assert decision.action == "truncate" + assert decision.text == "please stop-suffix" + + def test_truncate_keeps_text_after_later_hold(self): + manager = OutputFallbackManager( + strategies=["test-truncate", "test-hold"], + config={"test-truncate": {"trigger": "stop"}}, + ) + decision = manager.on_delta("request-1", "please stop", make_context("please stop", stream=True)) + assert decision.action == "truncate" + assert decision.text == "please stop" + + def test_finish_flushes_pending_buffer_through_pipeline(self): + manager = OutputFallbackManager(strategies=["test-hold", "test-suffix", "test-token-observer"]) + manager.on_delta("request-1", "held", make_context("held", stream=True, token_ids=[7])) + finish_decision = manager.on_finish("request-1", make_context("", stream=True, token_ids=[8])) + assert finish_decision.action == "flush" + assert finish_decision.text == "held-suffix" + assert manager.states["request-1"]["test-token-observer"]["token_count"] == 1 + + def test_finish_honors_empty_flush_after_hold(self): + manager = OutputFallbackManager(strategies=["test-hold-drop"]) + manager.on_delta("request-1", "held", make_context("held", stream=True)) + finish_decision = manager.on_finish("request-1", make_context("", stream=True)) + assert finish_decision.action == "flush" + assert finish_decision.text == "" + + def test_normalizes_strategy_and_config_names(self): + manager = OutputFallbackManager( + strategies=["test_truncate"], + config={"test_truncate": {"trigger": "stop"}}, + ) + decision = manager.on_delta("request-1", "please stop", make_context("please stop", stream=True)) + assert decision.action == "truncate" + assert decision.text == "please stop" + + def test_cleanup(self): + manager = OutputFallbackManager(strategies=["test-suffix"]) + manager._get_state("request-1", "test-suffix")["cache"] = "held" + manager._get_state("request-2", "test-suffix")["cache"] = "other" + manager.cleanup("request-1") + assert "request-1" not in manager.states + assert manager.states["request-2"]["test-suffix"]["cache"] == "other" + + +@OutputFallbackManager.register("test-raising-apply", force=True) +class RaisingApplyStrategy(OutputFallbackStrategy): + name = "test-raising-apply" + + def should_apply(self, text: str, context: OutputFallbackContext) -> bool: + return True + + def apply(self, text: str, context: OutputFallbackContext) -> str: + raise RuntimeError("apply boom") + + +@OutputFallbackManager.register("test-raising-finish", force=True) +class RaisingFinishStrategy(OutputFallbackStrategy): + name = "test-raising-finish" + + def should_apply(self, text: str, context: OutputFallbackContext) -> bool: + return True + + def apply(self, text: str, context: OutputFallbackContext) -> str: + return text + + def on_delta(self, delta_text: str, context: OutputFallbackContext, state: dict) -> StreamFallbackDecision: + return StreamFallbackDecision(action="hold") + + def on_finish(self, context: OutputFallbackContext, state: dict) -> StreamFallbackDecision: + raise RuntimeError("finish boom") + + +class TestGetStrategyNotFound: + def test_raises_key_error(self): + import pytest + + with pytest.raises(KeyError, match="not found"): + OutputFallbackManager.get_strategy("nonexistent-strategy-xyz") + + +class TestRegisterValidation: + def test_force_not_bool_raises(self): + import pytest + + with pytest.raises(TypeError, match="force must be a boolean"): + OutputFallbackManager.register(name="x", force="yes") + + def test_invalid_name_type_raises(self): + import pytest + + with pytest.raises(TypeError, match="name must be None"): + OutputFallbackManager.register(name=123) + + def test_register_with_module_param(self): + result = OutputFallbackManager.register(name="test-direct-reg", module=ReplaceStrategy) + assert result is ReplaceStrategy + assert OutputFallbackManager.get_strategy("test-direct-reg") is ReplaceStrategy + + def test_not_subclass_raises(self): + import pytest + + with pytest.raises(TypeError, match="must be subclass"): + OutputFallbackManager._register_strategy(module=str, strategy_name="bad") + + def test_no_force_duplicate_raises(self): + import pytest + + OutputFallbackManager._register_strategy(module=ReplaceStrategy, strategy_name="dup-test", force=True) + with pytest.raises(KeyError, match="already registered"): + OutputFallbackManager._register_strategy(module=SuffixStrategy, strategy_name="dup-test", force=False) + + def test_strategy_name_from_module(self): + class MyStrat(OutputFallbackStrategy): + name = "auto-name-test" + + def should_apply(self, text, context): + return False + + def apply(self, text, context): + return text + + OutputFallbackManager._register_strategy(module=MyStrat, strategy_name=None, force=True) + assert OutputFallbackManager.get_strategy("auto-name-test") is MyStrat + + +class TestApplyException: + def test_apply_exception_is_swallowed(self): + manager = OutputFallbackManager(strategies=["test-raising-apply"]) + result = manager.apply("hello", make_context("hello")) + assert result == "hello" + + +class TestOnFinishNoBufBranch: + def test_on_finish_no_buffer_collects_strategy_finish(self): + manager = OutputFallbackManager(strategies=["test-hold"]) + manager._get_state("r1", "test-hold")["held"] = "buffered" + decision = manager.on_finish("r1", make_context("", stream=True)) + assert decision.action == "flush" + assert decision.text == "buffered" + + def test_on_finish_no_buffer_exception_swallowed(self): + manager = OutputFallbackManager(strategies=["test-raising-finish"]) + decision = manager.on_finish("r1", make_context("", stream=True)) + assert decision.action == "flush" + assert decision.text == "" + + +class TestOnFinishBufferExceptions: + def test_on_finish_buffer_on_delta_exception(self): + manager = OutputFallbackManager(strategies=["test-raising-state", "test-suffix"]) + manager._buffers["r1"] = "hello" + decision = manager.on_finish("r1", make_context("", stream=True)) + assert decision.action == "flush" + assert decision.text == "hello-suffix" + + def test_on_finish_buffer_hold_then_finish_exception(self): + manager = OutputFallbackManager(strategies=["test-raising-finish"]) + manager._buffers["r1"] = "data" + decision = manager.on_finish("r1", make_context("", stream=True)) + assert decision.action == "flush" + assert decision.text == "data" + + +class TestOutputFallbackPlugin: + def test_import(self, tmp_path: Path): + plugin_path = tmp_path / "custom_fallback.py" + plugin_path.write_text( + "from fastdeploy.output.fallback import OutputFallbackManager, OutputFallbackStrategy\n" + "@OutputFallbackManager.register('custom-plugin-test')\n" + "class CustomPluginFallback(OutputFallbackStrategy):\n" + " name = 'custom-plugin-test'\n" + " def should_apply(self, text, context):\n" + " return 'bad' in text\n" + " def apply(self, text, context):\n" + " return text.replace('bad', 'good')\n" + ) + OutputFallbackManager.import_fallback_plugin(str(plugin_path)) + manager = OutputFallbackManager(strategies=["custom-plugin-test"]) + result = manager.apply("bad output", make_context("bad output")) + assert result == "good output" + + def test_import_invalid_path_swallowed(self): + OutputFallbackManager.import_fallback_plugin("/nonexistent/path/plugin.py")