wehub-resource-sync
994 lines
43 KiB
Python
994 lines
43 KiB
Python
import json
|
|
from copy import deepcopy
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from bfcl_eval.constants.category_mapping import VERSION_PREFIX
|
|
from bfcl_eval.constants.default_prompts import (
|
|
DEFAULT_USER_PROMPT_FOR_ADDITIONAL_FUNCTION_FC,
|
|
DEFAULT_USER_PROMPT_FOR_ADDITIONAL_FUNCTION_PROMPTING,
|
|
MAXIMUM_STEP_LIMIT,
|
|
)
|
|
from bfcl_eval.constants.enums import ModelStyle, ReturnFormat
|
|
from bfcl_eval.constants.eval_config import RESULT_PATH
|
|
from bfcl_eval.constants.executable_backend_config import (
|
|
OMIT_STATE_INFO_CLASSES,
|
|
STATELESS_CLASSES,
|
|
)
|
|
from bfcl_eval.eval_checker.multi_turn_eval.multi_turn_utils import (
|
|
execute_multi_turn_func_call,
|
|
is_empty_execute_response,
|
|
)
|
|
from bfcl_eval.model_handler.utils import add_memory_instruction_system_prompt
|
|
from bfcl_eval.utils import *
|
|
from overrides import final
|
|
|
|
if TYPE_CHECKING:
|
|
from bfcl_eval.eval_checker.multi_turn_eval.func_source_code.memory_api_metaclass import (
|
|
MemoryAPI,
|
|
)
|
|
|
|
|
|
class BaseHandler:
|
|
model_name: str
|
|
is_fc_model: bool
|
|
registry_name: str
|
|
temperature: float
|
|
registry_dir_name: str
|
|
model_name_underline_replaced: str
|
|
model_style: ModelStyle
|
|
|
|
def __init__(
|
|
self, model_name, temperature, registry_name, is_fc_model, **kwargs
|
|
) -> None:
|
|
"""
|
|
Args:
|
|
model_name: The name of the model as used in the vendor API or on Hugging Face.
|
|
temperature: The temperature of the model.
|
|
registry_name: The name of the model as used internally in BFCL, used for result directory naming.
|
|
is_fc_model: Whether the model is a function calling model.
|
|
**kwargs: Additional attributes passed via kwargs.
|
|
"""
|
|
self.model_name = model_name
|
|
self.is_fc_model = is_fc_model
|
|
self.registry_name = registry_name
|
|
|
|
# Replace the dash and dot with underscore for valid variable name
|
|
self.model_name_underline_replaced = (
|
|
model_name.replace("/", "_").replace("-", "_").replace(".", "_")
|
|
)
|
|
# The directory name for the model
|
|
# Replace the slash with underscore to avoid creating subdirectories
|
|
self.registry_dir_name = registry_name.replace("/", "_")
|
|
self.temperature = temperature
|
|
|
|
# Set any additional attributes passed via kwargs
|
|
for _key, _value in kwargs.items():
|
|
setattr(self, _key, _value)
|
|
|
|
def inference(
|
|
self,
|
|
test_entry: dict,
|
|
include_input_log: bool,
|
|
exclude_state_log: bool,
|
|
):
|
|
# This method is used to retrive model response for each model.
|
|
|
|
# FC model
|
|
# TODO: Let all models have the is_fc_model attribute and remove the "FC" check
|
|
if "FC" in self.registry_name or self.is_fc_model:
|
|
if contain_multi_turn_interaction(test_entry["id"]):
|
|
return self.inference_multi_turn_FC(
|
|
test_entry, include_input_log, exclude_state_log
|
|
)
|
|
else:
|
|
return self.inference_single_turn_FC(test_entry, include_input_log)
|
|
# Prompting model
|
|
else:
|
|
if contain_multi_turn_interaction(test_entry["id"]):
|
|
return self.inference_multi_turn_prompting(
|
|
test_entry, include_input_log, exclude_state_log
|
|
)
|
|
else:
|
|
return self.inference_single_turn_prompting(test_entry, include_input_log)
|
|
|
|
@final
|
|
def inference_multi_turn_FC(
|
|
self,
|
|
test_entry: dict,
|
|
include_input_log: bool,
|
|
exclude_state_log: bool,
|
|
) -> tuple[list[list], dict]:
|
|
initial_config: dict = test_entry.get("initial_config", {})
|
|
involved_classes: list = test_entry["involved_classes"]
|
|
test_entry_id: str = test_entry["id"]
|
|
test_category: str = test_entry_id.rsplit("_", 1)[0]
|
|
|
|
# This is only for the miss function category
|
|
# A mapping from turn index to function to holdout
|
|
holdout_function: dict[int, list] = test_entry.get("missed_function", {})
|
|
|
|
total_input_token_count: list[list[float]] = []
|
|
total_output_token_count: list[list[float]] = []
|
|
total_latency: list[list[float]] = []
|
|
all_model_response: list[list] = (
|
|
[]
|
|
) # The model response that will be used for later evaluation
|
|
all_inference_log: list[list[dict]] = (
|
|
[]
|
|
) # The debugging log for human to understand
|
|
force_quit = False # Whether the model has been forced to quit. If True, this whole entry will be failed.
|
|
|
|
all_reasoning_content: list[list] = []
|
|
|
|
# Execute no function call, but just to get a reference to all the instances to get the initial state for logging purpose
|
|
_, involved_instances = execute_multi_turn_func_call(
|
|
[],
|
|
initial_config,
|
|
involved_classes,
|
|
self.model_name_underline_replaced,
|
|
test_entry_id,
|
|
long_context=("long_context" in test_category or "composite" in test_category),
|
|
is_evaL_run=False,
|
|
)
|
|
|
|
if is_memory(test_category):
|
|
assert (
|
|
len(involved_instances) == 1
|
|
), "Memory category should only involve one class."
|
|
|
|
memory_instance: "MemoryAPI" = list(involved_instances.values())[0]
|
|
test_entry["question"] = add_memory_instruction_system_prompt(
|
|
test_entry["question"],
|
|
test_category,
|
|
test_entry["scenario"],
|
|
memory_instance,
|
|
)
|
|
|
|
if not exclude_state_log:
|
|
state_log = []
|
|
for class_name, class_instance in involved_instances.items():
|
|
if class_name in STATELESS_CLASSES or class_name in OMIT_STATE_INFO_CLASSES:
|
|
continue
|
|
# Avoid modification in future turns
|
|
class_instance = deepcopy(class_instance)
|
|
state_log.append(
|
|
{
|
|
"role": "state_info",
|
|
"class_name": class_name,
|
|
"content": {
|
|
key: value
|
|
for key, value in vars(class_instance).items()
|
|
if not key.startswith("_")
|
|
},
|
|
}
|
|
)
|
|
if len(state_log) > 0:
|
|
all_inference_log.append(state_log)
|
|
|
|
inference_data: dict = {}
|
|
inference_data = self._pre_query_processing_FC(inference_data, test_entry)
|
|
inference_data = self._compile_tools(inference_data, test_entry)
|
|
|
|
all_multi_turn_messages: list[list[dict]] = test_entry["question"]
|
|
for turn_idx, current_turn_message in enumerate(all_multi_turn_messages):
|
|
current_turn_message: list[dict]
|
|
|
|
if str(turn_idx) in holdout_function:
|
|
test_entry["function"].extend(holdout_function[str(turn_idx)])
|
|
# Since we have added new functions, we need to recompile the tools
|
|
inference_data = self._compile_tools(inference_data, test_entry)
|
|
assert (
|
|
len(current_turn_message) == 0
|
|
), "Holdout turn should not have user message."
|
|
# TODO: Move this to before pre_query_processing_FC.
|
|
# Shouldn't be happening in the inference loop.
|
|
current_turn_message = [
|
|
{
|
|
"role": "user",
|
|
"content": DEFAULT_USER_PROMPT_FOR_ADDITIONAL_FUNCTION_FC,
|
|
}
|
|
]
|
|
|
|
if turn_idx == 0:
|
|
inference_data = self.add_first_turn_message_FC(
|
|
inference_data, current_turn_message
|
|
)
|
|
else:
|
|
inference_data = self._add_next_turn_user_message_FC(
|
|
inference_data, current_turn_message
|
|
)
|
|
|
|
current_turn_response = []
|
|
current_turn_inference_log: list[dict] = {
|
|
"begin_of_turn_query": current_turn_message
|
|
}
|
|
current_turn_input_token_count: list[float] = []
|
|
current_turn_output_token_count: list[float] = []
|
|
current_turn_latency: list[float] = []
|
|
current_turn_reasoning_content = []
|
|
|
|
count = 0
|
|
while True:
|
|
print("-" * 100)
|
|
print(
|
|
f"ID: {test_entry_id.replace('multi_turn_', '')}, Turn: {turn_idx}, Step: {count}"
|
|
)
|
|
current_step_inference_log: list[dict] = []
|
|
# Add to the current_turn_inference_log at beginning of each step so that we don't need to bother dealing with the break statements
|
|
current_turn_inference_log[f"step_{count}"] = current_step_inference_log
|
|
|
|
api_response, query_latency = self._query_FC(inference_data)
|
|
|
|
# This part of logging is disabled by default because it is too verbose and will make the result file extremely large
|
|
# It is only useful to see if the inference pipeline is working as expected (eg, does it convert all the inputs correctly)
|
|
if include_input_log:
|
|
current_step_inference_log.append(
|
|
{
|
|
"role": "inference_input",
|
|
"content": inference_data.get("inference_input_log", ""),
|
|
}
|
|
)
|
|
|
|
# Try parsing the model response
|
|
model_response_data = self._parse_query_response_FC(api_response)
|
|
model_responses = model_response_data["model_responses"]
|
|
|
|
# Add the assistant message to the chat history
|
|
inference_data = self._add_assistant_message_FC(
|
|
inference_data, model_response_data
|
|
)
|
|
|
|
# Process the metadata
|
|
current_turn_input_token_count.append(model_response_data["input_token"])
|
|
current_turn_output_token_count.append(model_response_data["output_token"])
|
|
current_turn_latency.append(query_latency)
|
|
|
|
current_turn_response.append(model_responses)
|
|
|
|
reasoning_content = model_response_data.get("reasoning_content", "")
|
|
current_turn_reasoning_content.append(reasoning_content)
|
|
|
|
log_entry = {
|
|
"role": "assistant",
|
|
"content": model_responses,
|
|
}
|
|
if reasoning_content:
|
|
log_entry["reasoning_content"] = reasoning_content
|
|
|
|
current_step_inference_log.append(log_entry)
|
|
|
|
# Try decoding the model response
|
|
try:
|
|
decoded_model_responses = self.decode_execute(
|
|
model_responses, has_tool_call_tag=False
|
|
)
|
|
current_step_inference_log.append(
|
|
{
|
|
"role": "handler_log",
|
|
"content": "Successfully decoded model response.",
|
|
"model_response_decoded": decoded_model_responses,
|
|
}
|
|
)
|
|
|
|
if is_empty_execute_response(decoded_model_responses):
|
|
print("Empty response from the model. Proceed to next turn.")
|
|
current_step_inference_log.append(
|
|
{
|
|
"role": "handler_log",
|
|
"content": f"Empty response from the model. Proceed to next turn.",
|
|
"model_response_decoded": decoded_model_responses,
|
|
}
|
|
)
|
|
break
|
|
|
|
except Exception as e:
|
|
print("Failed to decode the model response. Proceed to next turn.")
|
|
current_step_inference_log.append(
|
|
{
|
|
"role": "handler_log",
|
|
"content": f"Error decoding the model response. Proceed to next turn.",
|
|
"error": str(e),
|
|
}
|
|
)
|
|
break
|
|
|
|
# Obtain the execution results
|
|
execution_results, involved_instances = execute_multi_turn_func_call(
|
|
decoded_model_responses,
|
|
initial_config,
|
|
involved_classes,
|
|
self.model_name_underline_replaced,
|
|
test_entry_id,
|
|
long_context=(
|
|
"long_context" in test_category or "composite" in test_category
|
|
),
|
|
is_evaL_run=False,
|
|
)
|
|
|
|
# Add the execution results to the chat history for the next turn
|
|
inference_data = self._add_execution_results_FC(
|
|
inference_data, execution_results, model_response_data
|
|
)
|
|
|
|
for execution_result in execution_results:
|
|
current_step_inference_log.append(
|
|
{
|
|
"role": "tool",
|
|
"content": execution_result,
|
|
}
|
|
)
|
|
|
|
count += 1
|
|
# Force quit after too many steps
|
|
if count > MAXIMUM_STEP_LIMIT:
|
|
force_quit = True
|
|
current_step_inference_log.append(
|
|
{
|
|
"role": "handler_log",
|
|
"content": f"Model has been forced to quit after {MAXIMUM_STEP_LIMIT} steps.",
|
|
}
|
|
)
|
|
|
|
break
|
|
|
|
# Add to the total list
|
|
all_model_response.append(current_turn_response)
|
|
all_inference_log.append(current_turn_inference_log)
|
|
all_reasoning_content.append(current_turn_reasoning_content)
|
|
total_input_token_count.append(current_turn_input_token_count)
|
|
total_output_token_count.append(current_turn_output_token_count)
|
|
total_latency.append(current_turn_latency)
|
|
|
|
if not exclude_state_log:
|
|
state_log = []
|
|
for class_name, class_instance in involved_instances.items():
|
|
if (
|
|
class_name in STATELESS_CLASSES
|
|
or class_name in OMIT_STATE_INFO_CLASSES
|
|
):
|
|
continue
|
|
# Avoid modification in future turns
|
|
class_instance = deepcopy(class_instance)
|
|
state_log.append(
|
|
{
|
|
"role": "state_info",
|
|
"class_name": class_name,
|
|
"content": {
|
|
key: value
|
|
for key, value in vars(class_instance).items()
|
|
if not key.startswith("_")
|
|
},
|
|
}
|
|
)
|
|
if len(state_log) > 0:
|
|
all_inference_log.append(state_log)
|
|
|
|
if force_quit:
|
|
break
|
|
|
|
# Special handling for the memory category
|
|
# Need to flush the memory to local file at the end of the conversation
|
|
if is_memory_prereq(test_entry_id):
|
|
assert (
|
|
len(involved_instances) == 1
|
|
), "Memory category should only involve one class."
|
|
memory_instance: "MemoryAPI" = list(involved_instances.values())[0]
|
|
memory_instance._flush_memory_to_local_file()
|
|
|
|
metadata = {
|
|
"input_token_count": total_input_token_count,
|
|
"output_token_count": total_output_token_count,
|
|
"latency": total_latency,
|
|
"inference_log": all_inference_log,
|
|
}
|
|
|
|
if not all(
|
|
all(content == "" for content in single_turn_reasoning_content)
|
|
for single_turn_reasoning_content in all_reasoning_content
|
|
):
|
|
metadata["reasoning_content"] = all_reasoning_content
|
|
|
|
return all_model_response, metadata
|
|
|
|
@final
|
|
def inference_multi_turn_prompting(
|
|
self,
|
|
test_entry: dict,
|
|
include_input_log: bool,
|
|
exclude_state_log: bool,
|
|
) -> tuple[list[list], dict]:
|
|
initial_config: dict = test_entry.get("initial_config", {})
|
|
involved_classes: list = test_entry["involved_classes"]
|
|
test_entry_id: str = test_entry["id"]
|
|
test_category: str = test_entry_id.rsplit("_", 1)[0]
|
|
|
|
# This is only for the miss function category
|
|
# A mapping from turn index to function to holdout
|
|
holdout_function: dict[int, list] = test_entry.get("missed_function", {})
|
|
|
|
total_input_token_count: list[list[float]] = []
|
|
total_output_token_count: list[list[float]] = []
|
|
total_latency: list[list[float]] = []
|
|
# The model response that will be used for later evaluation
|
|
all_model_response: list[list] = []
|
|
# Only for reasoning models, reasoning content will be stored as part of metadata and in inference log
|
|
all_reasoning_content: list[list] = []
|
|
# The debugging log for human to understand
|
|
all_inference_log: list[list[dict]] = []
|
|
force_quit = False # Whether the model has been forced to quit. If True, this whole entry will be failed.
|
|
|
|
# Execute no function call, but just to get a reference to all the instances to get the initial state for logging purpose
|
|
_, involved_instances = execute_multi_turn_func_call(
|
|
[],
|
|
initial_config,
|
|
involved_classes,
|
|
self.model_name_underline_replaced,
|
|
test_entry_id,
|
|
long_context=("long_context" in test_category or "composite" in test_category),
|
|
is_evaL_run=False,
|
|
)
|
|
|
|
if is_memory(test_category):
|
|
assert (
|
|
len(involved_instances) == 1
|
|
), "Memory category should only involve one class."
|
|
|
|
memory_instance: "MemoryAPI" = list(involved_instances.values())[0]
|
|
test_entry["question"] = add_memory_instruction_system_prompt(
|
|
test_entry["question"],
|
|
test_category,
|
|
test_entry["scenario"],
|
|
memory_instance,
|
|
)
|
|
|
|
if not exclude_state_log:
|
|
state_log = []
|
|
for class_name, class_instance in involved_instances.items():
|
|
if class_name in STATELESS_CLASSES or class_name in OMIT_STATE_INFO_CLASSES:
|
|
continue
|
|
# Avoid modification in future turns
|
|
class_instance = deepcopy(class_instance)
|
|
state_log.append(
|
|
{
|
|
"role": "state_info",
|
|
"class_name": class_name,
|
|
"content": {
|
|
key: value
|
|
for key, value in vars(class_instance).items()
|
|
if not key.startswith("_")
|
|
},
|
|
}
|
|
)
|
|
if len(state_log) > 0:
|
|
all_inference_log.append(state_log)
|
|
|
|
inference_data: dict = self._pre_query_processing_prompting(test_entry)
|
|
|
|
all_multi_turn_messages: list[list[dict]] = test_entry["question"]
|
|
for turn_idx, current_turn_message in enumerate(all_multi_turn_messages):
|
|
current_turn_message: list[dict]
|
|
|
|
if str(turn_idx) in holdout_function:
|
|
assert (
|
|
len(current_turn_message) == 0
|
|
), "Holdout turn should not have user message."
|
|
current_turn_message = [
|
|
{
|
|
"role": "user",
|
|
"content": DEFAULT_USER_PROMPT_FOR_ADDITIONAL_FUNCTION_PROMPTING.format(
|
|
functions=holdout_function[str(turn_idx)]
|
|
),
|
|
}
|
|
]
|
|
|
|
if turn_idx == 0:
|
|
inference_data = self.add_first_turn_message_prompting(
|
|
inference_data, current_turn_message
|
|
)
|
|
else:
|
|
inference_data = self._add_next_turn_user_message_prompting(
|
|
inference_data, current_turn_message
|
|
)
|
|
|
|
current_turn_response = []
|
|
current_turn_reasoning_content = []
|
|
current_turn_inference_log: list[dict] = {
|
|
"begin_of_turn_query": current_turn_message
|
|
}
|
|
current_turn_input_token_count: list[float] = []
|
|
current_turn_output_token_count: list[float] = []
|
|
current_turn_latency: list[float] = []
|
|
|
|
count = 0
|
|
while True:
|
|
print("-" * 100)
|
|
print(
|
|
f"ID: {test_entry_id.replace('multi_turn_', '')}, Turn: {turn_idx}, Step: {count}"
|
|
)
|
|
current_step_inference_log: list[dict] = []
|
|
# Add to the current_turn_inference_log at beginning of each step so that we don't need to bother dealing with the break statements
|
|
current_turn_inference_log[f"step_{count}"] = current_step_inference_log
|
|
|
|
api_response, query_latency = self._query_prompting(inference_data)
|
|
|
|
# This part of logging is disabled by default because it is too verbose and will make the result file extremely large
|
|
# It is only useful to see if the inference pipeline is working as expected (eg, does it convert all the inputs correctly)
|
|
if include_input_log:
|
|
current_step_inference_log.append(
|
|
{
|
|
"role": "inference_input",
|
|
"content": inference_data.get("inference_input_log", ""),
|
|
}
|
|
)
|
|
|
|
# Try parsing the model response
|
|
model_response_data = self._parse_query_response_prompting(api_response)
|
|
model_responses = model_response_data["model_responses"]
|
|
|
|
# Add the assistant message to the chat history
|
|
inference_data = self._add_assistant_message_prompting(
|
|
inference_data, model_response_data
|
|
)
|
|
|
|
# Process the metadata
|
|
current_turn_input_token_count.append(model_response_data["input_token"])
|
|
current_turn_output_token_count.append(model_response_data["output_token"])
|
|
current_turn_latency.append(query_latency)
|
|
|
|
current_turn_response.append(model_responses)
|
|
reasoning_content = model_response_data.get("reasoning_content", "")
|
|
current_turn_reasoning_content.append(reasoning_content)
|
|
|
|
log_entry = {
|
|
"role": "assistant",
|
|
"content": model_responses,
|
|
}
|
|
if reasoning_content:
|
|
log_entry["reasoning_content"] = reasoning_content
|
|
|
|
current_step_inference_log.append(log_entry)
|
|
|
|
# Try decoding the model response
|
|
try:
|
|
decoded_model_responses = self.decode_execute(
|
|
model_responses, has_tool_call_tag=False
|
|
)
|
|
current_step_inference_log.append(
|
|
{
|
|
"role": "handler_log",
|
|
"content": "Successfully decoded model response.",
|
|
"model_response_decoded": decoded_model_responses,
|
|
}
|
|
)
|
|
|
|
model_response_data["model_responses_decoded"] = decoded_model_responses
|
|
if is_empty_execute_response(decoded_model_responses):
|
|
print("Empty response from the model. Proceed to next turn.")
|
|
current_step_inference_log.append(
|
|
{
|
|
"role": "handler_log",
|
|
"content": f"Empty response from the model. Proceed to next turn.",
|
|
"model_response_decoded": decoded_model_responses,
|
|
}
|
|
)
|
|
break
|
|
|
|
except Exception as e:
|
|
print("Failed to decode the model response. Proceed to next turn.")
|
|
current_step_inference_log.append(
|
|
{
|
|
"role": "handler_log",
|
|
"content": f"Error decoding the model response. Proceed to next turn.",
|
|
"error": str(e),
|
|
}
|
|
)
|
|
break
|
|
|
|
# Obtain the execution results
|
|
execution_results, involved_instances = execute_multi_turn_func_call(
|
|
decoded_model_responses,
|
|
initial_config,
|
|
involved_classes,
|
|
self.model_name_underline_replaced,
|
|
test_entry_id,
|
|
long_context=(
|
|
"long_context" in test_category or "composite" in test_category
|
|
),
|
|
is_evaL_run=False,
|
|
)
|
|
|
|
# Add the execution results to the chat history for the next turn
|
|
inference_data = self._add_execution_results_prompting(
|
|
inference_data, execution_results, model_response_data
|
|
)
|
|
|
|
for execution_result in execution_results:
|
|
current_step_inference_log.append(
|
|
{
|
|
"role": "tool",
|
|
"content": execution_result,
|
|
}
|
|
)
|
|
|
|
count += 1
|
|
# Force quit after too many steps
|
|
if count > MAXIMUM_STEP_LIMIT:
|
|
force_quit = True
|
|
current_step_inference_log.append(
|
|
{
|
|
"role": "handler_log",
|
|
"content": f"Model has been forced to quit after {MAXIMUM_STEP_LIMIT} steps.",
|
|
}
|
|
)
|
|
break
|
|
|
|
# Add to the total list
|
|
all_model_response.append(current_turn_response)
|
|
all_reasoning_content.append(current_turn_reasoning_content)
|
|
all_inference_log.append(current_turn_inference_log)
|
|
total_input_token_count.append(current_turn_input_token_count)
|
|
total_output_token_count.append(current_turn_output_token_count)
|
|
total_latency.append(current_turn_latency)
|
|
|
|
if not exclude_state_log:
|
|
state_log = []
|
|
for class_name, class_instance in involved_instances.items():
|
|
if (
|
|
class_name in STATELESS_CLASSES
|
|
or class_name in OMIT_STATE_INFO_CLASSES
|
|
):
|
|
continue
|
|
# Avoid modification in future turns
|
|
class_instance = deepcopy(class_instance)
|
|
state_log.append(
|
|
{
|
|
"role": "state_info",
|
|
"class_name": class_name,
|
|
"content": {
|
|
key: value
|
|
for key, value in vars(class_instance).items()
|
|
if not key.startswith("_")
|
|
},
|
|
}
|
|
)
|
|
if len(state_log) > 0:
|
|
all_inference_log.append(state_log)
|
|
|
|
if force_quit:
|
|
break
|
|
|
|
# Special handling for the memory category
|
|
# Need to flush the memory to local file at the end of the conversation
|
|
if is_memory_prereq(test_entry_id):
|
|
assert (
|
|
len(involved_instances) == 1
|
|
), "Memory category should only involve one class."
|
|
memory_instance: "MemoryAPI" = list(involved_instances.values())[0]
|
|
memory_instance._flush_memory_to_local_file()
|
|
|
|
metadata = {
|
|
"input_token_count": total_input_token_count,
|
|
"output_token_count": total_output_token_count,
|
|
"latency": total_latency,
|
|
"inference_log": all_inference_log,
|
|
}
|
|
# We only include reasoning content if it exists and is not empty
|
|
if not all(
|
|
all(content == "" for content in single_turn_reasoning_content)
|
|
for single_turn_reasoning_content in all_reasoning_content
|
|
):
|
|
metadata["reasoning_content"] = all_reasoning_content
|
|
|
|
return all_model_response, metadata
|
|
|
|
@final
|
|
def inference_single_turn_FC(
|
|
self, test_entry: dict, include_input_log: bool
|
|
) -> tuple[any, dict]:
|
|
inference_data: dict = {}
|
|
inference_data = self._pre_query_processing_FC(inference_data, test_entry)
|
|
inference_data = self._compile_tools(inference_data, test_entry)
|
|
inference_data = self.add_first_turn_message_FC(
|
|
inference_data, test_entry["question"][0]
|
|
)
|
|
|
|
api_response, query_latency = self._query_FC(inference_data)
|
|
|
|
# Try parsing the model response
|
|
model_response_data = self._parse_query_response_FC(api_response)
|
|
|
|
# Process the metadata
|
|
metadata = {}
|
|
if include_input_log:
|
|
metadata["inference_log"] = [
|
|
{
|
|
"role": "inference_input",
|
|
"content": inference_data.get("inference_input_log", ""),
|
|
}
|
|
]
|
|
metadata["input_token_count"] = model_response_data["input_token"]
|
|
metadata["output_token_count"] = model_response_data["output_token"]
|
|
metadata["latency"] = query_latency
|
|
|
|
if (
|
|
"reasoning_content" in model_response_data
|
|
and model_response_data["reasoning_content"] != ""
|
|
):
|
|
metadata["reasoning_content"] = model_response_data["reasoning_content"]
|
|
|
|
return model_response_data["model_responses"], metadata
|
|
|
|
@final
|
|
def inference_single_turn_prompting(
|
|
self, test_entry: dict, include_input_log: bool
|
|
) -> tuple[any, dict]:
|
|
inference_data: dict = self._pre_query_processing_prompting(test_entry)
|
|
inference_data = self.add_first_turn_message_prompting(
|
|
inference_data, test_entry["question"][0]
|
|
)
|
|
|
|
api_response, query_latency = self._query_prompting(inference_data)
|
|
|
|
# Try parsing the model response
|
|
model_response_data = self._parse_query_response_prompting(api_response)
|
|
|
|
# Process the metadata
|
|
metadata = {}
|
|
if include_input_log:
|
|
metadata["inference_log"] = [
|
|
{
|
|
"role": "inference_input",
|
|
"content": inference_data.get("inference_input_log", ""),
|
|
}
|
|
]
|
|
metadata["input_token_count"] = model_response_data["input_token"]
|
|
metadata["output_token_count"] = model_response_data["output_token"]
|
|
metadata["latency"] = query_latency
|
|
|
|
if (
|
|
"reasoning_content" in model_response_data
|
|
and model_response_data["reasoning_content"] != ""
|
|
):
|
|
metadata["reasoning_content"] = model_response_data["reasoning_content"]
|
|
|
|
return model_response_data["model_responses"], metadata
|
|
|
|
def decode_ast(self, result, language: ReturnFormat, has_tool_call_tag: bool):
|
|
"""
|
|
This method takes raw model output (from `_parse_query_response_xxx`) and convert it to standard AST checker input.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def decode_execute(self, result, has_tool_call_tag: bool):
|
|
"""
|
|
This method takes raw model output (from `_parse_query_response_xxx`) and convert it to standard execute checker input.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@final
|
|
def write(self, result, result_dir, update_mode=False):
|
|
# Use the internal registry name to decide the result directory to avoid
|
|
# collisions between different variants that share the same API model name.
|
|
model_result_dir = result_dir / self.registry_dir_name
|
|
|
|
if isinstance(result, dict):
|
|
result = [result]
|
|
|
|
# Collect and format each entry for JSON compatibility
|
|
entries_to_write = [make_json_serializable(entry) for entry in result]
|
|
|
|
# Group entries by their `test_category` for efficient file handling
|
|
file_entries = {}
|
|
for entry in entries_to_write:
|
|
test_category = extract_test_category_from_id(entry["id"])
|
|
# Determine the high-level grouping folder (non_live, live, etc.)
|
|
group_dir_name = get_directory_structure_by_id(entry["id"])
|
|
group_dir_path = model_result_dir / group_dir_name
|
|
group_dir_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
file_path = group_dir_path / f"{VERSION_PREFIX}_{test_category}_result.json"
|
|
file_entries.setdefault(file_path, []).append(entry)
|
|
|
|
for file_path, entries in file_entries.items():
|
|
if update_mode:
|
|
# Load existing entries from the file
|
|
existing_entries = {}
|
|
if file_path.exists():
|
|
existing_entries = {
|
|
entry["id"]: entry for entry in load_file(file_path)
|
|
}
|
|
|
|
# Update existing entries with new data
|
|
for entry in entries:
|
|
existing_entries[entry["id"]] = entry
|
|
|
|
# Sort entries by `id` and write them back to ensure order consistency
|
|
sorted_entries = sorted(existing_entries.values(), key=sort_key)
|
|
with open(file_path, "w") as f:
|
|
for entry in sorted_entries:
|
|
content = json.dumps(entry) + "\n"
|
|
f.write(content)
|
|
f.flush()
|
|
|
|
else:
|
|
# Normal mode: Append to the end of the file
|
|
# Note: We will sort all the entries at the end of the generation pipeline to ensure the order is consistent
|
|
entries.sort(key=sort_key)
|
|
with open(file_path, "a") as f:
|
|
for entry in entries:
|
|
content = json.dumps(entry) + "\n"
|
|
f.write(content)
|
|
f.flush()
|
|
|
|
#### FC methods ####
|
|
|
|
def _query_FC(self, inference_data: dict):
|
|
"""
|
|
Call the model API in FC mode to get the response.
|
|
Return the response object that can be used to feed into the `_parse_query_response_FC` method.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def _pre_query_processing_FC(self, inference_data: dict, test_entry: dict) -> dict:
|
|
"""
|
|
Preprocess the testset entry before sending it to the model.
|
|
This might includes transforming the input user message into the format expected by the model, extract out the system prompt (if any), and any other necessary preprocessing steps. Those steps can also be done in the `add_first_turn_message_FC` and `_add_next_turn_user_message_FC` methods, but it's usually cleaner to do it here.
|
|
The inference_data dict is updated in place and returned.
|
|
|
|
Note: This method has different signature from its Prompting version.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def _compile_tools(self, inference_data: dict, test_entry: dict) -> dict:
|
|
"""
|
|
[Only for FC mode]
|
|
This method is used to prepare/compile the tools from the test entry and add them to the inference data to use for model query in FC mode.
|
|
Function docs usually need to be transformed to the format expected by the model, done through the `convert_to_tool` function from `model_handler/utils.py`.
|
|
The inference_data dict is updated in place and returned.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def _parse_query_response_FC(self, api_response: Any) -> dict:
|
|
"""
|
|
Parses the raw response from the model API to extract the result, input token count, and output token count.
|
|
|
|
Args:
|
|
api_response (any): The raw response from the model API.
|
|
|
|
Returns:
|
|
A dict containing the following elements:
|
|
- model_responses (any): The parsed result that can be directly used as input to the decode method.
|
|
- input_token (int): The number of tokens used in the input to the model.
|
|
- output_token (int): The number of tokens generated by the model as output.
|
|
- tool_call_ids (list[str]): The IDs of the tool calls that are generated by the model. Optional.
|
|
- Any other metadata that is specific to the model.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def add_first_turn_message_FC(
|
|
self, inference_data: dict, first_turn_message: list[dict]
|
|
) -> dict:
|
|
"""
|
|
Add the first turn message to the chat history, in the format that the model expects.
|
|
|
|
Args:
|
|
inference_data (dict): The inference data from previous processing steps.
|
|
first_turn_message (list[dict]): The first turn message from the test entry. It has variable length. It might contain one or more of the following roles:
|
|
- "system": The system message. This role will only appear at most once, at the beginning of the first turn. For most entry, this role will not appear.
|
|
- "user": The user message.
|
|
- "assistant": The assistant message. For most entry, this role will not appear.
|
|
|
|
Returns:
|
|
inference_data (dict): The updated inference data that will be send to `_query_FC` to call the model API.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def _add_next_turn_user_message_FC(
|
|
self, inference_data: dict, user_message: list[dict]
|
|
) -> dict:
|
|
"""
|
|
[Only for multi-turn]
|
|
Add next turn user message to the chat history for query.
|
|
user_message is a list of 1 element, which is guaranteed to be a `user` role message.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def _add_assistant_message_FC(
|
|
self, inference_data: dict, model_response_data: dict
|
|
) -> dict:
|
|
"""
|
|
Add assistant message to the chat history.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def _add_execution_results_FC(
|
|
self, inference_data: dict, execution_results: list[str], model_response_data: dict
|
|
) -> dict:
|
|
"""
|
|
Add the execution results to the chat history to prepare for the next turn of query.
|
|
Some models may need to add additional information to the chat history, such as tool call IDs.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
#### Prompting methods ####
|
|
|
|
def _query_prompting(self, inference_data: dict):
|
|
"""
|
|
Call the model API in prompting mode to get the response.
|
|
Return the response object that can be used to feed into the decode method.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def _pre_query_processing_prompting(self, test_entry: dict) -> dict:
|
|
"""
|
|
Preprocess the testset entry before sending it to the model.
|
|
This might includes transforming the input user message into the format expected by the model, extract out the system prompt (if any), and any other necessary preprocessing steps. Those steps can also be done in the `add_first_turn_message_prompting` and `_add_next_turn_user_message_prompting` methods, but it's usually cleaner to do it here.
|
|
The function docs are usually supplied to the prompting models as part of the system prompt, done via the `system_prompt_pre_processing_chat_model` function from `model_handler/utils.py`, unless the model has a different way of handling it.
|
|
Returns a dict that contains all the necessary information for the query method.
|
|
Things like `system_prompt` and `chat_history` are optional, specific to the model.
|
|
|
|
Note: This method has different signature from its FC version.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def _parse_query_response_prompting(self, api_response: Any) -> dict:
|
|
"""
|
|
Parses the raw response from the model API to extract the result, input token count, and output token count.
|
|
|
|
Args:
|
|
api_response (any): The raw response from the model API.
|
|
|
|
Returns:
|
|
A dict containing the following elements:
|
|
- model_responses (any): The parsed result that can be directly used as input to the decode method.
|
|
- input_token (int): The number of tokens used in the input to the model.
|
|
- output_token (int): The number of tokens generated by the model as output.
|
|
- Any other metadata that is specific to the model.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def add_first_turn_message_prompting(
|
|
self, inference_data: dict, first_turn_message: list[dict]
|
|
) -> dict:
|
|
"""
|
|
Add the first turn message to the chat history, in the format that the model expects.
|
|
|
|
Args:
|
|
inference_data (dict): The inference data from previous processing steps.
|
|
first_turn_message (list[dict]): The first turn message from the test entry. It has variable length. It might contain one or more of the following roles:
|
|
- "system": The system message. This role will only appear at most once, at the beginning of the first turn.
|
|
- "user": The user message.
|
|
- "assistant": The assistant message. For most entry, this role will not appear.
|
|
|
|
Returns:
|
|
inference_data (dict): The updated inference data that will be send to `_query_prompting` to call the model API.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def _add_next_turn_user_message_prompting(
|
|
self, inference_data: dict, user_message: list[dict]
|
|
) -> dict:
|
|
"""
|
|
[Only for multi-turn]
|
|
Add next turn user message to the chat history for query.
|
|
user_message is a list of 1 element, which is guaranteed to be a `user` role message.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def _add_assistant_message_prompting(
|
|
self, inference_data: dict, model_response_data: dict
|
|
) -> dict:
|
|
"""
|
|
Add assistant message to the chat history.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def _add_execution_results_prompting(
|
|
self, inference_data: dict, execution_results: list[str], model_response_data: dict
|
|
) -> dict:
|
|
"""
|
|
Add the execution results to the chat history to prepare for the next turn of query.
|
|
By default, execution results are added back as a `user` role message, as most models don't support the `tool` role in prompting mode.
|
|
"""
|
|
raise NotImplementedError
|