chat_utils.py 46.8 KB
Newer Older
1
# SPDX-License-Identifier: Apache-2.0
2
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3

4
import asyncio
5
import json
6
from abc import ABC, abstractmethod
7
from collections import Counter, defaultdict, deque
8
from collections.abc import Awaitable, Iterable
9
from functools import cached_property, lru_cache, partial
10
from pathlib import Path
11
12
from typing import (Any, Callable, Generic, Literal, Optional, TypeVar, Union,
                    cast)
13

14
15
import jinja2.nodes
import transformers.utils.chat_template_utils as hf_chat_utils
16
17
# yapf conflicts with isort for this block
# yapf: disable
18
from openai.types.chat import (ChatCompletionAssistantMessageParam,
19
20
                               ChatCompletionContentPartImageParam,
                               ChatCompletionContentPartInputAudioParam)
21
22
from openai.types.chat import (
    ChatCompletionContentPartParam as OpenAIChatCompletionContentPartParam)
23
24
from openai.types.chat import (ChatCompletionContentPartRefusalParam,
                               ChatCompletionContentPartTextParam)
25
26
from openai.types.chat import (
    ChatCompletionMessageParam as OpenAIChatCompletionMessageParam)
27
28
from openai.types.chat import (ChatCompletionMessageToolCallParam,
                               ChatCompletionToolMessageParam)
29
30
from openai.types.chat.chat_completion_content_part_input_audio_param import (
    InputAudio)
31
from openai.types.responses import ResponseInputImageParam
32
from openai_harmony import Message as OpenAIHarmonyMessage
33
34
from PIL import Image
from pydantic import BaseModel, ConfigDict, TypeAdapter
35
# yapf: enable
36
37
from transformers import (PreTrainedTokenizer, PreTrainedTokenizerFast,
                          ProcessorMixin)
38
# pydantic needs the TypedDict from typing_extensions
39
from typing_extensions import Required, TypeAlias, TypedDict
40

41
from vllm.config import ModelConfig
42
from vllm.logger import init_logger
43
from vllm.model_executor.models import SupportsMultiModal
44
from vllm.multimodal import MULTIMODAL_REGISTRY, MultiModalDataDict
45
from vllm.multimodal.utils import MediaConnector
46
47
48
49
# yapf: disable
from vllm.transformers_utils.chat_templates import (
    get_chat_template_fallback_path)
# yapf: enable
50
from vllm.transformers_utils.processor import cached_get_processor
51
from vllm.transformers_utils.tokenizer import AnyTokenizer, MistralTokenizer
52
from vllm.utils import random_uuid
53
54
55

logger = init_logger(__name__)

56
57
58
59
60
61
MODALITY_PLACEHOLDERS_MAP = {
    "image": "<##IMAGE##>",
    "audio": "<##AUDIO##>",
    "video": "<##VIDEO##>",
}

62

63
64
65
66
67
68
69
70
71
72
73
74
75
76
class AudioURL(TypedDict, total=False):
    url: Required[str]
    """
    Either a URL of the audio or a data URL with base64 encoded audio data.
    """


class ChatCompletionContentPartAudioParam(TypedDict, total=False):
    audio_url: Required[AudioURL]

    type: Required[Literal["audio_url"]]
    """The type of the content part."""


77
78
79
80
81
82
83
84
85
86
87
class ChatCompletionContentPartImageEmbedsParam(TypedDict, total=False):
    image_embeds: Required[Union[str, dict[str, str]]]
    """
    The image embeddings. It can be either:
    - A single base64 string.
    - A dictionary where each value is a base64 string.
    """
    type: Required[Literal["image_embeds"]]
    """The type of the content part."""


88
89
90
91
92
93
94
95
96
97
98
99
100
101
class VideoURL(TypedDict, total=False):
    url: Required[str]
    """
    Either a URL of the video or a data URL with base64 encoded video data.
    """


class ChatCompletionContentPartVideoParam(TypedDict, total=False):
    video_url: Required[VideoURL]

    type: Required[Literal["video_url"]]
    """The type of the content part."""


102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class PILImage(BaseModel):
    """
    A PIL.Image.Image object.
    """
    image_pil: Image.Image
    model_config = ConfigDict(arbitrary_types_allowed=True)


class CustomChatCompletionContentPILImageParam(TypedDict, total=False):
    """A simpler version of the param that only accepts a PIL image.

    Example:
    {
        "image_pil": ImageAsset('cherry_blossom').pil_image
    }
    """
    image_pil: Required[PILImage]


121
122
123
class CustomChatCompletionContentSimpleImageParam(TypedDict, total=False):
    """A simpler version of the param that only accepts a plain image_url.
    This is supported by OpenAI API, although it is not documented.
124

125
126
127
128
129
130
131
132
133
134
    Example:
    {
        "image_url": "https://example.com/image.jpg"
    }
    """
    image_url: Required[str]


class CustomChatCompletionContentSimpleAudioParam(TypedDict, total=False):
    """A simpler version of the param that only accepts a plain audio_url.
135

136
137
138
139
140
141
142
143
    Example:
    {
        "audio_url": "https://example.com/audio.mp3"
    }
    """
    audio_url: Required[str]


144
145
146
147
148
149
150
151
152
153
154
class CustomChatCompletionContentSimpleVideoParam(TypedDict, total=False):
    """A simpler version of the param that only accepts a plain audio_url.

    Example:
    {
        "video_url": "https://example.com/video.mp4"
    }
    """
    video_url: Required[str]


Julien Denize's avatar
Julien Denize committed
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
class CustomThinkCompletionContentParam(TypedDict, total=False):
    """A Think Completion Content Param that accepts a plain text and a boolean.

    Example:
    {
        "thinking": "I am thinking about the answer",
        "closed": True,
        "type": "thinking"
    }
    """

    thinking: Required[str]
    """The thinking content."""

    closed: bool
    """Whether the thinking is closed."""

    type: Required[Literal["thinking"]]
    """The thinking type."""


176
177
ChatCompletionContentPartParam: TypeAlias = Union[
    OpenAIChatCompletionContentPartParam, ChatCompletionContentPartAudioParam,
178
    ChatCompletionContentPartInputAudioParam,
179
    ChatCompletionContentPartVideoParam, ChatCompletionContentPartRefusalParam,
180
    CustomChatCompletionContentPILImageParam,
181
    CustomChatCompletionContentSimpleImageParam,
182
    ChatCompletionContentPartImageEmbedsParam,
183
    CustomChatCompletionContentSimpleAudioParam,
Julien Denize's avatar
Julien Denize committed
184
185
    CustomChatCompletionContentSimpleVideoParam, str,
    CustomThinkCompletionContentParam]
