SimulationEvals¶

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",
    background_noise_file: str | None = None,
    burst_noise_files: list[str] | None = None,
    use_tool_fakes: bool = False,
) -> 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.
    """
    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())
    eval_conv = LLMUserConversation(
        genai_client=self.genai_client,
        genai_model=sim_user_model,
        test_case=test_case,
    )

    if console_logging:
        print(
            f"Starting simulated conversation with session ID: {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:
        response = self._send_request_with_retry(
            session_id,
            user_utterance,
            accumulated_variables,
            modality,
            console_logging,
            background_noise_file,
            burst_noise_files,
            use_tool_fakes=use_tool_fakes,
        )
        if not response:
            break

        if console_logging:
            self.sessions_client.parse_result(response)

        agent_text, trace_chunks, session_ended = (
            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)
            # 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."
                )
            break

        # Get the next simulated user utterance based on the agent's
        # response
        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}")

    if console_logging:
        self._print_completion_status(eval_conv)

    self._evaluate_expectations(
        eval_conv, detailed_trace, eval_model, console_logging
    )
    eval_conv.detailed_trace = detailed_trace
    return eval_conv

def run_simulations(
    self,
    test_cases: list[dict[str, Any]],
    runs: int = 1,
    parallel: int = 1,
    sim_user_model: str | None = _DEFAULT_GEMINI_MODEL,
    eval_model: str | None = _DEFAULT_GEMINI_MODEL,
    modality: str = "text",
    verbose: bool = False,
    background_noise_file: str | None = None,
    burst_noise_files: list[str] | None = None,
    use_tool_fakes: bool = False,
) -> list[dict[str, Any]]:
    """Runs multiple simulations, optionally in parallel.

    Args:
        test_cases: List of test case dictionaries.
        runs: Number of runs per test case.
        parallel: Number of parallel workers (capped at 25).
        sim_user_model: Gemini model to use for simulated user.
        eval_model: Gemini model to use for evaluating expectations.
        modality: 'text' or 'audio'.
        verbose: Whether to log to console (only active if parallel=1).
        use_tool_fakes: Use fake tools for the session if available.
    """
    sim_user_model = sim_user_model or _DEFAULT_GEMINI_MODEL
    eval_model = eval_model or _DEFAULT_GEMINI_MODEL
    jobs = self._prepare_simulation_jobs(test_cases, runs)
    return self._aggregate_simulation_results(
        jobs,
        runs,
        parallel,
        sim_user_model,
        eval_model,
        modality,
        verbose,
        background_noise_file,
        burst_noise_files,
        use_tool_fakes=use_tool_fakes,
    )

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