Skip to content

vllm.reasoning.abs_reasoning_parsers

logger module-attribute 

logger = init_logger(__name__)

ReasoningParser

Abstract reasoning parser class that should not be used directly. Provided and methods should be used in derived classes.

It is used to extract reasoning content from the model output.

Source code in vllm/reasoning/abs_reasoning_parsers.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class ReasoningParser:
    """
    Abstract reasoning parser class that should not be used directly.
    Provided and methods should be used in derived classes.

    It is used to extract reasoning content from the model output.
    """

    def __init__(self, tokenizer: AnyTokenizer):
        self.model_tokenizer = tokenizer

    @cached_property
    def vocab(self) -> dict[str, int]:
        # NOTE: Only PreTrainedTokenizerFast is guaranteed to have .vocab
        # whereas all tokenizers have .get_vocab()
        return self.model_tokenizer.get_vocab()

    @abstractmethod
    def is_reasoning_end(self, input_ids: Sequence[int]) -> bool:
        """
        Check if the reasoning content ends in the input_ids.

        It is used in structured engines like `xgrammar` to check if the
        reasoning content ends in the model output.

        Parameters:
        input_ids: list[int]
            The input_ids of the model output.

        Returns:
        bool
            True if the reasoning content ends in the input_ids.
        """

    @abstractmethod
    def extract_content_ids(self, input_ids: list[int]) -> list[int]:
        """
        Extract content token ids from the input_ids.
        Parameters:
        input_ids: list[int]
            The input_ids of the model output.
        Returns:
        list[int]
            The extracted content from the input_ids.
        """

    @abstractmethod
    def extract_reasoning_content(
            self, model_output: str, request: ChatCompletionRequest
    ) -> tuple[Optional[str], Optional[str]]:
        """
        Extract reasoning content from a complete model-generated string.

        Used for non-streaming responses where we have the entire model response
        available before sending to the client.

        Parameters:
        model_output: str
            The model-generated string to extract reasoning content from.

        request: ChatCompletionRequest
            The request object that was used to generate the model_output.

        Returns:
        tuple[Optional[str], Optional[str]]
            A tuple containing the reasoning content and the content.
        """

    @abstractmethod
    def extract_reasoning_content_streaming(
        self,
        previous_text: str,
        current_text: str,
        delta_text: str,
        previous_token_ids: Sequence[int],
        current_token_ids: Sequence[int],
        delta_token_ids: Sequence[int],
    ) -> Union[DeltaMessage, None]:
        """
        Instance method that should be implemented for extracting reasoning
        from an incomplete response; for use when handling reasoning calls and
        streaming. Has to be an instance method because  it requires state -
        the current tokens/diffs, but also the information about what has
        previously been parsed and extracted (see constructor)
        """

model_tokenizer instance-attribute 

model_tokenizer = tokenizer

vocab cached property 

vocab: dict[str, int]

__init__ 

__init__(tokenizer: AnyTokenizer)
Source code in vllm/reasoning/abs_reasoning_parsers.py 
29
30
def __init__(self, tokenizer: AnyTokenizer):
    self.model_tokenizer = tokenizer

extract_content_ids abstractmethod 

extract_content_ids(input_ids: list[int]) -> list[int]

Extract content token ids from the input_ids. Parameters: input_ids: list[int] The input_ids of the model output. Returns: list[int] The extracted content from the input_ids.

Source code in vllm/reasoning/abs_reasoning_parsers.py 
55
56
57
58
59
60
61
62
63
64
65
@abstractmethod
def extract_content_ids(self, input_ids: list[int]) -> list[int]:
    """
    Extract content token ids from the input_ids.
    Parameters:
    input_ids: list[int]
        The input_ids of the model output.
    Returns:
    list[int]
        The extracted content from the input_ids.
    """

extract_reasoning_content abstractmethod 

extract_reasoning_content(
    model_output: str, request: ChatCompletionRequest
) -> tuple[Optional[str], Optional[str]]

Extract reasoning content from a complete model-generated string.

Used for non-streaming responses where we have the entire model response available before sending to the client.

model_output: str The model-generated string to extract reasoning content from.

ChatCompletionRequest

The request object that was used to generate the model_output.

tuple[Optional[str], Optional[str]] A tuple containing the reasoning content and the content.