186
187
188
189
190
191
192


class CustomChatCompletionMessageParam(TypedDict, total=False):
    """Enables custom roles in the Chat Completion API."""
    role: Required[str]
    """The role of the message's author."""

193
    content: Union[str, list[ChatCompletionContentPartParam]]
194
195
196
197
198
199
200
201
202
    """The contents of the message."""

    name: str
    """An optional name for the participant.

    Provides the model information to differentiate between participants of the
    same role.
    """

203
204
205
206
207
208
    tool_call_id: Optional[str]
    """Tool call that this message is responding to."""

    tool_calls: Optional[Iterable[ChatCompletionMessageToolCallParam]]
    """The tool calls generated by the model, such as function calls."""

209
210

ChatCompletionMessageParam = Union[OpenAIChatCompletionMessageParam,
211
212
                                   CustomChatCompletionMessageParam,
                                   OpenAIHarmonyMessage]
213
214


215
# TODO: Make fields ReadOnly once mypy supports it
216
217
218
219
class ConversationMessage(TypedDict, total=False):
    role: Required[str]
    """The role of the message's author."""

220
    content: Union[Optional[str], list[dict[str, str]]]
221
222
223
224
225
226
227
228
229
230
    """The contents of the message"""

    tool_call_id: Optional[str]
    """Tool call that this message is responding to."""

    name: Optional[str]
    """The name of the function to call"""

    tool_calls: Optional[Iterable[ChatCompletionMessageToolCallParam]]
    """The tool calls generated by the model, such as function calls."""
231
232


233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
# Passed in by user
ChatTemplateContentFormatOption = Literal["auto", "string", "openai"]

# Used internally
_ChatTemplateContentFormat = Literal["string", "openai"]


def _is_var_access(node: jinja2.nodes.Node, varname: str) -> bool:
    if isinstance(node, jinja2.nodes.Name):
        return node.ctx == "load" and node.name == varname

    return False


def _is_attr_access(node: jinja2.nodes.Node, varname: str, key: str) -> bool:
    if isinstance(node, jinja2.nodes.Getitem):
        return (_is_var_access(node.node, varname)
                and isinstance(node.arg, jinja2.nodes.Const)
                and node.arg.value == key)

    if isinstance(node, jinja2.nodes.Getattr):
        return _is_var_access(node.node, varname) and node.attr == key

    return False


def _is_var_or_elems_access(
    node: jinja2.nodes.Node,
    varname: str,
    key: Optional[str] = None,
) -> bool:
    if isinstance(node, jinja2.nodes.Filter):
        return (node.node is not None
                and _is_var_or_elems_access(node.node, varname, key))
    if isinstance(node, jinja2.nodes.Test):
        return _is_var_or_elems_access(node.node, varname, key)

    if (isinstance(node, jinja2.nodes.Getitem)
            and isinstance(node.arg, jinja2.nodes.Slice)):
        return _is_var_or_elems_access(node.node, varname, key)

    # yapf: disable
    return (
        _is_attr_access(node, varname, key) if key
        else _is_var_access(node, varname)
    ) # yapf: enable


def _iter_nodes_assign_var_or_elems(root: jinja2.nodes.Node, varname: str):
    # Global variable that is implicitly defined at the root
    yield root, varname

    # Iterative BFS
    related_varnames = deque([varname])
    while related_varnames:
        related_varname = related_varnames.popleft()

        for assign_ast in root.find_all(jinja2.nodes.Assign):
            lhs = assign_ast.target
            rhs = assign_ast.node

            if _is_var_or_elems_access(rhs, related_varname):
                assert isinstance(lhs, jinja2.nodes.Name)
                yield assign_ast, lhs.name

                # Avoid infinite looping for self-assignment
                if lhs.name != related_varname:
                    related_varnames.append(lhs.name)


# NOTE: The proper way to handle this is to build a CFG so that we can handle
# the scope in which each variable is defined, but that is too complicated
def _iter_nodes_assign_messages_item(root: jinja2.nodes.Node):
    messages_varnames = [
        varname
        for _, varname in _iter_nodes_assign_var_or_elems(root, "messages")
    ]

    # Search for {%- for message in messages -%} loops
    for loop_ast in root.find_all(jinja2.nodes.For):
        loop_iter = loop_ast.iter
        loop_target = loop_ast.target

        for varname in messages_varnames:
            if _is_var_or_elems_access(loop_iter, varname):
                assert isinstance(loop_target, jinja2.nodes.Name)
                yield loop_ast, loop_target.name
                break


def _iter_nodes_assign_content_item(root: jinja2.nodes.Node):
    message_varnames = [
        varname for _, varname in _iter_nodes_assign_messages_item(root)
    ]

    # Search for {%- for content in message['content'] -%} loops
    for loop_ast in root.find_all(jinja2.nodes.For):
        loop_iter = loop_ast.iter
        loop_target = loop_ast.target

        for varname in message_varnames:
            if _is_var_or_elems_access(loop_iter, varname, "content"):
                assert isinstance(loop_target, jinja2.nodes.Name)
                yield loop_ast, loop_target.name
                break


def _try_extract_ast(chat_template: str) -> Optional[jinja2.nodes.Template]:
    try:
        jinja_compiled = hf_chat_utils._compile_jinja_template(chat_template)
        return jinja_compiled.environment.parse(chat_template)
    except Exception:
        logger.exception("Error when compiling Jinja template")
        return None


349
@lru_cache(maxsize=32)
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
def _detect_content_format(
    chat_template: str,
    *,
    default: _ChatTemplateContentFormat,
) -> _ChatTemplateContentFormat:
    jinja_ast = _try_extract_ast(chat_template)
    if jinja_ast is None:
        return default

    try:
        next(_iter_nodes_assign_content_item(jinja_ast))
    except StopIteration:
        return "string"
    except Exception:
        logger.exception("Error when parsing AST of Jinja template")
        return default
    else:
        return "openai"


