Multimodal support with Phi 3 Vision + Transformers

guidance/_parser.py

@@ -1,6 +1,6 @@
 import json
 import os
-from typing import Any, Generator, Optional, Tuple, Union
+from typing import Any, Generator, Optional, Sequence, Tuple, Union
 
 import llguidance  # type: ignore[import-untyped]
 import numpy as np
@@ -30,29 +30,11 @@ class TokenParser:
 
     def __init__(
         self,
-        grammar: Union[GrammarFunction, str],
-        tokenizer: Tokenizer,
-        prompt: bytes = b"",
-        ensure_bos_token: bool = True,
+        ll_interpreter: llguidance.LLInterpreter,
+        prompt_tokens: list[int]
     ):
-        if isinstance(grammar, GrammarFunction):
-            # we can't have a terminal as the root
-            if isinstance(grammar, Terminal):
-                grammar = Join([grammar])
-            serialized_grammar = json.dumps(grammar.ll_serialize())
-        else:
-            serialized_grammar = grammar
-
-        self.tokenizer = tokenizer
-        self.ll_tokenizer = llguidance.LLTokenizer(
-            llguidance.TokenizerWrapper(tokenizer)
-        )
-        self.ll_interpreter = llguidance.LLInterpreter(
-            self.ll_tokenizer,
-            serialized_grammar,
-            log_level=int(os.environ.get("LLGUIDANCE_LOG_LEVEL", "1")),
-        )
-        self._generator = self._parse(prompt, ensure_bos_token)
+        self.ll_interpreter = ll_interpreter
+        self._generator = self._parse(prompt_tokens)
         self._done = False
 
     def is_accepting(self) -> bool:
