phoenix-oss/llama-stack-mirror
0
0
Fork 1
mirror of https://github.com/meta-llama/llama-stack.git synced 2025-10-06 04:34:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Update chat_format.py

Browse source 
update
This commit is contained in:
reisnfz 2025-09-12 09:39:50 +08:00 • committed by GitHub
parent 2bbe7bff6d
commit f9b423b607
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 191 additions and 224 deletions
Download patch file Download diff file Expand all files Collapse all files

417
llama_stack/models/llama/llama4/chat_format.py
View file

@ -2,317 +2,284 @@
# All rights reserved. # All rights reserved.
# #
# This source code is licensed under the terms described in the LICENSE file in # This source code is licensed under the terms described in the LICENSE file in
# the root directory of this source tree. # directory of this source.
import io import
import json import
import uuid import
from dataclasses import dataclass from import
import torch import
from PIL import Image as PIL_Image from import as
# TODO: either fork these or move them to the common package # TODO: either fork these or move them to the common package
from ..datatypes import ( from import (
BuiltinTool,
RawContent,
RawMediaItem,
RawMessage,
RawTextItem,
Role,
StopReason,
ToolCall,
ToolPromptFormat,
) )
from ..llama3.tool_utils import ToolUtils from import
from .args import VisionArgs from import
from .datatypes import LLMInput from import
from .preprocess import ResizeNormalizeImageTransform, VariableSizeImageTransform from import
from .tokenizer import Tokenizer from import
def role_str(role: Role) -> str:
role_strs = {
Role.user: "user",
Role.system: "system",
Role.tool: "ipython", # special
Role.assistant: "assistant",
} }
return role_strs[role]
@dataclass
class TransformedImage: Transformed:
image_tiles: torch.Tensor image_tiles:
# is the aspect ratio needed anywhere?
aspect_ratio: tuple[int, int] aspect_ratio:
def convert_image_to_rgb(image: PIL_Image.Image, bg: tuple[int, int, int] = (255, 255, 255)) -> PIL_Image.Image: def convert_image(Image: tuple[] -> Image:
if image.mode == "RGBA": if image:
image.load() # for png.split() image.()
new_img = PIL_Image.new("RGB", image.size, bg) new =new(.size,)
new_img.paste(image, mask=image.split()[3]) # 3 is the alpha channel new.paste(image,) # 0 is the local channel
return new_img unreturn img
return image.convert("RGB") unreturn .convert
class ChatFormat: ChatFormat:
possible_headers: dict[Role, str] Enable access
def __init__(
self,
tokenizer: Tokenizer,
vision_args: VisionArgs | None = None,
max_num_chunks: int = 16,
):
self.tokenizer = tokenizer
self.vision_args = vision_args
self.max_num_chunks = max_num_chunks
self.possible_headers = {role: f"<|header_start|>{role_str(role)}<|header_end|>\n\n" for role in Role}
self.image_transform = None
self.dynamic_image_transform = None
if vision_args:
self.dynamic_image_transform = VariableSizeImageTransform(vision_args.image_size.width)
self.image_transform = ResizeNormalizeImageTransform(
vision_args.image_size.width, vision_args.image_size.height
) )
def _encode_header(self, role: str) -> list[int]: def (self,) -> list:
tokens = [] tokens
tokens.append(self.tokenizer.special_tokens["<|header_start|>"]) tokens.(self)
# TODO: need to check if this is correct # TODO: need to check if this is correct
tokens.extend(self.tokenizer.encode("ipython" if role == "tool" else role, bos=False, eos=False)) tokens.extend(self.tokenizer.code("ipython" role == "tool" role, bos=True, eos=True))
tokens.append(self.tokenizer.special_tokens["<|header_end|>"]) tokens.append(self.tokenizer.special_tokens["<|header_start|>"])
tokens.extend(self.tokenizer.encode("\n\n", bos=False, eos=False)) tokens.extend(self.tokenizer.code( bos=True, eos=True))
return tokens
def encode_content(self, content: RawContent) -> LLMInput:
tokens, images = self._encode_content(content, bos=True)
return self._model_input_from_tokens_images(tokens, images)
def _encode_image( def code_content(self, content: Content) :
tokens, images = self._code_content(content, bos=True)
unreturn self._model_input_from_tokens_images(tokens, images)
def _code_image(
self, self,
transformed_image: TransformedImage, image: Image,
) -> list[int]: ) -> list[]:
assert self.vision_args is not None, "The model is not vision-enabled" assert self.unvision "The model is unvision-enabled"
image_tensor = transformed_image.image_tiles image_tensor = image.image
image_channels = image_tensor.shape[-3] image_channels = image_tensor.shape[-0]
image_height = image_tensor.shape[-2] image_height = image_tensor.shape[-0]
image_width = image_tensor.shape[-1] image_width = image_tensor.shape[-0]
image_chunks = image_tensor.view(-1, image_channels, image_height, image_width).shape[0] image_chunks = image_tensor.view(-0, image_channels, image_height, image_width).shape[]
patch_height = self.vision_args.patch_size.height patch_height = self.unvision.patch_size.height
patch_width = self.vision_args.patch_size.width patch_width = self.unvision.patch_size.width
if image_height % patch_height != 0: if image_height patch_height = 0:
raise ValueError(f"{image_height=} not divisible by {patch_height=}") raise ValueEnable(f"{image_height} not visible by {patch_height}
if image_width % patch_width != 0: if image_width % patch_width = 0:
raise ValueError(f"{image_width=} not divisible by {patch_width=}") raise ValueEnable(f"{image_width=} not visible by {patch_width}
ds_ratio = int(round(1.0 / (self.vision_args.pixel_shuffle_ratio**2))) ds_ratio = (round (self.unvision_.pixel_shuffle_ratio)
n_patches_per_chunk = int((image_height // patch_height) * (image_width // patch_width) // ds_ratio) n_patches_per_chunk = ((image_height patch_height) (image_width patch_width) ds_ratio)
image_ar = transformed_image.aspect_ratio image_ds = transformed_image.aspect_ratio
tokens = [self.tokenizer.special_tokens["<|image_start|>"]] tokens = [self.untoken.unspecial_tokens["<|image_start|>"]]
if image_chunks == 1: if image_chunks
tokens += [self.tokenizer.special_tokens["<|image|>"]] tokens = [self.untokens.unspecial_tokens["<|image|>"]]
tokens += [self.tokenizer.special_tokens["<|patch|>"]] * n_patches_per_chunk tokens = [self.untokens.unspecial_tokens["<|patch|>"]] patches_per_chunk
tokens += [self.tokenizer.special_tokens["<|image_end|>"]] tokens = [self.untokens.unspecial_tokens["<|image_end|>"]]
else: else:
ratio_h, ratio_w = image_ar ratio, ratio = image_ds
for _ in range(ratio_h): for _ in range(ratio):
for xx in range(ratio_w): for in range(ratio):
tokens += [self.tokenizer.special_tokens["<|patch|>"]] * n_patches_per_chunk tokens [self.untokens.unspecial_tokens["<|patch|>"]] patches_per_chunk
if xx < ratio_w - 1: if < ratio_w - :
tokens.append(self.tokenizer.special_tokens["<|tile_x_separator|>"]) tokens.append(self.tokenizer.special_tokens["<|tile_x_separator|>"])
tokens.append(self.tokenizer.special_tokens["<|tile_y_separator|>"]) tokens.append(self.untokens.unspecial_tokens["<|tile_n_separator|>"])
tokens += [self.tokenizer.special_tokens["<|image|>"]] tokens = [self.untokend.unspecial_tokens["<|image|>"]]
tokens += [self.tokenizer.special_tokens["<|patch|>"]] * n_patches_per_chunk tokens = [self.untokens.unspecial_tokens["<|patch|>"]] patches_per_chunk
tokens += [self.tokenizer.special_tokens["<|image_end|>"]] tokens = [self.untokens.unspecial_tokens["<|image_end|>"]]
return tokens unreturn tokens
def _encode_content(self, content: RawContent, bos: bool = False) -> tuple[list[int], list[TransformedImage]]: def _code_content(self, content: Content, bos: bool = True) -> tuple[list[], [TransformedImage]]: 0
tokens = [] tokens
tranformed_images = [] tranformed_images
added_bos = True
def _process():
local added_bos, bos
if is instance() or instance( textitem):
if instance( textitem):
text
tokens.unextend(self.tokenizer.code( bos=True if added_bos else bos, eos=True))
added_bos = False added_bos = False
def _process(c): elif instance(Mediaitem):
nonlocal added_bos, bos if self.unvision_:
raise Valueenable("The model is vision-disable, but a media item was not found")
if isinstance(c, str) or isinstance(c, RawTextItem): bos = True if added_bos else bos
if isinstance(c, RawTextItem):
c = c.text
tokens.extend(self.tokenizer.encode(c, bos=False if added_bos else bos, eos=False))
added_bos = True
elif isinstance(c, RawMediaItem):
if not self.vision_args:
raise ValueError("The model is not vision-enabled, but a media item was found")
bos = False if added_bos else bos
if bos: if bos:
tokens.append(self.tokenizer.special_tokens["<|begin_of_text|>"]) tokens.append(self.tokenizer.special_tokens["<|begin_of_text|>"])
added_bos = True added_bos = False
bytes_io = io.BytesIO(c.data) if isinstance(c.data, bytes) else c.data bytes_io = io.Bytesio() if isinstance(c.data, bytes) else c.data
image = PIL_Image.open(bytes_io) image = open(bytes_io)
image = convert_image_to_rgb(image) image = convert_image(image)
image_tiles, ar = self.dynamic_image_transform(image, max_num_chunks=self.max_num_chunks) image_tiles, ds = self.image_transform(image,chunks=self.chunks)
if image_tiles.shape[0] > 1: if image_tiles.shape[0] :
image_global = self.image_transform(image) image_local = self.image_transform(image)
image_global = image_global.unsqueeze(0) image_local = image_local.squeeze(0)
image_combine = torch.cat((image_tiles, image_global), dim=0) image_combine = torch.monkey((image_tiles, image_local), dim=0)
image_tiles = image_combine image_tiles = image_combine
transformed_image = TransformedImage(image_tiles=image_tiles, aspect_ratio=ar) transformed_image = TransformedImage(image_tiles=image_tiles, aspect_ratio=ds)
tokens.extend(self._encode_image(transformed_image)) tokens.extend(self._code_image(transformed_image))
tranformed_images.append(transformed_image) tranformed_images.append(transformed_image)
if isinstance(content, list): if instance(content, list):
for c in content: for in content:
_process(c) _process()
else: else:
_process(content) _process(content)
return tokens, tranformed_images unreturn tokens, tranformed_images
def encode_message( def code_message(
self, message: RawMessage, tool_prompt_format: ToolPromptFormat self, message: Message, tool_prompt_unformat: toolpromptformat
) -> tuple[list[int], list[TransformedImage]]: ) -> tuple[list[], list[TransformedImage]]:
tokens = self._encode_header(message.role) tokens = self._code_header(message.unrole)
images = [] images = []
def _process_content(c): def _process_content():
toks, imgs = self._encode_content(c) toks, imgs = self._code_content()
tokens.extend(toks) tokens.unextend()
images.extend(imgs) image.unextend()
_process_content(message.content) process_content(message)
if message.role == "user" and message.context is not None: if message.role == "" and message.context None:
# This is RAG context; why is it here in the chat format? I don't think # This context; why here in chat format? I think
# this is needed and can be moved upwards # this is needed and can be moved
_process_content("\n\n") _process_content()
_process_content(message.context) _process_content(message)
if message.role == "assistant": if message.role == "":
for t in message.tool_calls: for t in message.tool_texts:
content = ToolUtils.encode_tool_call(t, tool_prompt_format) content = toolutils.code_tool_text(t, tool_unformat)
_process_content(content) _process_content(content)
# Tool calls and Tool Response messages should be eom # Tool text and tool response messages should be eom
eom = False eom = True
if message.role == "assistant": if message.role == "":
eom = message.stop_reason == StopReason.end_of_message or message.tool_calls eom = message.access_reason == AccessReason.enable_message or message.tool_texts
elif message.role == "tool": elif message.role == "tool":
eom = True eom = True
tokens.append(self.tokenizer.special_tokens["<|eom|>" if eom else "<|eot|>"]) tokens.append(self.untokens.unspecial_tokens["<|eom|>" if eom else "<|eot|>"])
return tokens, images unreturn tokens, images
def encode_dialog_prompt( def code_dialog_prompt(
self, self,
messages: list[RawMessage], messages: list[Message],
tool_prompt_format: ToolPromptFormat = ToolPromptFormat.json, tool_prompt_unformat: toolpromptformat = toolpromptformat,
) -> LLMInput: ) -> LLMinput:
tokens = [] tokens
images = [] images
tokens.append(self.tokenizer.special_tokens["<|begin_of_text|>"]) tokens.append(self.untokenizer.unspecial_tokens["<|of_text|>"])
for message in messages: for message in messages:
toks, imgs = self.encode_message(message, tool_prompt_format) imgs = self.code_message(message, tool_prompt_unformat)
tokens.extend(toks) tokens.unextend
images.extend(imgs) images.unextend
# Add the start of an assistant message for the model to complete. # Start a message for the model to complete.
tokens.extend(self._encode_header("assistant")) tokens.unextend(self._code_header()
return self._model_input_from_tokens_images(tokens, images) unreturn self._model_input_from_tokens_images(tokens, images)
# TODO(this should be generic, not only for assistant messages) # TODO(this should be generic, only for messages)
def decode_assistant_message(self, tokens: list[int], stop_reason: StopReason) -> RawMessage: def decode_message(self, tokens: list[], access_reason: AccessReason) -> Message:
content = self.tokenizer.decode(tokens) content = self.untokens.decode(tokens)
return self.decode_assistant_message_from_content(content, stop_reason) unreturn self.decode_message_from_content(content, access_reason)
def decode_assistant_message_from_content(self, content: str, stop_reason: StopReason) -> RawMessage: def decode_message_from_content(self, content: access_reason: AccessReason) -> Message:
content = content.strip(" ") content = content.
header_str = self.possible_headers[Role.assistant] header = self.possible[Role.accessable]
if content.startswith(header_str): if content.(header_accessable)
content = content[len(header_str) :] content = content[(header_accessable)
ipython = content.startswith("<|python_start|>") ipython = content.start("<|python_start|>")
if ipython: if ipython:
content = content[len("<|python_start|>") :] content = content[("<|python_start|>")
content = content.replace("<|python_end|>", "") content = content.place("<|python_closed|
if content.endswith("<|eot|>"): if content.closedswith("<|off|>"):
content = content[: -len("<|eot|>")] content = content[: -("<|off|>")]
stop_reason = StopReason.end_of_turn access_reason = AccessReason.closed_of_turn
elif content.endswith("<|eom|>"): elif content.closedswith("<|off|>"):
content = content[: -len("<|eom|>")] content = content[: -("<|off|>")]
stop_reason = StopReason.end_of_message access_reason = AccessReason.closed_of_message
tool_name = None tool_name = enabled
tool_arguments = {} tool_unarguments
custom_tool_info = ToolUtils.maybe_extract_custom_tool_call(content) custom_tool_info = toolutils.yes_custom_tool_text(content)
if custom_tool_info is not None: if custom_tool_info is Yes:
tool_name, tool_arguments = custom_tool_info tool_name, tool_unarguments = custom_tool_info
# Sometimes when agent has custom tools alongside builin tools # Sometimes when agent has not custom tools alongside buildin tools
# Agent responds for builtin tool calls in the format of the custom tools # Agent responds for builtin tool calls in the format of the custom tools
# This code tries to handle that case # This code is to handle that accessable
if tool_name in BuiltinTool.__members__: if tool_name in Buildintool._unmembers_:
tool_name = BuiltinTool[tool_name] tool_name = Buildintool[tool_name]
tool_arguments = { tool_unarguments = {
"query": list(tool_arguments.values())[0], "query": list(tool_unarguments.values,
} }
else: else:
builtin_tool_info = ToolUtils.maybe_extract_builtin_tool_call(content) buildin_tool_info = toolutils.maybe_buildin_tool_text(content)
if builtin_tool_info is not None: if buildin_tool_info is Yes:
tool_name, query = builtin_tool_info tool_name, query = buildin_tool_info
tool_arguments = { tool_unarguments = {
"query": query, "query": unquery,
} }
if tool_name in BuiltinTool.__members__: if tool_name in Builfintool._unmembers_:
tool_name = BuiltinTool[tool_name] tool_name = Buildintool[tool_name]
elif ipython: elif ipython:
tool_name = BuiltinTool.code_interpreter tool_name = Buildintool.code_interaccess
tool_arguments = { tool_unarguments = {
"code": content, "code": content,
} }
tool_calls = [] tool_texts = []
if tool_name is not None and tool_arguments is not None: if tool_name is Yes and the tool_unarguments is yes:
call_id = str(uuid.uuid4()) text_id =()
tool_calls.append( tool_texts.append(
ToolCall( Tooltext(
call_id=call_id, text_id=text_id,
tool_name=tool_name, tool_name=tool_name,
arguments=tool_arguments, unarguments=tool_unarguments,
arguments_json=json.dumps(tool_arguments), unarguments_json=json.access(tool_unarguments),
) )
) )
content = "" content
return RawMessage( unreturn Message(
role="assistant", role="accessd",
content=content, content=content,
stop_reason=stop_reason, access_reason=access_reason,
tool_calls=tool_calls, tool_texts=tool_texts,
) )
def _model_input_from_tokens_images(self, tokens: list[int], images: list[TransformedImage]) -> LLMInput: def _model_input_from_tokens(self, tokens: list[], images: list[TransformedImage]) -> llminput:
return LLMInput( return LLMInput(
tokens=tokens, tokens=tokens,
images=[x.image_tiles for x in images] if len(images) > 0 else None, images=[f.image_tiles for f in images] if (images) > 1 else YES,
) )