370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
def resolve_mistral_chat_template(
    chat_template: Optional[str],
    **kwargs: Any,
) -> Optional[str]:
    if chat_template is not None:
        logger.warning_once(
            "'chat_template' cannot be overridden for mistral tokenizer.")
    if "add_generation_prompt" in kwargs:
        logger.warning_once(
            "'add_generation_prompt' is not supported for mistral tokenizer, "
            "so it will be ignored.")
    if "continue_final_message" in kwargs:
        logger.warning_once(
            "'continue_final_message' is not supported for mistral tokenizer, "
            "so it will be ignored.")
    return None

387

388
def resolve_hf_chat_template(
389
390
391
    tokenizer: Union[PreTrainedTokenizer, PreTrainedTokenizerFast],
    chat_template: Optional[str],
    tools: Optional[list[dict[str, Any]]],
392
393
    *,
    model_config: ModelConfig,
394
395
396
397
398
399
400
401
402
403
404
405
) -> Optional[str]:
    # 1st priority: The given chat template
    if chat_template is not None:
        return chat_template

    # 2nd priority: AutoProcessor chat template, unless tool calling is enabled
    if tools is None:
        try:
            processor = cached_get_processor(
                tokenizer.name_or_path,
                processor_cls=(PreTrainedTokenizer, PreTrainedTokenizerFast,
                               ProcessorMixin),
406
                trust_remote_code=model_config.trust_remote_code,
407
408
            )
            if isinstance(processor, ProcessorMixin) and \
409
                hasattr(processor, 'chat_template') and \
410
411
412
                processor.chat_template is not None:
                return processor.chat_template
        except Exception:
413
            logger.debug("Failed to load AutoProcessor chat template for %s", tokenizer.name_or_path, exc_info=True)  # noqa: E501
414
415
416
417
418
419
420
421

    # 3rd priority: AutoTokenizer chat template
    try:
        return tokenizer.get_chat_template(chat_template, tools=tools)
    except Exception:
        logger.debug("Failed to load AutoTokenizer chat template for %s",
                     tokenizer.name_or_path, exc_info=True)

422
423
424
425
426
427
428
429
430
431
432
433
434
435
    # 4th priority: Predefined fallbacks
    path = get_chat_template_fallback_path(
        model_type=model_config.hf_config.model_type,
        tokenizer_name_or_path=model_config.tokenizer,
    )
    if path is not None:
        logger.info("Loading chat template fallback for %s as there isn't one "
                    "defined on HF Hub.", tokenizer.name_or_path)
        chat_template = load_chat_template(path)
    else:
        logger.debug("There is no chat template fallback for %s",
                     tokenizer.name_or_path)

    return chat_template
436
437


438
439
def _resolve_chat_template_content_format(
    chat_template: Optional[str],
440
    tools: Optional[list[dict[str, Any]]],
441
    tokenizer: AnyTokenizer,
442
443
    *,
    model_config: ModelConfig,
444
445
) -> _ChatTemplateContentFormat:
    if isinstance(tokenizer, (PreTrainedTokenizer, PreTrainedTokenizerFast)):
446
        hf_chat_template = resolve_hf_chat_template(
447
448
449
            tokenizer,
            chat_template=chat_template,
            tools=tools,
450
            model_config=model_config,
451
        )
452
    else:
453
454
455
456
        hf_chat_template = None

    jinja_text = (hf_chat_template if isinstance(hf_chat_template, str)
                  else load_chat_template(chat_template, is_literal=True))
457
458
459
460

    detected_format = ("string" if jinja_text is None else
                       _detect_content_format(jinja_text, default="string"))

461
    return detected_format
462
463
464


@lru_cache
465
def _log_chat_template_content_format(
466
467
    chat_template: Optional[str],
    given_format: ChatTemplateContentFormatOption,
468
469
    detected_format: ChatTemplateContentFormatOption,
):
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
    logger.info(
        "Detected the chat template content format to be '%s'. "
        "You can set `--chat-template-content-format` to override this.",
        detected_format,
    )

    if given_format != "auto" and given_format != detected_format:
        logger.warning(
            "You specified `--chat-template-content-format %s` "
            "which is different from the detected format '%s'. "
            "If our automatic detection is incorrect, please consider "
            "opening a GitHub issue so that we can improve it: "
            "https://github.com/vllm-project/vllm/issues/new/choose",
            given_format,
            detected_format,
        )

487
488
489
490
491
492

def resolve_chat_template_content_format(
    chat_template: Optional[str],
    tools: Optional[list[dict[str, Any]]],
    given_format: ChatTemplateContentFormatOption,
    tokenizer: AnyTokenizer,
493
494
    *,
    model_config: ModelConfig,
495
) -> _ChatTemplateContentFormat:
496
497
498
    if given_format != "auto":
        return given_format

499
500
501
502
    detected_format = _resolve_chat_template_content_format(
        chat_template,
        tools,
        tokenizer,
503
        model_config=model_config,
504
505
506
507
508
509
510
511
    )

    _log_chat_template_content_format(
        chat_template,
        given_format=given_format,
        detected_format=detected_format,
    )

512
    return detected_format
513

514
515


516
ModalityStr = Literal["image", "audio", "video", "image_embeds"]
517
518
519
520
_T = TypeVar("_T")


class BaseMultiModalItemTracker(ABC, Generic[_T]):
521
522
523
524
525
526
527
    """
    Tracks multi-modal items in a given request and ensures that the number
    of multi-modal items in a given request does not exceed the configured
    maximum per prompt.
    """

    def __init__(self, model_config: ModelConfig, tokenizer: AnyTokenizer):
528
529
        super().__init__()

530
531
        self._model_config = model_config
        self._tokenizer = tokenizer
532

533
        self._items_by_modality = defaultdict[str, list[_T]](list)
534

535
536
537
538
    @property
    def model_config(self) -> ModelConfig:
        return self._model_config

539
    @cached_property
540
    def model_cls(self) -> type[SupportsMultiModal]:
541
        from vllm.model_executor.model_loader import get_model_cls
542
543
        model_cls = get_model_cls(self.model_config)
        return cast(type[SupportsMultiModal], model_cls)
544

545
546
547
548
    @property
    def allowed_local_media_path(self):
        return self._model_config.allowed_local_media_path

549
550
551
552
    @property
    def mm_registry(self):
        return MULTIMODAL_REGISTRY

553
554
555
556
    @cached_property
    def mm_processor(self):
        return self.mm_registry.create_processor(self.model_config)