@@ -70,28 +52,10 @@ def advance(
             self._done = True
             return None, e.value
 
-    def _process_prompt(self, prompt: bytes, ensure_bos_token: bool) -> list[int]:
-        prompt_tokens = self.ll_interpreter.process_prompt(
-            self.tokenizer.encode(prompt)
-        )
-        if (
-            ensure_bos_token
-            and self.tokenizer.bos_token is not None
-            and prompt_tokens[:1] != [self.tokenizer.bos_token_id]
-        ):
-            # add the beginning of sequence token if needed
-            prompt_tokens = [self.tokenizer.bos_token_id] + prompt_tokens
-
-        return self.tokenizer.recode(prompt_tokens)
-
-
     def _parse(
         self,
-        prompt: bytes,
-        ensure_bos_token: bool,
+        tokens: list[int],
     ) -> Generator[Tuple[Optional[GenData], EngineCallResponse], Optional[int], EngineCallResponse]:
-        tokens = self._process_prompt(prompt=prompt, ensure_bos_token=ensure_bos_token)
-
         while True:
             mask, resp = self.ll_interpreter.mid_process()
             r = LLInterpreterResponse.model_validate_json(resp)
@@ -133,6 +97,54 @@ def _parse(
         return response
 
 
+def process_prompt(prompt_tokens: Sequence[int], ll_interpreter: llguidance.LLInterpreter, bos_token_id: Optional[int]=None) -> list[int]:
+    # Allows ll_interpreter to make adjustments to prompt tokens, such as token healing
+    processed_tokens = ll_interpreter.process_prompt(prompt_tokens)
+    if (
+        bos_token_id is not None
+        and prompt_tokens[:1] != [bos_token_id]
+    ):
+        # add the beginning of sequence token if needed
+        processed_tokens = [bos_token_id] + processed_tokens
+
+    return processed_tokens
+
+
+def process_grammar(grammar: Union[GrammarFunction, str]) -> str:
+    if isinstance(grammar, GrammarFunction):
+        # we can't have a terminal as the root
+        if isinstance(grammar, Terminal):
+            grammar = Join([grammar])
+        return json.dumps(grammar.ll_serialize())
+    else:
+        return grammar
+
+
+def create_token_parser(
+    grammar: Union[GrammarFunction, str],
+    tokenizer: Tokenizer,
+    prompt: bytes = b"",
+    ensure_bos_token: bool = True,
+    trace: bool = False
+) -> TokenParser:
+    serialized_grammar = process_grammar(grammar)
+    ll_tokenizer = llguidance.LLTokenizer(
+        llguidance.TokenizerWrapper(tokenizer)
+    )
+    ll_interpreter = llguidance.LLInterpreter(
+        ll_tokenizer,
+        serialized_grammar,
+        log_level=2 if trace else int(os.environ.get("LLGUIDANCE_LOG_LEVEL", "1")),
+    )
+    if ensure_bos_token and tokenizer.bos_token_id is not None:
+        bos_token_id = tokenizer.bos_token_id
+    else:
+        bos_token_id = None
+    prompt_tokens = tokenizer.encode(prompt)
+    processed_tokens = process_prompt(prompt_tokens, ll_interpreter, bos_token_id)
+    return TokenParser(ll_interpreter, processed_tokens)
+
+
 class ByteParserException(Exception):
     def __init__(self, *args, **kwargs):
         self.current_byte = kwargs.pop("current_byte", None)
@@ -149,7 +161,7 @@ def __init__(
         ensure_bos_token: bool = True,
     ):
         self.tokenizer = ByteTokenizer()
-        self.token_parser = TokenParser(grammar, self.tokenizer, prompt, ensure_bos_token)
+        self.token_parser = create_token_parser(grammar, self.tokenizer, prompt, ensure_bos_token)
         self.bytes = b""
         self.gen_data: Optional[GenData] = None
         self.pos = 0
@@ -289,3 +301,4 @@ def _update_capture(self, response: EngineCallResponse):
                     pass
                 self._variables[k] = v
                 self._variables_log_probs[k] = response.capture_group_log_probs[k]
+

guidance/chat.py

@@ -214,6 +214,9 @@ def get_role_end(self, role_name=None):
 phi3_medium_template = "{% for message in messages %}{% if (message['role'] == 'user') %}{{'<|user|>' + '\n' + message['content'] + '<|end|>' + '\n' + '<|assistant|>' + '\n'}}{% elif (message['role'] == 'assistant') %}{{message['content'] + '<|end|>' + '\n'}}{% endif %}{% endfor %}"
 
 
+# https://huggingface.co/microsoft/Phi-3-vision-128k-instruct/blob/main/tokenizer_config.json#L397
+phi3_vision_template = "{% for message in messages %}{{'<|' + message['role'] + '|>' + '\n' + message['content'] + '<|end|>\n' }}{% endfor %}{% if add_generation_prompt and messages[-1]['role'] != 'assistant' %}{{- '<|assistant|>\n' -}}{% endif %}"
+
 # Although the templates are different, the roles are the same between medium and small (for now)
 class Phi3SmallMediumChatTemplate(ChatTemplate):
     # available_roles = ["user", "assistant"]
@@ -230,9 +233,24 @@ def get_role_start(self, role_name):
     def get_role_end(self, role_name=None):
         return "<|end|>\n"
 
+class Phi3VisionChatTemplate(ChatTemplate):
+    template_str = phi3_vision_template
+
+    def get_role_start(self, role_name):
+        if role_name == "user":
+            return "<|user|>\n"
+        elif role_name == "assistant":
+            return "<|assistant|>\n"
+        else:
+            raise UnsupportedRoleException(role_name, self)
+
+    def get_role_end(self, role_name=None):
+        return "<|end|>\n"
 
 CHAT_TEMPLATE_CACHE[phi3_small_template] = Phi3SmallMediumChatTemplate
 CHAT_TEMPLATE_CACHE[phi3_medium_template] = Phi3SmallMediumChatTemplate
+CHAT_TEMPLATE_CACHE[phi3_vision_template] = Phi3VisionChatTemplate
+
 
 # --------------------------------------------------
 # @@@@ Mistral-7B-Instruct-v0.2 @@@@

guidance/library/_image.py

@@ -29,9 +29,5 @@ def image(lm, src: typing.Union[str, pathlib.Path, bytes], allow_local: bool = T
     else:
         raise Exception(f"Unable to load image bytes from {src}!")
 
-    bytes_id = str(id(bytes_data))
-
-    # set the image bytes
-    lm = lm.set(bytes_id, bytes_data)
-    lm += f"<|_image:{bytes_id}|>"
+    lm = lm.append_image(bytes_data)
     return lm

guidance/models/__init__.py

@@ -2,6 +2,7 @@
 
 # local models
 from .transformers._transformers import Transformers, TransformersTokenizer
+from .transformers._transformers_phi3v import TransformersPhi3Vision
 from .llama_cpp import LlamaCpp
 from ._mock import Mock, MockChat
 

guidance/models/_grammarless.py

@@ -258,7 +258,7 @@ def _reset_shared_data(self, new_data: bytes, temperature: float):
         self._last_stream_start = self._data
 
     def get_next_token(
-        self, token_ids: list[int], mask: Optional[bytes], temperature: float) -> int:
+        self, prompt: str, token_ids: list[int], mask: Optional[bytes], temperature: float, media: Optional[dict]=None) -> int:
 
         logger.debug(
             f"Start Grammarless.get_next_token({token_ids=}, {mask=}, {temperature=})"

guidance/models/_mock.py

@@ -80,9 +80,9 @@ def __init__(self, tokenizer, byte_patterns, compute_log_probs, force):
         # seed the random number generator
         self._rand_generator = np.random.default_rng(seed=42)
 
-    def get_next_token(self, token_ids: list[int], mask: Optional[bytes], temperature: float) -> int:
+    def get_next_token(self, prompt: bytes, token_ids: list[int], mask: Optional[bytes], temperature: float, media: Optional[dict]=None) -> int:
         self.called_temperatures.append(temperature)
-        return super().get_next_token(token_ids, mask, temperature)
+        return super().get_next_token(prompt, token_ids, mask, temperature, media)
 
     def get_logits(self, token_ids: list[int]) -> np.ndarray:
         """Pretends to compute the logits for the given token state."""