Source code in vllm/reasoning/abs_reasoning_parsers.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@abstractmethod
def extract_reasoning_content(
        self, model_output: str, request: ChatCompletionRequest
) -> tuple[Optional[str], Optional[str]]:
    """
    Extract reasoning content from a complete model-generated string.

    Used for non-streaming responses where we have the entire model response
    available before sending to the client.

    Parameters:
    model_output: str
        The model-generated string to extract reasoning content from.

    request: ChatCompletionRequest
        The request object that was used to generate the model_output.

    Returns:
    tuple[Optional[str], Optional[str]]
        A tuple containing the reasoning content and the content.
    """

extract_reasoning_content_streaming abstractmethod 

extract_reasoning_content_streaming(
    previous_text: str,
    current_text: str,
    delta_text: str,
    previous_token_ids: Sequence[int],
    current_token_ids: Sequence[int],
    delta_token_ids: Sequence[int],
) -> Union[DeltaMessage, None]

Instance method that should be implemented for extracting reasoning from an incomplete response; for use when handling reasoning calls and streaming. Has to be an instance method because it requires state - the current tokens/diffs, but also the information about what has previously been parsed and extracted (see constructor)

Source code in vllm/reasoning/abs_reasoning_parsers.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@abstractmethod
def extract_reasoning_content_streaming(
    self,
    previous_text: str,
    current_text: str,
    delta_text: str,
    previous_token_ids: Sequence[int],
    current_token_ids: Sequence[int],
    delta_token_ids: Sequence[int],
) -> Union[DeltaMessage, None]:
    """
    Instance method that should be implemented for extracting reasoning
    from an incomplete response; for use when handling reasoning calls and
    streaming. Has to be an instance method because  it requires state -
    the current tokens/diffs, but also the information about what has
    previously been parsed and extracted (see constructor)
    """

is_reasoning_end abstractmethod 

is_reasoning_end(input_ids: Sequence[int]) -> bool

Check if the reasoning content ends in the input_ids.

It is used in structured engines like xgrammar to check if the reasoning content ends in the model output.

input_ids: list[int] The input_ids of the model output.

bool True if the reasoning content ends in the input_ids.

Source code in vllm/reasoning/abs_reasoning_parsers.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@abstractmethod
def is_reasoning_end(self, input_ids: Sequence[int]) -> bool:
    """
    Check if the reasoning content ends in the input_ids.

    It is used in structured engines like `xgrammar` to check if the
    reasoning content ends in the model output.

    Parameters:
    input_ids: list[int]
        The input_ids of the model output.

    Returns:
    bool
        True if the reasoning content ends in the input_ids.
    """

ReasoningParserManager

Source code in vllm/reasoning/abs_reasoning_parsers.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
class ReasoningParserManager:
    reasoning_parsers: dict[str, type] = {}

    @classmethod
    def get_reasoning_parser(cls, name: str | None) -> type[ReasoningParser]:
        """
        Get reasoning parser by name which is registered by `register_module`.

        Raise a KeyError exception if the name is not registered.
        """
        if name in cls.reasoning_parsers:
            return cls.reasoning_parsers[name]

        raise KeyError(
            f"reasoning helper: '{name}' not found in reasoning_parsers")

    @classmethod
    def _register_module(
        cls,
        module: type,
        module_name: Optional[Union[str, list[str]]] = None,
        force: bool = True,
    ) -> None:
        if not issubclass(module, ReasoningParser):
            raise TypeError("module must be subclass of ReasoningParser, "
                            f"but got {type(module)}")
        if module_name is None:
            module_name = module.__name__
        if isinstance(module_name, str):
            module_name = [module_name]
        for name in module_name:
            if not force and name in cls.reasoning_parsers:
                existed_module = cls.reasoning_parsers[name]
                raise KeyError(f"{name} is already registered "
                               f"at {existed_module.__module__}")
            cls.reasoning_parsers[name] = module

    @classmethod
    def register_module(
        cls,
        name: Optional[Union[str, list[str]]] = None,
        force: bool = True,
        module: Union[type, None] = None,
    ) -> Union[type, Callable]:
        """
        Register module with the given name or name list. it can be used as a
        decoder(with module as None) or normal function(with module as not
        None).
        """
        if not isinstance(force, bool):
            raise TypeError(f"force must be a boolean, but got {type(force)}")

        # raise the error ahead of time
        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)}")

        # use it as a normal method: x.register_module(module=SomeClass)
        if module is not None:
            cls._register_module(module=module, module_name=name, force=force)
            return module

        # use it as a decorator: @x.register_module()
        def _register(module):
            cls._register_module(module=module, module_name=name, force=force)
            return module

        return _register

    @classmethod
    def import_reasoning_parser(cls, plugin_path: str) -> None:
        """
        Import a user-defined reasoning parser by the path
        of the reasoning parser define file.
        """
        module_name = os.path.splitext(os.path.basename(plugin_path))[0]

        try:
            import_from_path(module_name, plugin_path)
        except Exception:
            logger.exception("Failed to load module '%s' from %s.",
                             module_name, plugin_path)
            return