557
558
559
560
561
    def add(self, modality: ModalityStr, item: _T) -> Optional[str]:
        """
        Add a multi-modal item to the current prompt and returns the
        placeholder string to use, if any.
        """
562
        input_modality = modality.replace("_embeds", "")
563
        num_items = len(self._items_by_modality[modality]) + 1
564

565
        self.mm_processor.validate_num_items(input_modality, num_items)
566

567
        self._items_by_modality[modality].append(item)
568

569
        return self.model_cls.get_placeholder_str(modality, num_items)
570
571
572
573
574
575

    @abstractmethod
    def create_parser(self) -> "BaseMultiModalContentParser":
        raise NotImplementedError


576
class MultiModalItemTracker(BaseMultiModalItemTracker[object]):
577
578

    def all_mm_data(self) -> Optional[MultiModalDataDict]:
579
580
581
582
583
584
585
586
587
588
589
590
591
592
        if not self._items_by_modality:
            return None
        mm_inputs = {}
        items_by_modality = dict(self._items_by_modality)
        if "image" in items_by_modality and "image_embeds" in items_by_modality:
            raise ValueError(\
                "Mixing raw image and embedding inputs is not allowed")

        if "image_embeds" in items_by_modality:
            image_embeds_lst = items_by_modality["image_embeds"]
            if len(image_embeds_lst) > 1:
                raise ValueError(\
                    "Only one message can have {'type': 'image_embeds'}")
            mm_inputs["image"] = image_embeds_lst[0]
593
        if "image" in items_by_modality:
594
            mm_inputs["image"] = items_by_modality["image"] # A list of images
595
        if "audio" in items_by_modality:
596
            mm_inputs["audio"] = items_by_modality["audio"] # A list of audios
597
        if "video" in items_by_modality:
598
599
            mm_inputs["video"] = items_by_modality["video"] # A list of videos
        return mm_inputs
600
601
602
603
604

    def create_parser(self) -> "BaseMultiModalContentParser":
        return MultiModalContentParser(self)


605
class AsyncMultiModalItemTracker(BaseMultiModalItemTracker[Awaitable[object]]):
606
607

    async def all_mm_data(self) -> Optional[MultiModalDataDict]:
608
609
610
611
        if not self._items_by_modality:
            return None
        mm_inputs = {}
        items_by_modality = {
612
613
614
                modality: await asyncio.gather(*items)
                for modality, items in self._items_by_modality.items()
            }
615

616
617
618
619
620
621
622
623
624
625
        if "image" in items_by_modality and "image_embeds" in items_by_modality:
            raise ValueError(
                "Mixing raw image and embedding inputs is not allowed")

        if "image_embeds" in items_by_modality:
            image_embeds_lst = items_by_modality["image_embeds"]
            if len(image_embeds_lst) > 1:
                raise ValueError(
                    "Only one message can have {'type': 'image_embeds'}")
            mm_inputs["image"] = image_embeds_lst[0]
626
        if "image" in items_by_modality:
627
            mm_inputs["image"] = items_by_modality["image"] # A list of images
628
        if "audio" in items_by_modality:
629
            mm_inputs["audio"] = items_by_modality["audio"] # A list of audios
630
        if "video" in items_by_modality:
631
632
            mm_inputs["video"] = items_by_modality["video"] # A list of videos
        return mm_inputs
633
634
635
636
637
638
639
640
641
642

    def create_parser(self) -> "BaseMultiModalContentParser":
        return AsyncMultiModalContentParser(self)


class BaseMultiModalContentParser(ABC):

    def __init__(self) -> None:
        super().__init__()

643
        # stores model placeholders list with corresponding
644
645
646
647
648
649
650
651
652
653
        # general MM placeholder:
        # {
        #   "<##IMAGE##>": ["<image>", "<image>", "<image>"],
        #   "<##AUDIO##>": ["<audio>", "<audio>"]
        # }
        self._placeholder_storage: dict[str, list] = defaultdict(list)

    def _add_placeholder(self, modality: ModalityStr,
                         placeholder: Optional[str]):
        mod_placeholder = MODALITY_PLACEHOLDERS_MAP[modality]
654
        if placeholder:
655
            self._placeholder_storage[mod_placeholder].append(placeholder)
656

657
658
    def mm_placeholder_storage(self) -> dict[str, list]:
        return dict(self._placeholder_storage)
659
660
661
662
663

    @abstractmethod
    def parse_image(self, image_url: str) -> None:
        raise NotImplementedError

664
665
666
667
668
    @abstractmethod
    def parse_image_embeds(self,
                           image_embeds: Union[str, dict[str, str]]) -> None:
        raise NotImplementedError

669
670
671
672
    @abstractmethod
    def parse_image_pil(self, image_pil: Image.Image) -> None:
        raise NotImplementedError

673
674
675
676
    @abstractmethod
    def parse_audio(self, audio_url: str) -> None:
        raise NotImplementedError

677
    @abstractmethod
678
    def parse_input_audio(self, input_audio: InputAudio) -> None:
679
680
        raise NotImplementedError

681
682
683
684
    @abstractmethod
    def parse_video(self, video_url: str) -> None:
        raise NotImplementedError

685
686
687
688
689
690
691
692

class MultiModalContentParser(BaseMultiModalContentParser):

    def __init__(self, tracker: MultiModalItemTracker) -> None:
        super().__init__()

        self._tracker = tracker

693
        self._connector = MediaConnector(
694
            media_io_kwargs=self._tracker._model_config.media_io_kwargs,
695
696
697
            allowed_local_media_path=tracker.allowed_local_media_path,
        )

698
    def parse_image(self, image_url: str) -> None:
699
        image = self._connector.fetch_image(image_url)
700
701

        placeholder = self._tracker.add("image", image)
702
        self._add_placeholder("image", placeholder)
703

704
705
706
707
708
709
710
711
712
713
714
715
716
    def parse_image_embeds(self,
                           image_embeds: Union[str, dict[str, str]]) -> None:
        if isinstance(image_embeds, dict):
            embeds = {
                k: self._connector.fetch_image_embedding(v)
                for k, v in image_embeds.items()
            }
            placeholder = self._tracker.add("image_embeds", embeds)

        if isinstance(image_embeds, str):
            embedding = self._connector.fetch_image_embedding(image_embeds)
            placeholder = self._tracker.add("image_embeds", embedding)

717
        self._add_placeholder("image", placeholder)
