kiln_ai.synthetic_user.role_swap

Role-swap eval-frame conversation roles into LLM-frame roles.

Chat models are trained to generate assistant-labeled responses, so we flip the eval-frame labels before the call: the LLM's "assistant" turn IS the synthetic user.

Eval frame:                       LLM frame (post-swap):
  user      = synthetic user        assistant = synthetic user
  assistant = target agent          user      = target agent

The driver keeps only user and assistant turns upstream, so a system or tool turn reaching here is an internal invariant violation — fail loud rather than silently drop.

 1"""Role-swap eval-frame conversation roles into LLM-frame roles.
 2
 3Chat models are trained to generate `assistant`-labeled responses, so we
 4flip the eval-frame labels before the call: the LLM's "assistant" turn IS
 5the synthetic user.
 6
 7    Eval frame:                       LLM frame (post-swap):
 8      user      = synthetic user        assistant = synthetic user
 9      assistant = target agent          user      = target agent
10
11The driver keeps only user and assistant turns upstream, so a system or tool
12turn reaching here is an internal invariant violation — fail loud rather
13than silently drop.
14"""
15
16from kiln_ai.utils.open_ai_types import (
17    ChatCompletionAssistantMessageParamWrapper,
18    ChatCompletionMessageParam,
19    ChatCompletionUserMessageParam,
20)
21
22
23def role_swap(
24    conversation: list[ChatCompletionMessageParam],
25) -> list[ChatCompletionMessageParam]:
26    """Flip eval-frame user/assistant labels into LLM-frame labels.
27
28    Only `user` and `assistant` are handled. The driver is expected to
29    have filtered out other roles before calling this.
30    """
31    result: list[ChatCompletionMessageParam] = []
32    for msg in conversation:
33        role = msg["role"]
34        if role not in ("user", "assistant"):
35            raise ValueError(
36                f"role_swap received unsupported role {role!r}; "
37                "the driver should have filtered it"
38            )
39        # The TypedDict union allows non-string content for multimodal /
40        # tool turns, but the synthetic user only ever sees plain text from
41        # the target. Narrowing here lets us assign into the swapped wrapper
42        # type without a cast.
43        content = msg["content"]
44        if not isinstance(content, str):
45            raise ValueError(
46                f"role_swap requires string content for role {role!r}; "
47                f"got {type(content).__name__}"
48            )
49        if role == "user":
50            assistant_msg: ChatCompletionAssistantMessageParamWrapper = {
51                "role": "assistant",
52                "content": content,
53            }
54            result.append(assistant_msg)
55        else:  # role == "assistant"
56            user_msg: ChatCompletionUserMessageParam = {
57                "role": "user",
58                "content": content,
59            }
60            result.append(user_msg)
61    return result
def role_swap( conversation: list[typing.Union[openai.types.chat.chat_completion_developer_message_param.ChatCompletionDeveloperMessageParam, openai.types.chat.chat_completion_system_message_param.ChatCompletionSystemMessageParam, openai.types.chat.chat_completion_user_message_param.ChatCompletionUserMessageParam, kiln_ai.utils.open_ai_types.ChatCompletionAssistantMessageParamWrapper, kiln_ai.utils.open_ai_types.ChatCompletionToolMessageParamWrapper, openai.types.chat.chat_completion_function_message_param.ChatCompletionFunctionMessageParam]]) -> list[typing.Union[openai.types.chat.chat_completion_developer_message_param.ChatCompletionDeveloperMessageParam, openai.types.chat.chat_completion_system_message_param.ChatCompletionSystemMessageParam, openai.types.chat.chat_completion_user_message_param.ChatCompletionUserMessageParam, kiln_ai.utils.open_ai_types.ChatCompletionAssistantMessageParamWrapper, kiln_ai.utils.open_ai_types.ChatCompletionToolMessageParamWrapper, openai.types.chat.chat_completion_function_message_param.ChatCompletionFunctionMessageParam]]: 
24def role_swap(
25    conversation: list[ChatCompletionMessageParam],
26) -> list[ChatCompletionMessageParam]:
27    """Flip eval-frame user/assistant labels into LLM-frame labels.
28
29    Only `user` and `assistant` are handled. The driver is expected to
30    have filtered out other roles before calling this.
31    """
32    result: list[ChatCompletionMessageParam] = []
33    for msg in conversation:
34        role = msg["role"]
35        if role not in ("user", "assistant"):
36            raise ValueError(
37                f"role_swap received unsupported role {role!r}; "
38                "the driver should have filtered it"
39            )
40        # The TypedDict union allows non-string content for multimodal /
41        # tool turns, but the synthetic user only ever sees plain text from
42        # the target. Narrowing here lets us assign into the swapped wrapper
43        # type without a cast.
44        content = msg["content"]
45        if not isinstance(content, str):
46            raise ValueError(
47                f"role_swap requires string content for role {role!r}; "
48                f"got {type(content).__name__}"
49            )
50        if role == "user":
51            assistant_msg: ChatCompletionAssistantMessageParamWrapper = {
52                "role": "assistant",
53                "content": content,
54            }
55            result.append(assistant_msg)
56        else:  # role == "assistant"
57            user_msg: ChatCompletionUserMessageParam = {
58                "role": "user",
59                "content": content,
60            }
61            result.append(user_msg)
62    return result

Flip eval-frame user/assistant labels into LLM-frame labels.

Only user and assistant are handled. The driver is expected to have filtered out other roles before calling this.