import copy
import json
import logging
import peewee as pw
from playhouse.shortcuts import model_to_dict
from flexeval.classes.base import BaseModel
from flexeval.classes.dataset import Dataset
from flexeval.classes.thread import Thread
from flexeval.configuration import completion_functions
logger = logging.getLogger(__name__)
[docs]
class Turn(BaseModel):
"""Holds a single turn
In a conversational exchange, each 'Turn' holds information
from 1 or more outputs from the same source or role in sequence
"""
id = pw.IntegerField(primary_key=True)
dataset = pw.ForeignKeyField(Dataset, backref="turns")
thread = pw.ForeignKeyField(Thread, backref="turns")
index_in_thread = pw.IntegerField()
role = pw.TextField()
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.metrics_to_evaluate = []
[docs]
def get_completion(self, completion_config: dict | None = None, evalsetrun=None):
# only get a completion if this is the final turn - we probably don't want to branch from mid-conversation
if self.is_final_turn_in_input:
if completion_config is None:
raise ValueError(
"completion_config must be provided to get_completion()"
)
completion_fn_name = completion_config.get("function_name", None)
completion_function_kwargs = completion_config.get("kwargs", None)
# Check if the function name exists in the global namespace and call it
if hasattr(completion_functions, completion_fn_name) and hasattr(
completion_functions, completion_fn_name
):
completion_function = getattr(
completion_functions, completion_fn_name, None
)
completion = completion_function(
conversation_history=self.get_formatted_prompt(
include_system_prompt=False
),
**completion_function_kwargs,
)
else:
logger.warning(
"In completion_functions.py: No callable function named "
+ completion_fn_name
+ " found."
)
completion = None
# "completion" will be the output of an existing completion function
# We need to make the message object
# and probably also a turn object
# which means it'll have a structure like this
# TODO - make this a requirement of the completion functions?
# - make the completion function just return content?
# {"choices": [{"message": {"content": "hi", "role": "assistant"}}]}
result = model_to_dict(self, exclude=[self.id])
result["evalsetrun"] = evalsetrun
result["dataset"] = self.dataset
result["datasetrow"] = self.datasetrow
result["turn_number"] = self.turn_number + 1
result["role"] = "assistant"
result["context"] = self.get_formatted_prompt(include_system_prompt=False)
result["is_final_turn_in_input"] = False # b/c it's not in input
self.is_final_turn_in_input = False
result["is_completion"] = True
result["completion"] = completion
result["model"] = completion.get("model", None)
result["prompt_tokens"] = completion.get("usage", {}).get(
"prompt_tokens", None
) / len(completion.get("choices", [1]))
result["completion_tokens"] = completion.get("usage", {}).get(
"completion_tokens", None
) / len(
completion.get("choices", [1])
) # TODO - use tiktoken here instead?? this will just give the average
result_list = []
for ix, choice in enumerate(completion["choices"]):
temp = copy.deepcopy(result)
temp["tool_used"] = choice["message"].get("tool_calls", None)
temp["turn"] = [choice["message"]]
temp["content"] = choice["message"]["content"]
temp["completion_number"] = ix + 1
result_list.append(temp)
return result_list
else:
return None
[docs]
def get_context(self, include_system_prompt=False) -> list[dict[str, str]]:
"""
Context is the context of the first message in the turn
"""
context = ""
for message in self.messages:
# TODO why not just use message.get_context(include_system_prompt=include_system_prompt) here?
context = message.context
break
context = json.loads(context)
if not include_system_prompt:
context = [
cur_dict for cur_dict in context if cur_dict.get("role") != "system"
]
return context
[docs]
def get_content(self, include_toolcalls=True, include_tool_messages=True):
"""
Content is a list of dictionaries where each dictionary
contains the role and content of messages and tool calls
in the turn. Each tool call appears after the message it's
associated with. If toolcalls are not desired, pass False
to include_toolcalls.
"""
content = []
for message in self.messages:
if include_tool_messages or message.langgraph_message_type != "ToolMessage":
content.append({"role": message.role, "content": message.content})
if include_toolcalls:
for toolcall in message.toolcalls:
content.append(toolcall.get_dict_representation())
return content