718

719
720
    def parse_image_pil(self, image_pil: Image.Image) -> None:
        placeholder = self._tracker.add("image", image_pil)
721
        self._add_placeholder("image", placeholder)
722

723
    def parse_audio(self, audio_url: str) -> None:
724
        audio = self._connector.fetch_audio(audio_url)
725
726

        placeholder = self._tracker.add("audio", audio)
727
        self._add_placeholder("audio", placeholder)
728

729
730
731
732
    def parse_input_audio(self, input_audio: InputAudio) -> None:
        audio_data = input_audio.get("data", "")
        audio_format = input_audio.get("format", "")
        audio_url = f"data:audio/{audio_format};base64,{audio_data}"
733

734
        return self.parse_audio(audio_url)
735

736
    def parse_video(self, video_url: str) -> None:
737
        video = self._connector.fetch_video(video_url=video_url)
738
739

        placeholder = self._tracker.add("video", video)
740
        self._add_placeholder("video", placeholder)
741

742
743
744
745
746
747
748

class AsyncMultiModalContentParser(BaseMultiModalContentParser):

    def __init__(self, tracker: AsyncMultiModalItemTracker) -> None:
        super().__init__()

        self._tracker = tracker
749
        self._connector = MediaConnector(
750
751
            media_io_kwargs=self._tracker._model_config.media_io_kwargs,
            allowed_local_media_path=tracker.allowed_local_media_path
752
        )
753
754

    def parse_image(self, image_url: str) -> None:
755
        image_coro = self._connector.fetch_image_async(image_url)
756
757

        placeholder = self._tracker.add("image", image_coro)
758
        self._add_placeholder("image", placeholder)
759

760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
    def parse_image_embeds(self,
                           image_embeds: Union[str, dict[str, str]]) -> None:
        future: asyncio.Future[Union[str, dict[str, str]]] = asyncio.Future()

        if isinstance(image_embeds, dict):
            embeds = {
                k: self._connector.fetch_image_embedding(v)
                for k, v in image_embeds.items()
            }
            future.set_result(embeds)

        if isinstance(image_embeds, str):
            embedding = self._connector.\
                fetch_image_embedding(image_embeds)
            future.set_result(embedding)

        placeholder = self._tracker.add("image_embeds", future)
777
        self._add_placeholder("image", placeholder)
778

779
780
781
782
783
    def parse_image_pil(self, image_pil: Image.Image) -> None:
        future: asyncio.Future[Image.Image] = asyncio.Future()
        future.set_result(image_pil)

        placeholder = self._tracker.add("image", future)
784
        self._add_placeholder("image", placeholder)
785

786
    def parse_audio(self, audio_url: str) -> None:
787
        audio_coro = self._connector.fetch_audio_async(audio_url)
788
789

        placeholder = self._tracker.add("audio", audio_coro)
790
        self._add_placeholder("audio", placeholder)
791

792
793
794
795
    def parse_input_audio(self, input_audio: InputAudio) -> None:
        audio_data = input_audio.get("data", "")
        audio_format = input_audio.get("format", "")
        audio_url = f"data:audio/{audio_format};base64,{audio_data}"
796

797
        return self.parse_audio(audio_url)
798

799
    def parse_video(self, video_url: str) -> None:
800
        video = self._connector.fetch_video_async(video_url=video_url)
801
802

        placeholder = self._tracker.add("video", video)
803
        self._add_placeholder("video", placeholder)
804

805

806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
def validate_chat_template(chat_template: Optional[Union[Path, str]]):
    """Raises if the provided chat template appears invalid."""
    if chat_template is None:
        return

    elif isinstance(chat_template, Path) and not chat_template.exists():
        raise FileNotFoundError(
            "the supplied chat template path doesn't exist")

    elif isinstance(chat_template, str):
        JINJA_CHARS = "{}\n"
        if not any(c in chat_template
                   for c in JINJA_CHARS) and not Path(chat_template).exists():
            raise ValueError(
                f"The supplied chat template string ({chat_template}) "
                f"appears path-like, but doesn't exist!")

    else:
        raise TypeError(
            f"{type(chat_template)} is not a valid chat template type")


828
def _load_chat_template(
829
830
831
832
    chat_template: Optional[Union[Path, str]],
    *,
    is_literal: bool = False,
) -> Optional[str]:
833
834
    if chat_template is None:
        return None
835
836
837
838
839
840

    if is_literal:
        if isinstance(chat_template, Path):
            raise TypeError("chat_template is expected to be read directly "
                            "from its value")

841
        return chat_template
842

843
    try:
844
        with open(chat_template) as f:
845
            return f.read()
846
    except OSError as e:
847
848
849
        if isinstance(chat_template, Path):
            raise

850
851
852
853
854
855
        JINJA_CHARS = "{}\n"
        if not any(c in chat_template for c in JINJA_CHARS):
            msg = (f"The supplied chat template ({chat_template}) "
                   f"looks like a file path, but it failed to be "
                   f"opened. Reason: {e}")
            raise ValueError(msg) from e
856

857
858
        # If opening a file fails, set chat template to be args to
        # ensure we decode so our escape are interpreted correctly
859
860
861
862
863
864
865
866
867
868
869
870
        return _load_chat_template(chat_template, is_literal=True)


_cached_load_chat_template = lru_cache(_load_chat_template)


def load_chat_template(
    chat_template: Optional[Union[Path, str]],
    *,
    is_literal: bool = False,
) -> Optional[str]:
    return _cached_load_chat_template(chat_template, is_literal=is_literal)
871
872


873
874
875
876
877
878
879
880
881
def _get_interleaved_text_prompt(placeholder_storage: dict[str, list],
                                 texts: list[str]) -> str:
    for idx, elem in enumerate(texts):
        if elem in placeholder_storage:
            texts[idx] = placeholder_storage[elem].pop(0)

    return "\n".join(texts)


882
# TODO: Let user specify how to insert multimodal tokens into prompt
883
# (similar to chat template)
884
885
886
887
def _get_full_multimodal_text_prompt(placeholder_storage: dict[str, list],
                                     texts: list[str],
                                     interleave_strings: bool
                                     ) -> str:
888
    """Combine multimodal prompts for a multimodal language model."""
889