reasoning_parsers class-attribute instance-attribute 

reasoning_parsers: dict[str, type] = {}

_register_module classmethod 

_register_module(
    module: type,
    module_name: Optional[Union[str, list[str]]] = None,
    force: bool = True,
) -> None
Source code in vllm/reasoning/abs_reasoning_parsers.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@classmethod
def _register_module(
    cls,
    module: type,
    module_name: Optional[Union[str, list[str]]] = None,
    force: bool = True,
) -> None:
    if not issubclass(module, ReasoningParser):
        raise TypeError("module must be subclass of ReasoningParser, "
                        f"but got {type(module)}")
    if module_name is None:
        module_name = module.__name__
    if isinstance(module_name, str):
        module_name = [module_name]
    for name in module_name:
        if not force and name in cls.reasoning_parsers:
            existed_module = cls.reasoning_parsers[name]
            raise KeyError(f"{name} is already registered "
                           f"at {existed_module.__module__}")
        cls.reasoning_parsers[name] = module

get_reasoning_parser classmethod 

get_reasoning_parser(
    name: str | None,
) -> type[ReasoningParser]

Get reasoning parser by name which is registered by register_module.

Raise a KeyError exception if the name is not registered.

Source code in vllm/reasoning/abs_reasoning_parsers.py 
111
112
113
114
115
116
117
118
119
120
121
122
@classmethod
def get_reasoning_parser(cls, name: str | None) -> type[ReasoningParser]:
    """
    Get reasoning parser by name which is registered by `register_module`.

    Raise a KeyError exception if the name is not registered.
    """
    if name in cls.reasoning_parsers:
        return cls.reasoning_parsers[name]

    raise KeyError(
        f"reasoning helper: '{name}' not found in reasoning_parsers")

import_reasoning_parser classmethod 

import_reasoning_parser(plugin_path: str) -> None

Import a user-defined reasoning parser by the path of the reasoning parser define file.

Source code in vllm/reasoning/abs_reasoning_parsers.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
@classmethod
def import_reasoning_parser(cls, plugin_path: str) -> None:
    """
    Import a user-defined reasoning parser by the path
    of the reasoning parser define file.
    """
    module_name = os.path.splitext(os.path.basename(plugin_path))[0]

    try:
        import_from_path(module_name, plugin_path)
    except Exception:
        logger.exception("Failed to load module '%s' from %s.",
                         module_name, plugin_path)
        return

register_module classmethod 

register_module(
    name: Optional[Union[str, list[str]]] = None,
    force: bool = True,
    module: Union[type, None] = None,
) -> Union[type, Callable]

Register module with the given name or name list. it can be used as a decoder(with module as None) or normal function(with module as not None).

Source code in vllm/reasoning/abs_reasoning_parsers.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
@classmethod
def register_module(
    cls,
    name: Optional[Union[str, list[str]]] = None,
    force: bool = True,
    module: Union[type, None] = None,
) -> Union[type, Callable]:
    """
    Register module with the given name or name list. it can be used as a
    decoder(with module as None) or normal function(with module as not
    None).
    """
    if not isinstance(force, bool):
        raise TypeError(f"force must be a boolean, but got {type(force)}")

    # raise the error ahead of time
    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)}")

    # use it as a normal method: x.register_module(module=SomeClass)
    if module is not None:
        cls._register_module(module=module, module_name=name, force=force)
        return module

    # use it as a decorator: @x.register_module()
    def _register(module):
        cls._register_module(module=module, module_name=name, force=force)
        return module

    return _register