被测对象：从一个模型到模型加脚手架

def chat_prompt_to_text_prompt(
    prompt: OpenAICreateChatPrompt,
    for_completion: bool = True,
    chat_to_prefixes: Optional[Dict] = None,
) -> str:
    """
    Render a chat prompt as a text prompt. User and assistant messages are separated by newlines
    and prefixed with "User: " and "Assistant: ", respectively, unless there is only one message.
    System messages have no prefix.
    """
    assert is_chat_prompt(prompt), f"Expected a chat prompt, got {prompt}"
    if chat_to_prefixes is None:
        chat_to_prefixes = {
            # roles
            "system": "",
            # names
            "example_user": "User: ",
            "example_assistant": "Assistant: ",
        }

    # For a single message, be it system, user, or assistant, just return the message
    if len(prompt) == 1:
        return prompt[0]["content"]

    text = ""
    for msg in prompt:
        role = msg["name"] if "name" in msg else msg["role"]
        prefix = chat_to_prefixes.get(role, role.capitalize() + ": ")
        content = msg["content"]
        text += f"{prefix}{content}\n"
    if for_completion:
        text += chat_to_prefixes.get(
            "assistant", "Assistant: "
        ).rstrip()  # rstrip to remove trailing whitespace
    return text.lstrip()


def text_prompt_to_chat_prompt(prompt: str, role: str = "system") -> OpenAICreateChatPrompt:
    assert isinstance(prompt, str), f"Expected a text prompt, got {prompt}"
    return [
        {"role": role, "content": prompt},
    ]

class CompletionResult(ABC):
    @abstractmethod
    def get_completions(self) -> list[str]:
        pass


@runtime_checkable
class CompletionFn(Protocol):
    def __call__(
        self,
        prompt: Union[str, OpenAICreateChatPrompt],
        **kwargs,
    ) -> CompletionResult:
        """
        ARGS
        ====
        `prompt`: Either a `Prompt` object or a raw prompt that will get wrapped in
            the appropriate `Prompt` class.
        `kwargs`: Other arguments passed to the API.

        RETURNS
        =======
        The result of the API call.
        The prompt that was fed into the API call as a str.
        """

    def make_completion_fn(
        self,
        name: str,
        **kwargs: Any,
    ) -> CompletionFn:
        """
        Create a CompletionFn. The name can be one of the following formats:
        1. openai-model-id (e.g. "gpt-3.5-turbo")
        2. completion-fn-id (from the registry)
        """
        if name == "dummy":
            return DummyCompletionFn()

        n_ctx = n_ctx_from_model_name(name)

        if is_chat_model(name):
            return OpenAIChatCompletionFn(model=name, n_ctx=n_ctx, **kwargs)
        elif name in self.api_model_ids:
            return OpenAICompletionFn(model=name, n_ctx=n_ctx, **kwargs)

        # No match, so try to find a completion-fn-id in the registry
        spec = self.get_completion_fn(name) or self.get_solver(name)
        if spec is None:
            raise ValueError(f"Could not find CompletionFn/Solver in the registry with ID {name}")
        if spec.args is None:
            spec.args = {}
        spec.args.update(kwargs)

        spec.args["registry"] = self
        instance = make_object(spec.cls)(**spec.args or {})
        assert isinstance(instance, CompletionFn), f"{name} must be a CompletionFn"
        return instance

    def __call__(
        self,
        prompt: Union[str, OpenAICreateChatPrompt],
        **kwargs,
    ) -> OpenAIChatCompletionResult:
        if not isinstance(prompt, Prompt):
            assert (
                isinstance(prompt, str)
                or (isinstance(prompt, list) and all(isinstance(token, int) for token in prompt))
                or (isinstance(prompt, list) and all(isinstance(token, str) for token in prompt))
                or (isinstance(prompt, list) and all(isinstance(msg, dict) for msg in prompt))
            ), f"Got type {type(prompt)}, with val {type(prompt[0])} for prompt, expected str or list[int] or list[str] or list[dict[str, str]]"

            prompt = ChatCompletionPrompt(
                raw_prompt=prompt,
            )

        openai_create_prompt: OpenAICreateChatPrompt = prompt.to_formatted_prompt()

        result = openai_chat_completion_create_retrying(
            OpenAI(api_key=self.api_key, base_url=self.api_base),
            model=self.model,
            messages=openai_create_prompt,
            **{**kwargs, **self.extra_options},
        )
        result = OpenAIChatCompletionResult(raw_data=result, prompt=openai_create_prompt)
        record_sampling(
            prompt=result.prompt,
            sampled=result.get_completions(),
            model=result.raw_data.model,
            usage=result.raw_data.usage,
        )
        return result

        prompt = sample["input"]
        if self.num_few_shot > 0:
            assert is_chat_prompt(sample["input"]), "few shot requires chat prompt"
            prompt = sample["input"][:-1]
            for s in self.few_shot[: self.num_few_shot]:
                prompt += s["sample"]
            prompt += sample["input"][-1:]