890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
    # flatten storage to make it looks like
    # {
    #   "<|image|>": 2,
    #   "<|audio|>": 1
    # }
    placeholder_counts = Counter(
        [v for elem in placeholder_storage.values() for v in elem]
    )

    if interleave_strings:
        text_prompt = _get_interleaved_text_prompt(placeholder_storage, texts)
    else:
        text_prompt = "\n".join(texts)

    # Pass interleaved text further in case the user used image placeholders
    # himself, but forgot to disable the 'interleave_strings' flag

907
    # Look through the text prompt to check for missing placeholders
908
    missing_placeholders: list[str] = []
909
910
911
912
913
914
    for placeholder in placeholder_counts:

        # For any existing placeholder in the text prompt, we leave it as is
        placeholder_counts[placeholder] -= text_prompt.count(placeholder)

        if placeholder_counts[placeholder] < 0:
915
916
917
918
919
920
921
            logger.error(
                "Placeholder count is negative! "
                "Ensure that the 'interleave_strings' flag is disabled "
                "(current value: %s) "
                "when manually placing image placeholders.", interleave_strings
            )
            logger.debug("Input prompt: %s", text_prompt)
922
923
924
925
926
927
928
            raise ValueError(
                f"Found more '{placeholder}' placeholders in input prompt than "
                "actual multimodal data items.")

        missing_placeholders.extend([placeholder] *
                                    placeholder_counts[placeholder])

929
930
    # NOTE: Default behaviour: we always add missing placeholders
    # at the front of the prompt, if interleave_strings=False
931
    return "\n".join(missing_placeholders + [text_prompt])
932
933


934
935
# No need to validate using Pydantic again
_TextParser = partial(cast, ChatCompletionContentPartTextParam)
936
_ImageEmbedsParser = partial(cast, ChatCompletionContentPartImageEmbedsParam)
937
_InputAudioParser = partial(cast, ChatCompletionContentPartInputAudioParam)
938
_RefusalParser = partial(cast, ChatCompletionContentPartRefusalParam)
939
_PILImageParser = partial(cast, CustomChatCompletionContentPILImageParam)
Julien Denize's avatar
Julien Denize committed
940
_ThinkParser = partial(cast, CustomThinkCompletionContentParam)
941
942
943
944
# Need to validate url objects
_ImageParser = TypeAdapter(ChatCompletionContentPartImageParam).validate_python
_AudioParser = TypeAdapter(ChatCompletionContentPartAudioParam).validate_python
_VideoParser = TypeAdapter(ChatCompletionContentPartVideoParam).validate_python
945

946
947
_ResponsesInputImageParser = TypeAdapter(
    ResponseInputImageParam).validate_python
948
_ContentPart: TypeAlias = Union[str, dict[str, str], InputAudio, PILImage]
949

950
# Define a mapping from part types to their corresponding parsing functions.
951
MM_PARSER_MAP: dict[
952
953
954
    str,
    Callable[[ChatCompletionContentPartParam], _ContentPart],
] = {
955
    "text":
956
    lambda part: _TextParser(part).get("text", None),
Julien Denize's avatar
Julien Denize committed
957
958
    "thinking":
    lambda part: _ThinkParser(part).get("thinking", None),
959
960
    "input_text":
    lambda part: _TextParser(part).get("text", None),
961
962
    "input_image":
    lambda part: _ResponsesInputImageParser(part).get("image_url", None),
963
    "image_url":
964
    lambda part: _ImageParser(part).get("image_url", {}).get("url", None),
965
    "image_embeds":
966
    lambda part: _ImageEmbedsParser(part).get("image_embeds", None),
967
    "image_pil": lambda part: _PILImageParser(part).get("image_pil", None),
968
    "audio_url":
969
    lambda part: _AudioParser(part).get("audio_url", {}).get("url", None),
970
    "input_audio":
971
    lambda part: _InputAudioParser(part).get("input_audio", None),
972
    "refusal":
973
    lambda part: _RefusalParser(part).get("refusal", None),
974
    "video_url":
975
    lambda part: _VideoParser(part).get("video_url", {}).get("url", None),
976
977
978
979
}


def _parse_chat_message_content_mm_part(
980
        part: ChatCompletionContentPartParam) -> tuple[str, _ContentPart]:
981
    """
982
    Parses a given multi-modal content part based on its type.
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002

    Args:
        part: A dict containing the content part, with a potential 'type' field.

    Returns:
        A tuple (part_type, content) where:
        - part_type: Type of the part (e.g., 'text', 'image_url').
        - content: Parsed content (e.g., text, image URL).

    Raises:
        ValueError: If the 'type' field is missing and no direct URL is found.
    """
    assert isinstance(
        part, dict)  # This is needed to avoid mypy errors: part.get() from str
    part_type = part.get("type", None)

    if isinstance(part_type, str) and part_type in MM_PARSER_MAP:
        content = MM_PARSER_MAP[part_type](part)

        # Special case for 'image_url.detail'
1003
1004
        # We only support 'auto', which is the default
        if part_type == "image_url" and part.get("detail", "auto") != "auto":
1005
1006
1007
1008
1009
1010
            logger.warning("'image_url.detail' is currently not supported "
                           "and will be ignored.")

        return part_type, content

    # Handle missing 'type' but provided direct URL fields.
1011
    # 'type' is required field by pydantic
1012
1013
1014
1015
1016
1017
1018
1019
1020
    if part_type is None:
        if part.get("image_url") is not None:
            image_params = cast(CustomChatCompletionContentSimpleImageParam,
                                part)
            return "image_url", image_params.get("image_url", "")
        if part.get("audio_url") is not None:
            audio_params = cast(CustomChatCompletionContentSimpleAudioParam,
                                part)
            return "audio_url", audio_params.get("audio_url", "")
1021
        if part.get("input_audio") is not None:
1022
            input_audio_params = cast(dict[str, str], part)
1023
            return "input_audio", input_audio_params
1024
1025
1026
1027
        if part.get("video_url") is not None:
            video_params = cast(CustomChatCompletionContentSimpleVideoParam,
                                part)
            return "video_url", video_params.get("video_url", "")
1028
1029
1030
1031
1032
1033
1034
1035
1036
        # Raise an error if no 'type' or direct URL is found.
        raise ValueError("Missing 'type' field in multimodal part.")

    if not isinstance(part_type, str):
        raise ValueError("Invalid 'type' field in multimodal part.")
    return part_type, "unknown part_type content"


