初始化项目，由ModelHub XC社区提供模型

Model: kakaocorp/kanana-1.5-15.7b-a3b-instruct
Source: Original Platform

kanana_tool_calls/functionary_kanana_tool_parser.py

@@ -0,0 +1,401 @@
+import json
+from json import JSONDecodeError, JSONDecoder
+import copy
+import uuid
+from pathlib import Path
+from typing import Dict, List, Optional, Union, Sequence
+from abc import ABC, abstractmethod
+
+import partial_json_parser
+from partial_json_parser.core.options import Allow
+
+from transformers import PreTrainedTokenizerBase
+from vllm.entrypoints.openai.protocol import (
+    ChatCompletionRequest,
+    ExtractedToolCallInformation,
+    DeltaMessage,
+    DeltaFunctionCall,
+    DeltaToolCall
+)
+from vllm.entrypoints.openai.tool_parsers.abstract_tool_parser import ToolParser, ToolParserManager
+from vllm.entrypoints.openai.tool_parsers.utils import find_common_prefix
+from vllm.logger import init_logger
+from vllm.utils import random_uuid
+from vllm.transformers_utils.tokenizer import AnyTokenizer
+
+
+
+from transformers import PreTrainedTokenizer
+
+
+logger = init_logger(__name__)
+
+
+
+class BaseTemplate(ABC):
+    @abstractmethod
+    def response_to_messages(self, generated_text):
+        raise NotImplementedError
+
+class FunctionaryV3Llama31Template(BaseTemplate):
+    def __init__(
+            self,
+            tokenizer: PreTrainedTokenizer
+    ):
+        self.tokenizer = tokenizer
+
+        self.system_tokens = "<|start_header_id|>system<|end_header_id|>\n\n"
+        self.user_tokens = "<|start_header_id|>user<|end_header_id|>\n\n"
+        self.assistant_tokens = "<|start_header_id|>assistant<|end_header_id|>\n\n"
+        self.tool_tokens = "<|start_header_id|>ipython<|end_header_id|>\n\n"
+
+    def parse_function_call_from_text(self, function_call_text: str) -> Optional[Dict]:
+        index = function_call_text.find(">")
+        if index >= 0:
+            func_name = function_call_text[:index].strip()
+            arguments = function_call_text[index + 1:].strip()
+            return {"name": func_name, "arguments": arguments}
+        return None
+
+    def response_to_messages(self, generated_text):
+        # first remove stop tokens if there exists
+        for stop in [self.tokenizer.eos_token, "<|eom_id|>", "<|end_of_text|>"]:
+            if generated_text.endswith(stop):
+                generated_text = generated_text[: -len(stop)]
+
+        tool_calls = []
+        text_response = ""
+
+        func_prefix = "<function="
+        end_func = "</function>"
+        python_tag = "<|python_tag|>"
+
+        while len(generated_text) > 0:
+            if generated_text.startswith(python_tag):  # check if use code interpreter
+                code = generated_text[len(python_tag) :]
+                function_call = {
+                    "name": "python",
+                    "arguments": code,
+                }
+
+                tool_calls.append(
+                    {
+                        "type": "function",
+                        "id": str(uuid.uuid4()),
+                        "function": function_call,
+                    }
+                )
+                generated_text = ""
+            elif generated_text.startswith(func_prefix):  # Check if function_call
+                end_index = generated_text.find(end_func)
+                if end_index >= 0:
+                    function_call_text = generated_text[len(func_prefix) : end_index]
+                    function_call = self.parse_function_call_from_text(function_call_text)
+
+                    tool_calls.append(
+                        {
+                            "type": "function",
+                            "id": str(uuid.uuid4()),
+                            "function": function_call,
+                        }
+                    )
+                    generated_text = generated_text[end_index + len(end_func) :]
+                else:
+                    # TODO cannot find close function call
+                    text_response += generated_text
+                    break
+            else:  # If text-response
+                text_response += generated_text[0]
+                generated_text = generated_text[1:]
+
+        if not text_response:
+            text_response = None
+        elif len(text_response.strip()) == 0:
+            text_response = None
+
+        if not tool_calls:
+            tool_calls = None
+
+        if tool_calls:
+            return {"role": "assistant", "content": text_response, "tool_calls": tool_calls}
+        else:
+            return {"role": "assistant", "content": text_response}
+
+
+
+# partial_json_parser doesn't support extra data and
+# JSONDecorder.raw_decode doesn't support partial JSON
+def partial_json_loads(input_str, flags):
+    try:
+        return partial_json_parser.loads(input_str, flags), len(input_str)
+    except JSONDecodeError as e:
+        if "Extra data" in e.msg:
+            dec = JSONDecoder()
+            return dec.raw_decode(input_str)
+        else:
+            raise
+
+
+def is_complete_json(input_str):
+    try:
+        json.loads(input_str)
+        return True
+    except JSONDecodeError:
+        return False
+
+@ToolParserManager.register_module(["functionary_v3_llama_31"])
+class FunctionaryV3Llama31ToolParser(ToolParser):
+    def __init__(self, tokenizer: Union[PreTrainedTokenizerBase, AnyTokenizer]):
+        super().__init__(tokenizer)
+        self.template = FunctionaryV3Llama31Template(tokenizer=tokenizer)
+
+        self._func_prefix = "<function="
+        self._func_suffix = "</function>"
+        self._python_tag = "<|python_tag|>"
+        self.current_tool_name_sent: bool = False
+        self.current_tool_id: int = -1
+        self.streamed_args_for_tool: List[str] = [
+        ]  # map what has been streamed for each tool so far to a list
+        self.eos_token = tokenizer.eos_token
+        self._python_tag_id = tokenizer.encode(self._python_tag,
+                                             add_special_tokens=False)[0]
+
+        # added buffer for each tool call parser
+        self._buffer = ""
+    def extract_tool_calls(
+            self, model_output: str,
+            request: ChatCompletionRequest) -> ExtractedToolCallInformation:
+        result = self.template.response_to_messages(model_output)
+        if "tool_calls" in result:
+            return ExtractedToolCallInformation(tools_called=True,
+                                         tool_calls=result["tool_calls"],
+                                         content=None)
+        else:
+            return ExtractedToolCallInformation(tools_called=False,
+                                                tool_calls=[],
+                                                content=result["content"])
+
+    # our template: <function=function_name>{"arg":"var"}</function>
+    def extract_tool_calls_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],
+        request: ChatCompletionRequest,
+    ) -> Union[DeltaMessage, None]:
+    
+        # if current_text does not start with function tag (or python tag),
+        if not (current_text.startswith(self._python_tag)
+                or current_text.startswith(self._func_prefix)
+                or self._func_prefix.startswith(current_text)):
+            # for cases like "The answer is <function="
+            # let current_text="<function="
+            if self._func_prefix in current_text:
+                idx = current_text.find(self._func_prefix)
+                current_text = current_text[idx:]
+                self._buffer = ""
+            # for cases like "The answer is <function"
+            # add delta_text to buffer to figure out whether to print or not later
+            elif delta_text.endswith("<")  or (current_text.endswith("<function") and delta_text.endswith("function")):
+                self._buffer += delta_text
+                return DeltaMessage(content=None)
+            # for cases that does not include "<function" at all,
+            # stream right away as delta.content                
+            else:
+                delta_text = self._buffer + delta_text
+                self._buffer = ""
+                return DeltaMessage(content=delta_text)
+
+        # for cases like "<" or "<function"
+        # add to delta_text to buffer 
+        if delta_text.endswith("<") or (current_text.endswith("<function") and delta_text.endswith("function")):
+            self._buffer += delta_text
+            
+            
+        # CHECK: this part not working (should use current_token_ids)
+        # if current_text ends with stop token,
+        # remove it from the text
+        # CHECK: sometimes text is generated beyond <|eom_id|>
+        for stop in [self.eos_token, "<|eom_id|>", "<|end_of_text|>"]:
+            if current_text.rstrip().endswith(stop):
+                current_text = current_text.rstrip()[: -len(stop)]
+
+        # if current_tool_name is not sent yet,
+        # don't allow partial sending of strings
+        # (supposedly, openai also only sends the entire tool name at once)
+        # not really relevant to kanana
+        flags = Allow.ALL if self.current_tool_name_sent \
+            else Allow.ALL & ~Allow.STR
+
+        try:
+            tool_call_arr = []
+            is_complete = []
+            obj = {}
+            try:
+                # set start_idx
+                start_idx = 0
+                while start_idx < len(current_text):
+                    name = None
+                    # set function_name and move start_idx to the end of function_name
+                    if current_text[start_idx:].startswith(self._python_tag):
+                        name = "python"
+                        start_idx += len(self._python_tag)
+                    elif current_text[start_idx:].startswith(self._func_prefix):
+                        idx = current_text[start_idx:].find(">")
+                        # found ">" which means function name is ready
+                        if idx != -1:
+                            function_name = current_text[start_idx+len(self._func_prefix):start_idx+idx]
+                            name = function_name
+                            start_idx += len(self._func_prefix) + len(function_name) + len(">")
+                    # for cases like: </function>abcd<function=
+                    # ignore tokens in between </function> and <function=
+                    # and move start_idx to the start of the new function
+                    else:
+                        idx = current_text[start_idx:].find(self._func_prefix)
+                        if idx != -1:
+                            start_idx += idx
+                            continue
+
+                    # partially load the function arguments
+                    (obj, end_idx) = partial_json_loads(current_text[start_idx:], flags)
+                    if "arguments" not in obj:
+                        obj["arguments"] = json.loads(json.dumps(obj, ensure_ascii=False))
+                    if name is not None:
+                        obj["name"] = name
+
+                    is_complete.append(
+                        is_complete_json(current_text[start_idx:start_idx +
+                                                      end_idx]))
+
+                    start_idx += end_idx
+
+                    # TODO: is this necessary for kanana? can we remove this?
+                    # depending on the prompt Llama can use
+                    # either arguments or parameters
+                    if "parameters" in obj:
+                        assert "arguments" not in obj, \
+                            "model generated both parameters and arguments"
+                        obj["arguments"] = obj["parameters"]
+                    tool_call_arr.append(obj)
+
+                    # if you can find </fuction>, which declares the end of a function,
+                    # move start_idx to the end so it can skip </fuction>
+                    function_end_idx = current_text[start_idx:].find(self._func_suffix)
+                    if function_end_idx != -1:
+                        start_idx += len(self._func_suffix)
+                    # if </function> hasn't been generated fully yet,
+                    # but the function is complete, we break out from while loop
+                    # and stream it right away
+                    elif is_complete[-1]:
+                        break
+            except partial_json_parser.core.exceptions.MalformedJSON:
+                logger.debug('not enough tokens to parse into JSON yet')
+                return None
+
+            # current_tool_call is the one that is being streamed
+            current_tool_call: dict = tool_call_arr[self.current_tool_id] \
+                if len(tool_call_arr) > 0 else {}
+
+            # case0: if there is no tool call in the array, return None
+            if len(tool_call_arr) == 0:
+                return None
+
+            # case1: we are starting a new tool in the array
+            #   -> tool_call_arr has > 0 length AND has more elements than cursor
+            elif (len(tool_call_arr) > 0
+                  and len(tool_call_arr) > self.current_tool_id + 1):
+                # if we're moving on to a new call, first make sure we
+                # haven't missed anything in the previous one that was
+                # auto-generated due to JSON completions, but wasn't
+                # streamed to the client yet.
+                # print("starting a new tool in the array, print remaining")
+                if self.current_tool_id >= 0:
+                    cur_arguments = current_tool_call.get("arguments")
+                    if cur_arguments:
+                        cur_args_json = json.dumps(cur_arguments, ensure_ascii=False)
+                        sent = len(
+                            self.streamed_args_for_tool[self.current_tool_id]) # streamed args for current tool call
+                        argument_diff = cur_args_json[sent:] # args not yet streamed for current tool call
+
+                        logger.debug("got arguments diff: %s", argument_diff)
+                        delta = DeltaMessage(tool_calls=[
+                            DeltaToolCall(index=self.current_tool_id,
+                                          function=DeltaFunctionCall(
+                                              arguments=argument_diff).
+                                          model_dump(exclude_none=True))
+                        ])
+                        self.streamed_args_for_tool[
+                            self.current_tool_id] += argument_diff # update streamed args for current tool call
+                    else:
+                        delta = None
+                else:
+                    delta = None
+                # re-set stuff pertaining to progress in the current tool
+                self.current_tool_id = len(tool_call_arr) - 1 # update current tool call 
+                self.current_tool_name_sent = False
+                self.streamed_args_for_tool.append("")
+                logger.debug("starting on new tool %d", self.current_tool_id)
+                return delta
+
+            # case2: if the current tool name hasn't been sent, send if available
+            elif not self.current_tool_name_sent:
+                function_name = current_tool_call.get("name")
+                if function_name:
+                    delta = DeltaMessage(tool_calls=[
+                        DeltaToolCall(index=self.current_tool_id,
+                                      type="function",
+                                      id=f"chatcmpl-tool-{random_uuid()}",
+                                      function=DeltaFunctionCall(
+                                          name=function_name).model_dump(
+                                              exclude_none=True))
+                    ])
+                    self.current_tool_name_sent = True
+                else:
+                    delta = None
+
+            # case3: now we know we're on the same tool call
+            # and can stream arguments
+            else:
+                cur_arguments = current_tool_call.get("arguments")
+                delta = None
+                # '{}' 인경우도 출력 해야함.
+                sent = len(
+                    self.streamed_args_for_tool[self.current_tool_id]) # streamed args for current tool call
+                cur_args_json = json.dumps(cur_arguments, ensure_ascii=False)
+                prev_arguments = self.prev_tool_call_arr[
+                    self.current_tool_id].get("arguments")
+                argument_diff = None
+                # if current tool call is complete
+                if is_complete[self.current_tool_id]:
+                    argument_diff = cur_args_json[sent:]
+                # if current tool call is not complete and prev_arguments exists
+                elif prev_arguments:
+                    prev_args_json = json.dumps(prev_arguments, ensure_ascii=False)
+                    if cur_args_json != prev_args_json:
+                        # common prefix betwn prev and current args
+                        prefix = find_common_prefix(
+                            prev_args_json, cur_args_json)
+                        argument_diff = prefix[sent:]
+                if argument_diff is not None and len(argument_diff) > 0:
+                    delta = DeltaMessage(tool_calls=[
+                        DeltaToolCall(index=self.current_tool_id,
+                                        function=DeltaFunctionCall(
+                                            arguments=argument_diff).
+                                        model_dump(exclude_none=True))
+                    ])
+
+                    self.streamed_args_for_tool[
+                        self.current_tool_id] += argument_diff
+            # update previous tool call array
+            self.prev_tool_call_arr = tool_call_arr
+            return delta
+
+        except Exception as e:
+            logger.exception("Error trying to handle streaming tool call.")
+            logger.debug(
+                "Skipping chunk as a result of tool streaming extraction "
+                "error")
+            return None

