SimulationEvals¶

@cleanup_session_dir
def simulate_conversation(
    self,
    test_case: dict[str, Any],
    sim_user_model: str | None = _DEFAULT_GEMINI_MODEL,
    eval_model: str | None = _DEFAULT_GEMINI_MODEL,
    session_id: str | None = None,
    console_logging: bool = True,
    modality: str = "text",
    capture_agent_audio: bool = False,
    background_noise_file: str | None = None,
    burst_noise_files: list[str] | None = None,
    use_tool_fakes: bool = False,
    voice_config: dict[str, Any] | None = None,
    initial_utterance: str = _FIRST_UTTERANCE,
    skip_playback_wait: bool = False,
    single_bidi_stream: bool = False,
    **kwargs: Any,
) -> LLMUserConversation:
    """Runs the simulated conversation loop.

    Args:
        test_case: The test case dictionary defining evaluation steps.
        sim_user_model: The Gemini model used for the simulated user.
        eval_model: The Gemini model used for evaluating expectations.
        console_logging: Whether to print interaction transcript to
            the console.
        single_bidi_stream: For audio modality, keep one persistent
            bidi WebSocket open for the whole conversation instead of
            opening a new connection per turn (the default).
    """
    sim_user_model = sim_user_model or _DEFAULT_GEMINI_MODEL
    eval_model = eval_model or _DEFAULT_GEMINI_MODEL
    if session_id is None:
        session_id = str(uuid.uuid4())
    voice_config = voice_config or test_case.get("voice_config")
    eval_conv = LLMUserConversation(
        genai_client=self.genai_client,
        genai_model=sim_user_model,
        test_case=test_case,
        initial_utterance=initial_utterance,
    )

    # Initialize audio paths tracking
    eval_conv.agent_audio_paths = {}
    current_sim_turn = 0

    interactive_session = None
    if modality == "audio" and single_bidi_stream:
        client = self.sessions_client
        interactive_session = client.create_interactive_session(
            session_id=session_id,
            capture_agent_audio=capture_agent_audio,
            background_noise_file=background_noise_file,
            use_tool_fakes=use_tool_fakes,
            skip_playback_wait=skip_playback_wait,
            voice_config=voice_config,
        )
        interactive_session.start()

    try:
        if console_logging:
            print(
                f"Starting simulated conversation with session ID: "
                f"{session_id}"
            )

        # Initialize the first turn manually
        user_utterance, variables = eval_conv.next_user_utterance()
        accumulated_variables = {}
        if variables:
            accumulated_variables.update(variables)

        detailed_trace = []
        detailed_trace.append(f"User: {user_utterance}")

        while user_utterance:
            if modality == "audio" and interactive_session:
                response = interactive_session.send_turn(
                    user_utterance,
                    accumulated_variables,
                )
                # Check if session ended via WebSocket endSession
                if isinstance(response, dict) and response.get(
                    "session_ended"
                ):
                    if response.get("connection_error"):
                        err_msg = (
                            f"Interactive session WebSocket error: "
                            f"{response['connection_error']}"
                        )
                        raise BidiSessionError(err_msg)
                    break
            else:
                response = self._send_request_with_retry(
                    session_id=session_id,
                    user_utterance=user_utterance,
                    variables=accumulated_variables,
                    modality=modality,
                    console_logging=console_logging,
                    turn_num=current_sim_turn,
                    capture_agent_audio=capture_agent_audio,
                    background_noise_file=background_noise_file,
                    burst_noise_files=burst_noise_files,
                    use_tool_fakes=use_tool_fakes,
                    voice_config=voice_config,
                )
            if not response:
                break

            # Extract and save the agent turn audio WAV if present
            # in response.
            if response and getattr(response, "agent_audio_paths", None):
                audio_path = response.agent_audio_paths.get(0)
                if audio_path:
                    paths = eval_conv.agent_audio_paths
                    paths[current_sim_turn] = audio_path

            if console_logging:
                self.sessions_client.parse_result(response)

            agent_text, trace_chunks, session_ended, tool_calls = (
                self._parse_agent_response(response)
            )
            detailed_trace.append("\n".join(trace_chunks))

            if session_ended:
                if agent_text:
                    eval_conv._add_agent_response(agent_text)
                eval_conv._add_agent_tool_calls(tool_calls)
                # Ensure the final agent response is evaluated
                # so that steps_progress is updated on session end.
                eval_conv._next_user_utterance()
                if console_logging:
                    print(
                        "\nSession has been closed by the Agent via "
                        "end_session tool."
                    )
                # Mark current step as completed if the session ending
                # is a valid success (escalation evals)
                for prog in eval_conv.steps_progress:
                    criteria = prog.step.success_criteria.lower()
                    if prog.status != StepStatus.COMPLETED and (
                        "escalat" in criteria
                        or "transfer" in criteria
                        or "being transferred" in criteria
                    ):
                        prog.status = StepStatus.COMPLETED
                        prog.justification = (
                            "Agent ended session via escalation/transfer — "
                            "matches success criteria."
                        )
                break

            # Get the next simulated user utterance based on the agent's
            # response
            eval_conv._add_agent_tool_calls(tool_calls)
            user_utterance, variables = eval_conv.next_user_utterance(
                agent_text
            )
            if variables:
                accumulated_variables.update(variables)
            if user_utterance:
                detailed_trace.append(f"User: {user_utterance}")

            current_sim_turn += 1

        if console_logging:
            self._print_completion_status(eval_conv)

        self._evaluate_expectations(
            eval_conv,
            detailed_trace,
            eval_model,
            console_logging,
            capture_agent_audio=capture_agent_audio,
        )
        eval_conv._session_id = session_id
        eval_conv.session_id = session_id
        eval_conv._detailed_trace = detailed_trace
        eval_conv.detailed_trace = detailed_trace
        return eval_conv
    finally:
        if interactive_session:
            interactive_session.close()

def export_results_to_golden(
    self,
    results: list[dict[str, Any]],
    output_path: str | None = None,
) -> str:
    """Exports simulation results to a Golden Evaluation YAML file.

    Fetches the full conversation trace for each simulation from the
    platform to ensure accuracy.

    Args:
        results: The list of results returned by run_simulations.
        output_path: Optional local path to save the generated YAML.

    Returns:
        The generated YAML string.
    """
    conversations_list = []

    for res in results:
        turns = self._get_turns(res)
        if not turns:
            continue

        expectations = [
            e["expectation"] for e in res.get("expectation_details", [])
        ]
        params = res.get("session_parameters", {})

        conversations_list.append(
            GoldenConversation(
                conversation=res.get("name", "Simulated_Conversation"),
                turns=turns,
                expectations=expectations,
                session_parameters=params,
            )
        )

    dataset = GoldenConversations(conversations=conversations_list)
    yaml_content = yaml.dump(
        dataset.model_dump(exclude_none=True),
        sort_keys=False,
        allow_unicode=True,
    )

    if output_path:
        with open(output_path, "w", encoding="utf-8") as f:
            f.write(yaml_content)

    return yaml_content