聊天机器人基准测试使用仿真¶

    import functools
    from typing import Annotated, Any, Callable, Dict, List, Optional, Union

    from langchain_community.adapters.openai import convert_message_to_dict
    from langchain_core.messages import AIMessage, AnyMessage, BaseMessage, HumanMessage
    from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
    from langchain_core.runnables import Runnable, RunnableLambda
    from langchain_core.runnables import chain as as_runnable
    from langchain_openai import ChatOpenAI
    from typing_extensions import TypedDict

    from langgraph.graph import END, StateGraph, START


    def langchain_to_openai_messages(messages: List[BaseMessage]):
        """
        将一组 langchain 基础消息转换为一组 openai 消息。

        参数:
            messages (List[BaseMessage]): 一组 langchain 基础消息。

        返回:
            List[dict]: 一组 openai 消息。
        """

        return [
            convert_message_to_dict(m) if isinstance(m, BaseMessage) else m
            for m in messages
        ]


    def create_simulated_user(
        system_prompt: str, llm: Runnable | None = None
    ) -> Runnable[Dict, AIMessage]:
        """
        创建一个用于聊天机器人模拟的模拟用户。

        参数:
            system_prompt (str): 模拟用户将使用的系统提示。
            llm (Runnable | None, 可选): 用于模拟的语言模型。
                默认为 gpt-3.5-turbo。

        返回:
            Runnable[Dict, AIMessage]: 用于聊天机器人模拟的模拟用户。
        """
        return ChatPromptTemplate.from_messages(
            [
                ("system", system_prompt),
                MessagesPlaceholder(variable_name="messages"),
            ]
        ) | (llm or ChatOpenAI(model="gpt-3.5-turbo")).with_config(
            run_name="simulated_user"
        )


    Messages = Union[list[AnyMessage], AnyMessage]


    def add_messages(left: Messages, right: Messages) -> Messages:
        if not isinstance(left, list):
            left = [left]
        if not isinstance(right, list):
            right = [right]
        return left + right


    class SimulationState(TypedDict):
        """
        表示模拟的状态。

        属性:
            messages (List[AnyMessage]): 模拟中的消息列表。
            inputs (Optional[dict[str, Any]]): 模拟的可选输入。
        """

        messages: Annotated[List[AnyMessage], add_messages]
        inputs: Optional[dict[str, Any]]


    def create_chat_simulator(
        assistant: (
            Callable[[List[AnyMessage]], str | AIMessage]
            | Runnable[List[AnyMessage], str | AIMessage]
        ),
        simulated_user: Runnable[Dict, AIMessage],
        *,
        input_key: str,
        max_turns: int = 6,
        should_continue: Optional[Callable[[SimulationState], str]] = None,
    ):
        """创建一个用于评估聊天机器人的聊天模拟器。

        参数:
            assistant: 聊天机器人助手函数或可运行对象。
            simulated_user: 模拟用户对象。
            input_key: 聊天模拟的输入键。
            max_turns: 聊天模拟的最大轮数，默认为 6。
            should_continue: 可选函数以确定模拟是否应继续。
                如果未提供，将使用默认函数。

        返回:
            编译的聊天模拟图。

        """
        graph_builder = StateGraph(SimulationState)
        graph_builder.add_node(
            "user",
            _create_simulated_user_node(simulated_user),
        )
        graph_builder.add_node(
            "assistant", _fetch_messages | assistant | _coerce_to_message
        )
        graph_builder.add_edge("assistant", "user")
        graph_builder.add_conditional_edges(
            "user",
            should_continue or functools.partial(_should_continue, max_turns=max_turns),
        )
        # 如果您的数据集中有一个“引导问题/输入”，则我们首先指向助手，否则，我们让用户先行。
        graph_builder.add_edge(START, "assistant" if input_key is not None else "user")

        return (
            RunnableLambda(_prepare_example).bind(input_key=input_key)
            | graph_builder.compile()
        )


    ## 私有方法


    def _prepare_example(inputs: dict[str, Any], input_key: Optional[str] = None):
        if input_key is not None:
            if input_key not in inputs:
                raise ValueError(
                    f"数据集的示例输入必须包含提供的输入键: '{input_key}'。\n找到: {list(inputs.keys())}"
                )
            messages = [HumanMessage(content=inputs[input_key])]
            return {
                "inputs": {k: v for k, v in inputs.items() if k != input_key},
                "messages": messages,
            }
        return {"inputs": inputs, "messages": []}


    def _invoke_simulated_user(state: SimulationState, simulated_user: Runnable):
        """调用模拟用户节点。"""
        runnable = (
            simulated_user
            if isinstance(simulated_user, Runnable)
            else RunnableLambda(simulated_user)
        )
        inputs = state.get("inputs", {})
        inputs["messages"] = state["messages"]
        return runnable.invoke(inputs)


    def _swap_roles(state: SimulationState):
        new_messages = []
        for m in state["messages"]:
            if isinstance(m, AIMessage):
                new_messages.append(HumanMessage(content=m.content))
            else:
                new_messages.append(AIMessage(content=m.content))
        return {
            "inputs": state.get("inputs", {}),
            "messages": new_messages,
        }


    @as_runnable
    def _fetch_messages(state: SimulationState):
        """调用模拟用户节点。"""
        return state["messages"]


    def _convert_to_human_message(message: BaseMessage):
        return {"messages": [HumanMessage(content=message.content)]}


    def _create_simulated_user_node(simulated_user: Runnable):
        """模拟用户接受一个 {"messages": [...]} 参数并返回一条消息。"""
        return (
            _swap_roles
            | RunnableLambda(_invoke_simulated_user).bind(simulated_user=simulated_user)
            | _convert_to_human_message
        )


    def _coerce_to_message(assistant_output: str | BaseMessage):
        if isinstance(assistant_output, str):
            return {"messages": [AIMessage(content=assistant_output)]}
        else:
            return {"messages": [assistant_output]}


    def _should_continue(state: SimulationState, max_turns: int = 6):
        messages = state["messages"]
        # TODO 支持其他停止标准
        if len(messages) > max_turns:
            return END
        elif messages[-1].content.strip() == "FINISHED":
            return END
        else:
            return "assistant"