@dataclass
class TaskState:
    """
    TaskState is the object provided from an Eval to a Solver. This must
    contain all the information that a Solver needs to provide a response to
    the Eval environment.

    Args:
        task_description: A string describing the task, including instructions
            and the expected response. Fixed across all instances of the eval.
        messages: The list of messages in the conversation so far. For example,
            it is often useful to include an input sample as the first message.
            Any previous interactions should also be included here.
        current_state: Any relevant state variables that should be passed to
            the Solver. While the current state of the eval should be apparent
            from previous messages, it is sometimes useful to include explicit
            state information here (e.g. the current game score or number of
            turns remaining) for easy access by the Solver without having to
            parse the messages.
    """

    task_description: str
    messages: list[Message] = field(default_factory=list)
    current_state: Any = None

    @abstractmethod
    def _solve(
        self,
        task_state: TaskState,
        **kwargs,
    ) -> SolverResult:
        """
        ARGS
        ====
        `task_state`: A `TaskState` object that contains the task description and the input.
        `kwargs`: Other arguments passed to the solver.

        RETURNS
        =======
        The result of the solver.
        """

    def __call__(
        self,
        task_state: TaskState,
        **kwargs,
    ) -> SolverResult:
        """Deepcopies task_state to prevent solvers from modifying the original object."""
        res = self._solve(deepcopy(task_state), **kwargs)

        if hasattr(self, "postprocessors"):
            # Iteratively apply postprocessors to the output
            for postprocessor in self.postprocessors:
                prev_output = res.output
                res = postprocessor(res)
                record_event(
                    "postprocessor",
                    {
                        "name": postprocessor.__class__.__name__,
                        "input": prev_output,
                        "output": res.output,
                    },
                )
        return res

    def _solve(self, task_state: TaskState, **kwargs) -> SolverResult:
        raw_msgs = [
            {"role": "system", "content": task_state.task_description},
        ] + [msg.to_dict() for msg in task_state.messages]

        precheck_outcome = self._perform_prechecks(raw_msgs)
        if precheck_outcome is not None:
            return precheck_outcome

        msgs = self._process_msgs(raw_msgs)

        try:
            if self._is_chat_model(self.model):
                completion_result = self.completion_fn(prompt=msgs, **kwargs)

                completion_output = completion_result.get_completions()[0]

                # Chat model output is already parsed, just return it
                solver_result = SolverResult(
                    completion_output, raw_completion_result=completion_result
                )
            else:
                # Manually render the prompt for completion models so that we can
                # implement things like custom render formats and/or fixed_start
                prompt = self._render_completion_prompt(msgs)

                stop_sequences = self._get_msg_separators()
                if len(stop_sequences) > 4:
                    logging.warn("Using more than 4 stop sequences is unsupported")
                completion_result = self.completion_fn(prompt=prompt, stop=stop_sequences, **kwargs)

                completion_output = completion_result.get_completions()[0]

                # Completion model output needs to be parsed to remove role prefixes
                solver_result = SolverResult(
                    self._parse_completion_response(completion_output),
                    raw_output=completion_output,
                    raw_completion_result=completion_result,
                )
        except self._completion_exception as e:
            solver_result = self._handle_completion_exception(e)

        return solver_result

