wehub-resource-sync
70 lines
3.7 KiB
Python
70 lines
3.7 KiB
Python
from bfcl_eval.model_handler.local_inference.base_oss_handler import OSSHandler
|
|
from bfcl_eval.model_handler.utils import (
|
|
combine_consecutive_user_prompts,
|
|
system_prompt_pre_processing_chat_model,
|
|
)
|
|
from overrides import override
|
|
|
|
|
|
class GemmaHandler(OSSHandler):
|
|
def __init__(
|
|
self,
|
|
model_name,
|
|
temperature,
|
|
registry_name,
|
|
is_fc_model,
|
|
dtype="bfloat16",
|
|
**kwargs,
|
|
) -> None:
|
|
super().__init__(model_name, temperature, registry_name, is_fc_model, **kwargs)
|
|
|
|
@override
|
|
def _format_prompt(self, messages, function):
|
|
"""
|
|
"bos_token": "<bos>",
|
|
"chat_template": "{{ bos_token }}\n{%- if messages[0]['role'] == 'system' -%}\n {%- if messages[0]['content'] is string -%}\n {%- set first_user_prefix = messages[0]['content'] + '\n\n' -%}\n {%- else -%}\n {%- set first_user_prefix = messages[0]['content'][0]['text'] + '\n\n' -%}\n {%- endif -%}\n {%- set loop_messages = messages[1:] -%}\n{%- else -%}\n {%- set first_user_prefix = \"\" -%}\n {%- set loop_messages = messages -%}\n{%- endif -%}\n{%- for message in loop_messages -%}\n {%- if (message['role'] == 'user') != (loop.index0 % 2 == 0) -%}\n {{ raise_exception(\"Conversation roles must alternate user/assistant/user/assistant/...\") }}\n {%- endif -%}\n {%- if (message['role'] == 'assistant') -%}\n {%- set role = \"model\" -%}\n {%- else -%}\n {%- set role = message['role'] -%}\n {%- endif -%}\n {{ '<start_of_turn>' + role + '\n' + (first_user_prefix if loop.first else \"\") }}\n {%- if message['content'] is string -%}\n {{ message['content'] | trim }}\n {%- elif message['content'] is iterable -%}\n {%- for item in message['content'] -%}\n {%- if item['type'] == 'image' -%}\n {{ '<start_of_image>' }}\n {%- elif item['type'] == 'text' -%}\n {{ item['text'] | trim }}\n {%- endif -%}\n {%- endfor -%}\n {%- else -%}\n {{ raise_exception(\"Invalid content type\") }}\n {%- endif -%}\n {{ '<end_of_turn>\n' }}\n{%- endfor -%}\n{%- if add_generation_prompt -%}\n {{'<start_of_turn>model\n'}}\n{%- endif -%}\n",
|
|
"""
|
|
formatted_prompt = "<bos>"
|
|
|
|
if messages[0]["role"] == "system":
|
|
first_user_prefix = messages[0]["content"].strip() + "\n\n"
|
|
messages = messages[1:]
|
|
else:
|
|
first_user_prefix = ""
|
|
|
|
is_first = True
|
|
for message in messages:
|
|
formatted_prompt += f"<start_of_turn>{message['role']}\n{first_user_prefix if is_first else ''}{message['content'].strip()}<end_of_turn>\n"
|
|
is_first = False
|
|
|
|
formatted_prompt += f"<start_of_turn>model\n"
|
|
|
|
return formatted_prompt
|
|
|
|
@override
|
|
def _pre_query_processing_prompting(self, test_entry: dict) -> dict:
|
|
functions: list = test_entry["function"]
|
|
test_entry_id: str = test_entry["id"]
|
|
|
|
test_entry["question"][0] = system_prompt_pre_processing_chat_model(
|
|
test_entry["question"][0], functions, test_entry_id
|
|
)
|
|
|
|
for round_idx in range(len(test_entry["question"])):
|
|
test_entry["question"][round_idx] = combine_consecutive_user_prompts(
|
|
test_entry["question"][round_idx]
|
|
)
|
|
test_entry["question"][round_idx] = self._substitute_prompt_role(
|
|
test_entry["question"][round_idx]
|
|
)
|
|
|
|
return {"message": [], "function": functions}
|
|
|
|
@staticmethod
|
|
def _substitute_prompt_role(prompts: list[dict]) -> list[dict]:
|
|
for prompt in prompts:
|
|
if prompt["role"] == "assistant":
|
|
prompt["role"] = "model"
|
|
|
|
return prompts
|