kanana_tool_calls/lmalign_v1.jinja

@@ -0,0 +1,320 @@
+{# version=v3-llama3.1 #}{%- macro append_new_param_info(param_declaration, comment_info, examples_info, depth) -%}
+    {%- set offset = "" -%}
+    {%- if depth >= 1 -%}
+        {%- set offset = "    " * depth -%}
+    {%- endif -%}
+    {%- if comment_info != "<|NONE|>" -%}
+        {{ "\n" + offset + comment_info }}
+        {%- if examples_info | length > 0 -%}
+            {# Append each example info #}
+            {%- for example in examples_info -%}
+                {{ "\n" + offset + "// " + example|string|replace("'", '"') }}
+            {%- endfor -%}
+        {%- endif -%}
+    {%- endif -%}
+    {{ "\n" + offset + param_declaration }}
+{%- endmacro -%}
+
+{%- macro convert_data_type(param_type) -%}
+    {%- if param_type == "integer" or param_type == "float" -%}
+        {{ "number" }}
+    {%- else -%}
+        {{ param_type }}
+    {%- endif -%}
+{%- endmacro -%}
+
+{%- macro get_param_type(param) -%}
+    {%- set param_type = "any" -%}
+
+    {%- if "type" in param -%}
+        {%- set raw_param_type = param["type"] -%}
+        {%- if raw_param_type is iterable and raw_param_type is not string -%}
+            {%- set param_type = raw_param_type | join(" | ") -%}
+        {%- else -%}
+            {%- set param_type = raw_param_type -%}
+        {%- endif -%}
+        {{ convert_data_type(param_type) }}
+    {%- elif "oneOf" in param -%}
+        {%- set one_of_types = param["oneOf"]|selectattr("type", "defined")|list -%}
+        {%- set one_of_types = one_of_types|map(attribute="type")|unique|list -%}
+        {{ convert_data_type(one_of_types | join(" | ")) }}
+    {%- endif -%}
+{%- endmacro -%}
+
+{%- macro get_format_param(param) -%}
+    {%- if "format" in param -%}
+        {{ param["format"] }}
+    {%- elif "oneOf" in param -%}
+        {%- set formats = [] -%}
+        {%- for item in param["oneOf"] -%}
+            {%- if "format" in item -%}
+                {%- if item["format"] == param["oneOf"][-1]["format"] -%}
+                    {{ item["format"] }}
+                {%- else -%}
+                    {{ item["format"] + " or "}}
+                {%- endif -%}
+            {%- endif -%}
+        {%- endfor -%}
+    {%- else -%}
+        {{ "<|NONE|>" }}
+    {%- endif -%}
+{%- endmacro -%}
+
+{%- macro get_param_info(param) -%}
+    {%- set param_type = param.get("type", "any") -%}
+    {%- set format_param = get_format_param(param) -%}
+
+    {%- if "description" in param or "default" in param or format_param != "<|NONE|>" or param["maximum"] or param["minimum"] or param["maxLength"] or param["minLength"] -%}
+        {{ "//" }}
+        {%- if "description" in param -%}
+            {%- set desc = param["description"] -%}
+            {%- if not desc.endswith(".") -%}
+                {%- set desc = desc + "." -%}
+            {%- endif -%}
+            {{ " " + desc }}
+        {%- endif -%}
+
+        {%- if "default" in param -%}
+            {%- set default_value = param["default"] -%}
+            {%- if param_type == "string" -%}
+                {%- set default_value = '"' ~ default_value ~ '"' -%}
+            {%- endif -%}
+            {{ " Default=" ~ default_value ~ "." }}
+        {%- endif -%}
+
+        {%- set format_param = get_format_param(param) -%}
+        {%- if format_param != "<|NONE|>" -%}
+            {{ " Format=" ~ format_param }}
+        {%- endif -%}
+
+        {%- for field, field_name in [("maximum", "Maximum"), ("minimum", "Minimum"), ("maxLength", "Maximum length"), ("minLength", "Minimum length")] -%}
+            {%- if field in param -%}
+                {{ " " + field_name ~ "=" ~ param[field] }}
+            {%- endif -%}
+        {%- endfor -%}
+    {%- else -%}
+        {{ "<|NONE|>"}}
+    {%- endif -%}
+{%- endmacro -%}
+
+{%- macro get_enum_option_str(enum_options) -%}
+    {%- for v in enum_options -%}
+        {%- if v is string -%}
+            {{ '"' + v + '"' }}
+        {%- else -%}
+            {{ v }}
+        {%- endif -%}
+        {%- if enum_options|length > 0 and v != enum_options[-1] -%}
+            {{ " | " }}
+        {%- endif -%}
+    {%- endfor -%}
+{%- endmacro -%}
+
+{%- macro get_array_typescript(param_name, param_dic, depth) -%}
+    {%- set offset = '' -%}
+    {%- if depth >= 1 -%}
+        {%- set offset = "    " * depth -%}
+    {%- endif -%}
+    {%- set items_info = param_dic.get('items', {}) -%}
+
+    {%- if items_info|length == 0 -%}
+        {%- if param_name -%}
+            {{ "\n" + offset + param_name + ": []" }}
+        {%- else -%}
+            {{ "\n" + offset + "[]" }}
+        {%- endif -%}
+    {%- else -%}
+        {%- set array_type = get_param_type(items_info) -%}
+        {%- if array_type == 'object' -%}
+            {%- if param_name -%}
+                {{ "\n" + offset + param_name + ": {" }}
+            {%- else -%}
+                {{ "\n" + offset + "{" }}
+            {%- endif -%}
+            {{ get_parameter_typescript(items_info.get('properties', {}), items_info.get('required', []), depth + 1) -}}
+            {{- "\n" + offset + "}[]" }}
+        {%- elif array_type == 'array' -%}
+            {%- set item_info = get_array_typescript(None, items_info, depth + 1) -%}
+            {%- if not param_name -%}
+                {{ "\n" + item_info + "[]" }}
+            {%- else -%}
+                {{ "\n" + offset + param_name + ": " + item_info|trim + "[]" }}
+            {%- endif -%}
+        {%- else -%}
+            {%- if 'enum' in items_info -%}
+                {%- set item_type = get_enum_option_str(items_info['enum']) -%}
+                {%- if param_name is none -%}
+                    {{ "(" + item_type + ")[]"}}
+                {%- else -%}
+                    {{ "\n" + offset + param_name + ": (" + item_type + ")[]" }}
+                {%- endif -%}
+            {%- else -%}
+                {%- if param_name is none -%}
+                    {{ "\n" + array_type + "[]" }}
+                {%- else -%}
+                    {{ "\n" + offset + param_name + ": " + array_type + "[]," }}
+                {%- endif -%}
+            {%- endif -%}
+        {%- endif -%}
+    {%- endif -%}
+{%- endmacro -%}
+
+{%- macro get_parameter_typescript(properties, required_params, depth=0) -%}
+    {%- set res = "" -%}
+    {%- for param_name, param in properties.items() -%}
+        {%- if param is mapping -%}
+            {%- set comment_info = get_param_info(param) -%}
+            {# Param Examples #}
+            {%- set examples_info = [] -%}
+            {%- if "examples" in param -%}
+                {%- set examples_info = ["Example " + param_name + ":"] -%}
+                {%- set examples_info = examples_info + param["examples"] -%}
+            {%- endif -%}
+
+            {# Param Name declaration #}
+            {%- set param_declaration = param_name -%}
+            {%- if required_params is iterable and param_name not in required_params -%}
+                {%- set param_declaration = param_declaration + "?" -%}
+            {%- endif -%}
+
+            {%- set param_type = get_param_type(param) -%}
+
+            {# Handle indentation based on depth #}
+            {%- set offset = "" -%}
+            {%- if depth >= 1 -%}
+                {%- set offset = "    " * depth -%}
+            {%- endif -%}
+
+            {%- if param_type == "object" -%}
+                {%- if comment_info != "<|NONE|>" -%}
+                    {{ "\n" + offset + comment_info }}
+                {%- endif -%}
+                {%- if examples_info|length > 0 -%}
+                    {%- for example in examples_info -%}
+                        {{ "\n" + offset + "// " + example|string|replace("'", '"') }}
+                    {%- endfor -%}
+                {%- endif -%}
+                {%- set param_declaration = param_declaration + ": {" -%}
+                {{ "\n" + offset + param_declaration -}}
+                {{- get_parameter_typescript(param.get("properties", {}), param.get("required", []), depth + 1) -}}
+                {{- "\n" + offset + "}," }}
+            {%- elif param_type == "array" -%}
+                {%- set item_info = param.get("items", {}) -%}
+                {%- if "type" not in item_info -%}
+                    {%- set param_declaration = param_declaration + ": []," -%}
+                    {{ append_new_param_info(param_declaration, comment_info, examples_info, depth) }}
+                {%- else -%}
+                    {%- if comment_info != "<|NONE|>" -%}
+                        {{ "\n" + offset + comment_info }}
+                    {%- endif -%}
+                    {%- if examples_info|length > 0 -%}
+                        {%- for example in examples_info -%}
+                            {{ "\n" + offset + "// " + example|string|replace("'", '"') }}
+                        {%- endfor -%}
+                    {%- endif -%}
+                    {%- set array_declaration = get_array_typescript(param_declaration, param, depth) -%}
+                    {%- if not array_declaration.endswith(",") -%}
+                        {%- set array_declaration = array_declaration + "," -%}
+                    {%- endif -%}
+                    {{ array_declaration}}
+                {%- endif -%}
+            {%- else -%}
+                {%- if "enum" in param -%}
+                    {%- set param_type = get_enum_option_str(param["enum"]) -%}
+                {%- endif -%}
+                {%- if "nullable" in param and param["nullable"] -%}
+                    {%- set param_type = param_type + " | null" -%}
+                {%- endif -%}
+                {%- set param_declaration = param_declaration + ": " + param_type + "," -%}
+                {{ append_new_param_info(param_declaration, comment_info, examples_info, depth) }}
+            {%- endif -%}
+        {%- endif -%}
+    {%- endfor -%}
+{%- endmacro -%}
+
+{%- macro generate_schema_from_functions(functions, namespace='functions') -%}
+    {{ "// Supported function definitions that should be called when necessary.\n" -}}
+    {{- "namespace " + namespace + " {\n\n" -}}
+
+    {%- for function in functions -%}
+        {%- if function.get("function") -%}
+            {%- set function = function.get("function") -%}
+        {%- endif -%}
+
+        {%- set function_name = function.get("name") -%}
+        {%- if function_name -%}
+            {%- set description = function.get('description', '') -%}
+            {%- set parameters = function.get('parameters', {}) -%}
+            {{- "// " + description + "\n" -}}
+            {{- "type " + function_name -}}
+            {%- if parameters and parameters.get("properties") -%}
+                {{- " = (_: {" -}}
+                {%- set required_params = parameters.get("required", []) -%}
+                {{ get_parameter_typescript(parameters.get("properties"), required_params, 0) -}}
+                {{- "\n}) => any;\n\n" }}
+            {%- else -%}
+                {{ " = () => any;\n\n" }}
+            {%- endif -%}
+        {%- endif -%}
+    {%- endfor -%}
+    {{ "} // namespace " + namespace }}
+{%- endmacro -%}
+{%- if not tools is defined -%}
+    {%- set tools = none -%}
+{%- endif -%}
+
+{%- set has_code_interpreter = tools | selectattr("type", "equalto", "code_interpreter") | list | length > 0 -%}
+{%- if has_code_interpreter -%}
+    {%- set tools = tools | rejectattr("type", "equalto", "code_interpreter") | list -%}
+{%- endif -%}
+
+{#- System message + builtin tools #}
+{{- bos_token + "<|start_header_id|>system<|end_header_id|>\n\n" }}
+{%- if has_code_interpreter %}
+    {{- "Environment: ipython\n\n" }}
+{%- else -%}
+    {{ "\n"}}
+{%- endif %}
+{%- if tools %}
+    {{- "\nYou have access to the following functions:\n\n" }}
+    {%- for t in tools %}
+        {%- if "type" in t -%}
+            {{ "Use the function '" + t["function"]["name"] + "' to '" + t["function"]["description"] + "'\n" + t["function"] | tojson() }}
+        {%- else -%}
+            {{ "Use the function '" + t["name"] + "' to '" + t["description"] + "'\n" + t | tojson }}
+        {%- endif -%}
+        {{- "\n\n" }}
+    {%- endfor %}
+    {{- '\nThink very carefully before calling functions.\nIf a you choose to call a function ONLY reply in the following format:\n<{start_tag}={function_name}>{parameters}{end_tag}\nwhere\n\nstart_tag => `<function`\nparameters => a JSON dict with the function argument name as key and function argument value as value.\nend_tag => `</function>`\n\nHere is an example,\n<function=example_function_name>{"example_name": "example_value"}</function>\n\nReminder:\n- If looking for real time information use relevant functions before falling back to brave_search\n- Function calls MUST follow the specified format, start with <function= and end with </function>\n- Required parameters MUST be specified\n- Only call one function at a time\n- Put the entire function call reply on one line\n\n' -}}
+{%- endif %}
+{{- "<|eot_id|>" -}}
+
+{%- for message in messages -%}
+    {%- if message['role'] == 'user' or message['role'] == 'system' -%}
+        {{ '<|start_header_id|>' + message['role'] + '<|end_header_id|>\n\n' + message['content'] + '<|eot_id|>' }}
+    {%- elif message['role'] == 'tool' -%}
+        {{ '<|start_header_id|>ipython<|end_header_id|>\n\n' + message['content'] + '<|eot_id|>' }}
+    {%- else -%}
+        {%- if (message['content'] and message['content']|length > 0) or ('tool_calls' in message and message['tool_calls'] and message['tool_calls']|length > 0) -%}
+            {{ '<|start_header_id|>' + message['role'] + '<|end_header_id|>\n\n'}}
+        {%- endif -%}
+        {%- if message['content'] and message['content']|length > 0 -%}
+            {{ message['content'] }}
+        {%- endif -%}
+        {%- if 'tool_calls' in message and message['tool_calls'] and message['tool_calls']|length > 0 -%}
+            {%- for tool_call in message['tool_calls'] -%}
+                {%- if tool_call["function"]["name"] == "python" -%}
+                    {{ '<|python_tag|>' + tool_call['function']['arguments']|tojson }}
+                {%- else -%}
+                    {{ '<function=' + tool_call['function']['name'] + '>' + tool_call['function']['arguments']|tojson + '</function>' }}
+                {%- endif -%}
+            {%- endfor -%}
+            {{ '<|eom_id|>' }}
+        {%- elif message['content'] and message['content']|length > 0 -%}
+            {{ '<|eot_id|>' }}
+        {%- endif -%}
+    {%- endif -%}
+{%- endfor -%}
+{%- if add_generation_prompt -%}
+    {{ '<|start_header_id|>assistant<|end_header_id|>\n\n' }}
+{%- endif -%}