Repository: LAION-AI/Open-Assistant
File: inference/worker/work.py (Python)

Code before:
def make_prompt_and_parameters(
    tokenizer: transformers.PreTrainedTokenizer,
    work_request: inference.WorkRequest,
) -> tuple[str, interface.GenerateStreamParameters]:
    """Prepare a formatted prompt and stream generation parameters based on a work request."""
    if settings.oa_protocol_version != "v2":
        raise RuntimeError(f"Unsupported oa protocol version: {settings.oa_protocol_version}")

    eos_token = tokenizer.eos_token if hasattr(tokenizer, "eos_token") else ""

    def _prepare_message(message: inference.MessageRead) -> str:
        prefix = V2_ASST_PREFIX if message.is_assistant else V2_PROMPTER_PREFIX
        return prefix + message.content + eos_token

    # Construct prompt
    messages = [_prepare_message(message) for message in work_request.thread.messages]

    # Prepend system prompt and custom_instructions if it was specified in work parameters
    if (
        work_request.parameters.system_prompt
        or work_request.parameters.user_profile
        or work_request.parameters.user_response_instructions
    ):
        work_params = work_request.parameters
        if work_params.system_prompt or work_params.user_profile or work_params.user_response_instructions:
            pre_prompt = V2_SYSTEM_PREFIX + (work_params.system_prompt or "")

        if work_params.user_profile or work_params.user_response_instructions:
            pre_prompt = f"""{pre_prompt}\n{CUSTOM_INSTRUCTIONS_PREFIX.format(user_profile=work_params.user_profile or "", user_response_instructions=work_params.user_response_instructions or "")}"""

        pre_prompt = pre_prompt + eos_token
        messages = [pre_prompt] + messages

    # Stringify and append assistant prefix to signify start of generation
    prompt = "".join(messages) + V2_ASST_PREFIX

    parameters = interface.GenerateStreamParameters.from_work_parameters(work_request.parameters)
    if settings.use_stop_sequences:
        parameters.stop = [
            V2_PROMPTER_PREFIX,
            V2_ASST_PREFIX,
            V2_SYSTEM_PREFIX,
        ]
        if eos_token:
            parameters.stop.append(eos_token)
    else:
        parameters.stop = []

    return prompt, parameters



Review comment (suggestion):
This can be simplified a bit (currently the same `if` condition is repeated)
```suggestion
    work_params = work_request.parameters
    if work_params.system_prompt or work_params.user_profile or work_params.user_response_instructions:
        pre_prompt = V2_SYSTEM_PREFIX + (work_params.system_prompt or "")
```

Code after:
def make_prompt_and_parameters(
    tokenizer: transformers.PreTrainedTokenizer,
    work_request: inference.WorkRequest,
) -> tuple[str, interface.GenerateStreamParameters]:
    """Prepare a formatted prompt and stream generation parameters based on a work request."""
    if settings.oa_protocol_version != "v2":
        raise RuntimeError(f"Unsupported oa protocol version: {settings.oa_protocol_version}")

    eos_token = tokenizer.eos_token if hasattr(tokenizer, "eos_token") else ""

    def _prepare_message(message: inference.MessageRead) -> str:
        prefix = V2_ASST_PREFIX if message.is_assistant else V2_PROMPTER_PREFIX
        return prefix + message.content + eos_token

    # Construct prompt
    messages = [_prepare_message(message) for message in work_request.thread.messages]

    # Prepend system prompt and custom_instructions if it was specified in work parameters
    work_params = work_request.parameters
    if work_params.system_prompt or work_params.user_profile or work_params.user_response_instructions:
        pre_prompt = V2_SYSTEM_PREFIX + (work_params.system_prompt or "")

        if work_params.user_profile or work_params.user_response_instructions:
            pre_prompt = f"""{pre_prompt}\n{CUSTOM_INSTRUCTIONS_PREFIX.format(user_profile=work_params.user_profile or "", user_response_instructions=work_params.user_response_instructions or "")}"""

        pre_prompt = pre_prompt + eos_token
        messages = [pre_prompt] + messages

    # Stringify and append assistant prefix to signify start of generation
    prompt = "".join(messages) + V2_ASST_PREFIX

    parameters = interface.GenerateStreamParameters.from_work_parameters(work_request.parameters)
    if settings.use_stop_sequences:
        parameters.stop = [
            V2_PROMPTER_PREFIX,
            V2_ASST_PREFIX,
            V2_SYSTEM_PREFIX,
        ]
        if eos_token:
            parameters.stop.append(eos_token)
    else:
        parameters.stop = []

    return prompt, parameters


def prepare_safe_prompt(prompt: str, label: str, rots: str) -> str:
    """Given a prompt, safety label, and safety rule of thumb, prepare a 'safe prompt' to replace the prompt."""
    pre_prompt = f"Answer the following request with {label} as responsible chatbot that believes that {rots}: "
    input_list = prompt.split(V2_PROMPTER_PREFIX)
    input_list[-1] = pre_prompt + input_list[-1]