def maybe_wrap_with_solver(ambiguous_executor: Union[Solver, CompletionFn]) -> Solver:
    """
    Converts a basic completion_fn into a Solver if it isn't already one.
    If it is already a Solver, it is returned unchanged.
    """

    if isinstance(ambiguous_executor, Solver):
        # Use the solver directly
        solver = ambiguous_executor
    elif isinstance(ambiguous_executor, SolverCompletionFn):
        # unwrap previously wrapped solver
        solver = ambiguous_executor.solver
    else:
        # Wrap the completion_fn in an appropriate solver for its type
        if isinstance(ambiguous_executor, OpenAIChatCompletionFn) or isinstance(
            ambiguous_executor, OpenAICompletionFn
        ):
            solver = OpenAISolver(
                completion_fn_options={
                    "model": ambiguous_executor.model,
                }
            )
            solver.completion_fn = ambiguous_executor
        elif isinstance(ambiguous_executor, DummyCompletionFn):
            solver = DummySolver()
        else:
            raise ValueError(f"Unsupported completion_fn type: {type(ambiguous_executor)}")
    return solver

class NestedSolver(Solver):
    """An abstract solver class that receives specification of any number of other solvers as an argument."""

    # TODO: Should we allow nested solvers to (also) take Solver classes instead of SolverSpecs?

    def __init__(self, *, postprocessors: list[str] = [], registry=None, **solver_specs):
        super().__init__(postprocessors=postprocessors)
        self.solver_specs = {}
        self._solver_cache = {}

        SolverSpecValidator = TypeAdapter(SolverSpec)
        for name, value in solver_specs.items():
            try:
                SolverSpecValidator.validate_python(value)
                self.solver_specs[name] = value
                self.get_solver(name)  # Initialize the solver
            except ValidationError:
                raise ValueError(f"Expected a sub-solver spec at '{name}', got '{value}'")

        assert (
            self.solver_specs
        ), f"{type(self).__name__} requires at least one sub-solver as an argument"

    def get_solver(self, solver_name: str) -> Solver:
        """
        IMPORTANT: All subclasses of NestedSolver should use this method to reference any
        sub-solvers, otherwise solver copies will not work properly.

        For convenience, your subclass can have a @property method like this:
        ```python
        @property
        def my_sub_solver(self) -> Solver:
            return self.get_solver("my_sub_solver")
        ```
        which is used in the _solve method like this:
        ```python
        def _solve(
            self,
            task_state: TaskState,
            **kwargs,
        ) -> SolverResult:
            ...
            solver_result = self.my_sub_solver(task_state=task_state, **kwargs)
            ...
        ```
        """
        if solver_name not in self._solver_cache:
            solver_spec = self.solver_specs[solver_name]
            self._solver_cache[solver_name] = self._create_solver(solver_spec)
        return self._solver_cache[solver_name]

DEFAULT_COT_TEMPLATE = "Please reason in a step-by-step manner before giving a response. (You now have an opportunity to reason privately; your next response will not be evaluated.)"
DEFAULT_EXTRACT_ANSWER_TEMPLATE = (
    "Given the above reasoning, your response in the format requested by the instructions is:"
)

    def _solve(
        self,
        task_state: TaskState,
        **kwargs,
    ) -> SolverResult:
        if self.interaction_cache:
            # Add in previous (private) memories
            task_state.messages = self.interaction_cache.load_private_interaction(task_state)

        # Reasoning step
        task_state.messages.append(Message(role="system", content=self.cot_template(task_state)))
        reasoning_result = self.cot_solver(task_state=task_state, **kwargs)
        reasoning_output = reasoning_result.output

        # Extract answer step
        task_state.messages.append(Message(role="assistant", content=reasoning_output))
        task_state.messages.append(
            Message(role="system", content=self.extract_template(task_state))
        )
        extracted_result = self.extract_solver(task_state=task_state, **kwargs)
        extracted_answer = extracted_result.output

        task_state.messages.append(Message(role="assistant", content=extracted_answer))

        # Save the interaction
        if self.interaction_cache:
            self.interaction_cache.save_private_interaction(task_state)

        return SolverResult(
            output=extracted_answer,
            reasoning_output=reasoning_output,
        )

        for i in range(self.num_generations):
            raw_result = self.solver(task_state)

            # We don't immediately append this to the task state messages because
            # doing so would influence subsequent reasonings, which we do not want.
            reasoning_completions.append(raw_result.output)
            try:
                answer = self._extract_answer(raw_result)
                unique_answers[answer] += 1
            except ValueError as ve:
                logger.info(f"ValueError while extracting answer: {ve}")
                continue

    def _modify_task_state(self, task_state: TaskState) -> TaskState:
        assert all(
            user_message not in task_state.messages
            for (user_message, assistant_message) in self.train_data
        ), f"The few-shot training data provided contains the current test set point: {task_state.messages}. Check datasets for contamination."