VALID_MESSAGE_CONTENT_MM_PART_TYPES = ("text", "refusal", "image_url",
1037
                                       "image_embeds", "image_pil",
1038
                                       "audio_url", "input_audio", "video_url")
1039

1040

1041
1042
1043
def _parse_chat_message_content_parts(
    role: str,
    parts: Iterable[ChatCompletionContentPartParam],
1044
    mm_tracker: BaseMultiModalItemTracker,
1045
1046
    *,
    wrap_dicts: bool,
1047
    interleave_strings: bool,
1048
) -> list[ConversationMessage]:
1049
    content = list[_ContentPart]()
1050

1051
    mm_parser = mm_tracker.create_parser()
1052
1053

    for part in parts:
1054
        parse_res = _parse_chat_message_content_part(
1055
1056
1057
            part,
            mm_parser,
            wrap_dicts=wrap_dicts,
1058
            interleave_strings=interleave_strings
1059
        )
1060
1061
        if parse_res:
            content.append(parse_res)
1062

1063
    if wrap_dicts:
1064
        # Parsing wraps images and texts as interleaved dictionaries
1065
        return [ConversationMessage(role=role,
1066
                                    content=content)]  # type: ignore
1067
    texts = cast(list[str], content)
1068
1069
1070
1071
1072
1073
1074
1075
    mm_placeholder_storage = mm_parser.mm_placeholder_storage()
    if mm_placeholder_storage:
        text_prompt = _get_full_multimodal_text_prompt(mm_placeholder_storage,
                                                       texts,
                                                       interleave_strings)
    else:
        text_prompt = "\n".join(texts)

1076
1077
1078
1079
    return [ConversationMessage(role=role, content=text_prompt)]


def _parse_chat_message_content_part(
1080
1081
1082
1083
    part: ChatCompletionContentPartParam,
    mm_parser: BaseMultiModalContentParser,
    *,
    wrap_dicts: bool,
1084
    interleave_strings: bool,
1085
) -> Optional[_ContentPart]:
1086
1087
1088
1089
1090
1091
1092
1093
    """Parses a single part of a conversation. If wrap_dicts is True,
    structured dictionary pieces for texts and images will be
    wrapped in dictionaries, i.e., {"type": "text", "text", ...} and
    {"type": "image"}, respectively. Otherwise multimodal data will be
    handled by mm_parser, and texts will be returned as strings to be joined
    with multimodal placeholders.
    """
    if isinstance(part, str):  # Handle plain text parts
1094
        return part
1095
1096
    # Handle structured dictionary parts
    part_type, content = _parse_chat_message_content_mm_part(part)
1097
    # if part_type is text/refusal/image_url/audio_url/video_url/input_audio but
1098
1099
    # content is None, log a warning and skip
    if part_type in VALID_MESSAGE_CONTENT_MM_PART_TYPES and content is None:
1100
        logger.warning(
1101
1102
            "Skipping multimodal part '%s' (type: '%s') "
            "with empty / unparsable content.", part, part_type)
1103
1104
        return None

Julien Denize's avatar
Julien Denize committed
1105
    if part_type in ("text", "input_text", "refusal", "thinking"):
1106
1107
1108
1109
1110
        str_content = cast(str, content)
        if wrap_dicts:
            return {'type': 'text', 'text': str_content}
        else:
            return str_content
1111

1112
    modality = None
1113
1114
1115
    if part_type == "image_pil":
        image_content = cast(Image.Image, content)
        mm_parser.parse_image_pil(image_content)
1116
        modality = "image"
1117
    elif part_type in ("image_url", "input_image"):
1118
1119
        str_content = cast(str, content)
        mm_parser.parse_image(str_content)
1120
1121
        modality = "image"
    elif part_type == "image_embeds":
1122
1123
        content = cast(Union[str, dict[str, str]], content)
        mm_parser.parse_image_embeds(content)
1124
1125
        modality = "image"
    elif part_type == "audio_url":
1126
1127
        str_content = cast(str, content)
        mm_parser.parse_audio(str_content)
1128
1129
        modality = "audio"
    elif part_type == "input_audio":
1130
        dict_content = cast(InputAudio, content)
1131
        mm_parser.parse_input_audio(dict_content)
1132
1133
        modality = "audio"
    elif part_type == "video_url":
1134
1135
        str_content = cast(str, content)
        mm_parser.parse_video(str_content)
1136
1137
1138
        modality = "video"
    else:
        raise NotImplementedError(f"Unknown part type: {part_type}")
1139

1140
1141
1142
    return {'type': modality} if wrap_dicts else (
        MODALITY_PLACEHOLDERS_MAP[modality] if interleave_strings else None
    )
1143
1144


1145
1146
1147
1148
1149
# No need to validate using Pydantic again
_AssistantParser = partial(cast, ChatCompletionAssistantMessageParam)
_ToolParser = partial(cast, ChatCompletionToolMessageParam)


1150
def _parse_chat_message_content(
1151
1152
    message: ChatCompletionMessageParam,
    mm_tracker: BaseMultiModalItemTracker,
1153
    content_format: _ChatTemplateContentFormat,
1154
    interleave_strings: bool,
1155
) -> list[ConversationMessage]:
1156
1157
1158
1159
    role = message["role"]
    content = message.get("content")

    if content is None:
1160
1161
1162
1163
1164
1165
        content = []
    elif isinstance(content, str):
        content = [
            ChatCompletionContentPartTextParam(type="text", text=content)
        ]
    result = _parse_chat_message_content_parts(
1166
1167
        role,
        content,  # type: ignore
1168
        mm_tracker,
1169
        wrap_dicts=(content_format == "openai"),
1170
        interleave_strings=interleave_strings,
1171
    )
1172

1173
1174
1175
1176
    for result_msg in result:
        if role == 'assistant':
            parsed_msg = _AssistantParser(message)

1177
1178
1179
1180
1181
            # The 'tool_calls' is not None check ensures compatibility.
            # It's needed only if downstream code doesn't strictly
            # follow the OpenAI spec.
            if ("tool_calls" in parsed_msg
                and parsed_msg["tool_calls"] is not None):
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
                result_msg["tool_calls"] = list(parsed_msg["tool_calls"])
        elif role == "tool":
            parsed_msg = _ToolParser(message)
            if "tool_call_id" in parsed_msg:
                result_msg["tool_call_id"] = parsed_msg["tool_call_id"]

        if "name" in message and isinstance(message["name"], str):
            result_msg["name"] = message["name"]

    return result

