| 3481 | |
| 3482 | |
| 3483 | class Jinja2ChatFormatter: |
| 3484 | def __init__(self, template: str, *, bos_token: str, eos_token: str) -> None: |
| 3485 | self._eos_token = eos_token |
| 3486 | self._bos_token = bos_token |
| 3487 | self._template_text = template |
| 3488 | environment = ImmutableSandboxedEnvironment( |
| 3489 | loader=jinja2.BaseLoader(), |
| 3490 | trim_blocks=True, |
| 3491 | lstrip_blocks=True, |
| 3492 | ) |
| 3493 | environment.filters["from_json"] = self._from_json |
| 3494 | self._template = environment.from_string(template) |
| 3495 | |
| 3496 | @staticmethod |
| 3497 | def _from_json(value: Any) -> Any: |
| 3498 | if isinstance(value, str): |
| 3499 | return json.loads(value) |
| 3500 | return value |
| 3501 | |
| 3502 | @staticmethod |
| 3503 | def media_inputs_from_messages( |
| 3504 | messages: Sequence[ChatCompletionRequestMessage], |
| 3505 | ) -> List[MediaInput]: |
| 3506 | media_inputs: List[MediaInput] = [] |
| 3507 | for message in messages: |
| 3508 | content = message.content |
| 3509 | if not isinstance(content, list): |
| 3510 | continue |
| 3511 | for part in content: |
| 3512 | if not isinstance(part, dict): |
| 3513 | continue |
| 3514 | part_type = part.get("type") |
| 3515 | if part_type in {"image_url", "input_image", "image"}: |
| 3516 | image_url = part.get("image_url") or part.get("url") |
| 3517 | if isinstance(image_url, str): |
| 3518 | media_inputs.append(MediaInput(kind="image", url=image_url)) |
| 3519 | elif isinstance(image_url, dict) and isinstance(image_url.get("url"), str): |
| 3520 | media_inputs.append(MediaInput(kind="image", url=cast(str, image_url["url"]))) |
| 3521 | else: |
| 3522 | raise ValueError("image_url content part requires a URL string") |
| 3523 | continue |
| 3524 | if part_type == "audio_url": |
| 3525 | audio_url = part.get("audio_url") |
| 3526 | if isinstance(audio_url, str): |
| 3527 | media_inputs.append(MediaInput(kind="audio", url=audio_url)) |
| 3528 | elif isinstance(audio_url, dict) and isinstance(audio_url.get("url"), str): |
| 3529 | media_inputs.append(MediaInput(kind="audio", url=cast(str, audio_url["url"]))) |
| 3530 | else: |
| 3531 | raise ValueError("audio_url content part requires a URL string") |
| 3532 | continue |
| 3533 | if part_type == "input_audio": |
| 3534 | input_audio = part.get("input_audio") |
| 3535 | if isinstance(input_audio, dict): |
| 3536 | data = input_audio.get("data") |
| 3537 | audio_format = input_audio.get("format") |
| 3538 | else: |
| 3539 | data = part.get("data") |
| 3540 | audio_format = part.get("format") |
no outgoing calls
no test coverage detected
searching dependent graphs…