class Strip(PostProcessor):
    """
    Strip leading and trailing whitespace from the output, including newlines.
    """

    def __call__(self, result: SolverResult) -> SolverResult:
        return SolverResult(
            result.output.strip(),
            **result.metadata,
        )


class RemoveQuotes(PostProcessor):
    """
    Remove quotes from the beginning and end of the output. This works only if:
    - The quotes are exactly at the beginning and end (if there is a space
      between the quote and the first/last character, the quote is not removed)
    - There is a matching pair of quotes (if there is only one quote at either
      end, it is not removed)
    """

    def __call__(self, result: SolverResult) -> SolverResult:
        if len(result.output) >= 2:
            if result.output[0] == '"' and result.output[-1] == '"':
                result._output = result.output[1:-1]
            elif result.output[0] == "'" and result.output[-1] == "'":
                result._output = result.output[1:-1]
        return result


class RemovePeriod(PostProcessor):
    """
    Remove a period from the end of the output. The period must be exactly the
    last character in the output or it will not be removed.
    """

    def __call__(self, result: SolverResult) -> SolverResult:
        result._output = result.output.rstrip(".")
        return result

    @staticmethod
    def _convert_msgs_to_anthropic_format(msgs: list[Message]) -> list[MessageParam]:
        """
        Anthropic API requires that the message list has
        - Roles as 'user' or 'assistant'
        - Alternating 'user' and 'assistant' messages

        Note: the top-level system prompt is handled separately and should not be
        included in the messages list.
        """
        # enforce valid roles; convert to Anthropic message type
        anth_msgs = [
            MessageParam(
                role=oai_to_anthropic_role[msg.role],
                content=[ContentBlock(text=msg.content, type="text")],
            )
            for msg in msgs
        ]
        # enforce alternating roles by merging consecutive messages with the same role
        # e.g. [user1, user2, assistant1, user3] -> [user12, assistant1, user3]
        alt_msgs = []
        for msg in anth_msgs:
            if len(alt_msgs) > 0 and msg["role"] == alt_msgs[-1]["role"]:
                # Merge consecutive messages from the same role
                alt_msgs[-1]["content"].extend(msg["content"])
            else:
                alt_msgs.append(msg)

        return alt_msgs

    def _make_logit_bias(self, valid_answers: list[str], model: str) -> dict[int, float]:
        enc = tiktoken.encoding_for_model(model)
        token_ids = []
        for answer in valid_answers:
            encoded_answer = enc.encode(answer)
            if len(encoded_answer) > 1:
                raise ValueError(
                    f"Answer {answer} was encoded to {encoded_answer}, but we expected a single token."
                )
            token_ids.append(encoded_answer[0])
        return {token_id: 100 for token_id in token_ids}

    def _solve(self, task_state: TaskState, **kwargs) -> SolverResult:
        msgs = [Message("system", task_state.task_description)]
        msgs += task_state.messages

        prompt = (
            "\n".join([f"{msg.role}: {msg.content}" for msg in msgs]) + f"\n{self.input_prompt}"
        )
        answer = input(prompt)

        record_sampling(
            prompt=prompt,
            sampled=answer,
            model="human",
        )

        return SolverResult(answer)