1193

1194
def _postprocess_messages(messages: list[ConversationMessage]) -> None:
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
    # per the Transformers docs & maintainers, tool call arguments in
    # assistant-role messages with tool_calls need to be dicts not JSON str -
    # this is how tool-use chat templates will expect them moving forwards
    # so, for messages that have tool_calls, parse the string (which we get
    # from openAI format) to dict
    for message in messages:
        if (message["role"] == "assistant" and "tool_calls" in message
                and isinstance(message["tool_calls"], list)):

            for item in message["tool_calls"]:
                item["function"]["arguments"] = json.loads(
                    item["function"]["arguments"])


1209
def parse_chat_messages(
1210
    messages: list[ChatCompletionMessageParam],
1211
    model_config: ModelConfig,
1212
    tokenizer: AnyTokenizer,
1213
    content_format: _ChatTemplateContentFormat,
1214
1215
) -> tuple[list[ConversationMessage], Optional[MultiModalDataDict]]:
    conversation: list[ConversationMessage] = []
1216
    mm_tracker = MultiModalItemTracker(model_config, tokenizer)
1217
1218

    for msg in messages:
1219
1220
1221
        sub_messages = _parse_chat_message_content(
            msg,
            mm_tracker,
1222
            content_format,
1223
1224
1225
1226
1227
            interleave_strings=(
                content_format == "string"
                and model_config.multimodal_config is not None
                and model_config.multimodal_config.interleave_mm_strings
            )
1228
        )
1229

1230
        conversation.extend(sub_messages)
1231

1232
1233
    _postprocess_messages(conversation)

1234
    return conversation, mm_tracker.all_mm_data()
1235
1236


1237
def parse_chat_messages_futures(
1238
    messages: list[ChatCompletionMessageParam],
1239
1240
    model_config: ModelConfig,
    tokenizer: AnyTokenizer,
1241
    content_format: _ChatTemplateContentFormat,
1242
1243
) -> tuple[list[ConversationMessage], Awaitable[Optional[MultiModalDataDict]]]:
    conversation: list[ConversationMessage] = []
1244
1245
1246
    mm_tracker = AsyncMultiModalItemTracker(model_config, tokenizer)

    for msg in messages:
1247
1248
1249
        sub_messages = _parse_chat_message_content(
            msg,
            mm_tracker,
1250
            content_format,
1251
1252
1253
1254
1255
            interleave_strings=(
                content_format == "string"
                and model_config.multimodal_config is not None
                and model_config.multimodal_config.interleave_mm_strings
            )
1256
        )
1257
1258
1259

        conversation.extend(sub_messages)

1260
1261
    _postprocess_messages(conversation)

1262
1263
1264
    return conversation, mm_tracker.all_mm_data()


1265
1266
def apply_hf_chat_template(
    tokenizer: Union[PreTrainedTokenizer, PreTrainedTokenizerFast],
1267
    conversation: list[ConversationMessage],
1268
    chat_template: Optional[str],
1269
    tools: Optional[list[dict[str, Any]]],
1270
    *,
1271
    model_config: ModelConfig,
1272
1273
    tokenize: bool = False,  # Different from HF's default
    **kwargs: Any,
1274
) -> str:
1275
    hf_chat_template = resolve_hf_chat_template(
1276
1277
1278
        tokenizer,
        chat_template=chat_template,
        tools=tools,
1279
        model_config=model_config,
1280
    )
1281

1282
    if hf_chat_template is None:
1283
1284
1285
1286
1287
        raise ValueError(
            "As of transformers v4.44, default chat template is no longer "
            "allowed, so you must provide a chat template if the tokenizer "
            "does not define one.")

1288
1289
1290
1291
1292
1293
1294
1295
1296
    try:

        return tokenizer.apply_chat_template(
            conversation=conversation,  # type: ignore[arg-type]
            tools=tools,  # type: ignore[arg-type]
            chat_template=hf_chat_template,
            tokenize=tokenize,
            **kwargs,
        )
1297

1298
1299
1300
1301
1302
1303
1304
1305
    # External library exceptions can sometimes occur despite the framework's
    # internal exception management capabilities.
    except Exception as e:

        # Log and report any library-related exceptions for further
        # investigation.
        logger.exception(
            "An error occurred in `transformers` while applying chat template")
1306
        raise ValueError(str(e)) from e
1307

1308
1309
def apply_mistral_chat_template(
    tokenizer: MistralTokenizer,
1310
    messages: list[ChatCompletionMessageParam],
1311
1312
    chat_template: Optional[str],
    tools: Optional[list[dict[str, Any]]],
1313
    **kwargs: Any,
1314
) -> list[int]:
1315
1316
    from mistral_common.exceptions import MistralCommonException

1317
1318
1319
1320
1321
1322
    # The return value of resolve_mistral_chat_template is always None,
    # and we won't use it.
    resolve_mistral_chat_template(
        chat_template=chat_template,
        **kwargs,
    )
1323

1324
1325
1326
1327
1328
1329
1330
1331
1332
    try:
        return tokenizer.apply_chat_template(
            messages=messages,
            tools=tools,
            **kwargs,
        )
    # mistral-common uses assert statements to stop processing of input
    # if input does not comply with the expected format.
    # We convert those assertion errors to ValueErrors so they can be
1333
    # properly caught in the preprocessing_input step
1334
    except (AssertionError, MistralCommonException) as e:
1335
        raise ValueError(str(e)) from e
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345

    # External library exceptions can sometimes occur despite the framework's
    # internal exception management capabilities.
    except Exception as e:

        # Log and report any library-related exceptions for further
        # investigation.
        logger.exception(
            "An error occurred in `mistral_common` while applying chat "
            "template")
1346
        raise ValueError(str(e)) from e
1347

1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
def get_history_tool_calls_cnt(conversation: list[ConversationMessage]):
    idx = 0
    for msg in conversation:
        if msg['role'] == 'assistant':
            tool_calls = msg.get('tool_calls')
            idx += len(list(tool_calls)) if tool_calls is not None else 0 # noqa
    return idx

def make_tool_call_id(id_type:str='random', func_name=None, idx=None):

    if id_type=='kimi_k2':
        return f'functions.{func_name}:{idx}'
    else:
        # by default return random
        return f"chatcmpl-tool-{random_uuid()}"