Merge pull request #230 from GoogleCloudPlatform/gemini-cli-evals

Add Simulated LLM user for true multi turn conversation

Author: prernakakkar-google

datasets/gemini-cli-tools/example_run_config.yaml

@@ -8,6 +8,7 @@ setup_directory: datasets/gemini-cli-tools/setup
 # Orchestrator Configuration
 orchestrator: geminicli
 model_config: datasets/model_configs/gemini_cli_model.yaml
+simulated_user_model_config: datasets/model_configs/gemini_2.5_pro_model.yaml
 
 ############################################################
 ### Scorer Related Configs

datasets/gemini-cli-tools/gemini-cli.evalset.json

@@ -3,8 +3,8 @@
     {
       "id": "cloud-sql-debug-01",
       "starting_prompt": "list all instances in project astana-evaluation",
-      "conversation_plan": "Ask the agent to list instances in project astana-evaluation.",
-      "expected_trajectory": ["list_instances"],
+      "conversation_plan": "Ask the agent to list instances in project astana-evaluation. Once all instances are listed if nl2code exist get its state and validate its RUNNABLE",
+      "expected_trajectory": ["list_instances", "get_instance"],
       "env": {
         "GOOGLE_CLOUD_PROJECT": "astana-evaluation"
       },

datasets/model_configs/gemini_cli_model.yaml

@@ -1,2 +1,2 @@
-gemini_cli_version: "@google/gemini-cli@0.23.0"
+gemini_cli_version: "@google/gemini-cli@0.25.1"
 generator: gemini_cli

evalbench/evaluator/agentevaluator.py

@@ -8,6 +8,11 @@
 from util.config import load_yaml_config
 from mp import mprunner
 from work.agentgenwork import AgentGenWork
+from evaluator.simulateduser import SimulatedUser
+from work.agentscorework import AgentScoreWork
+import json
+import subprocess
+from typing import Dict
 
 
 class AgentEvaluator:
@@ -68,7 +73,14 @@ def _evaluate_gemini_cli(
         }
 
         for item in dataset:
-            work = AgentGenWork(self.generator, self.agent_version, item, job_id=job_id, metadata=metadata)
+            simulated_user = SimulatedUser(self.config)
+            work = AgentGenWork(
+                processor=self.process_scenario,
+                eval_result=item,
+                job_id=job_id,
+                metadata=metadata,
+                simulated_user=simulated_user
+            )
             self.agentrunner.execute_work(work)
 
         for future in concurrent.futures.as_completed(self.agentrunner.futures):
@@ -80,3 +92,117 @@ def _evaluate_gemini_cli(
                 scoring_results.extend(item.scoring_results)
 
         return eval_outputs, scoring_results
+
+    def process_scenario(
+        self,
+        scenario: Dict[str, Any],
+        eval_result: Any,
+        job_id: str,
+        metadata: Dict[str, Any],
+        simulated_user: Any = None
+    ):
+        """Processes a single scenario."""
+        current_prompt = scenario["starting_prompt"]
+        env = scenario.get("env", {})
+        max_turns = scenario.get("max_turns", 1)
+        conversation_plan = scenario.get("conversation_plan", "")
+        conversation_history = []
+        accumulated_tools = []
+        last_result = None
+
+        for turn in range(max_turns):
+            logging.info(f"Turn {turn + 1}/{max_turns} - Prompt: {current_prompt}")
+
+            if isinstance(self.generator, GeminiCliGenerator):
+                cli_cmd = self.generator.create_command(
+                    cli=self.agent_version,
+                    prompt=current_prompt,
+                    env=env,
+                    resume=(turn > 0)
+                )
+                result = self.generator.safe_generate(cli_cmd)
+            else:
+                result = self.generator.generate(current_prompt)
+
+            last_result = result
+
+            self._log_cli_result(turn, max_turns, result)
+
+            tools = []
+            if isinstance(self.generator, GeminiCliGenerator):
+                tools = self.generator.extract_tools(result.stdout)
+            accumulated_tools.extend(tools)
+
+            conversation_history.append({
+                "user": current_prompt,
+                "agent": result.stdout
+            })
+
+            if turn < max_turns - 1:
+                if simulated_user:
+                    next_response = simulated_user.get_next_response(
+                        conversation_plan,
+                        conversation_history,
+                        result.stdout
+                    )
+                    if "TERMINATE" in next_response:
+                        logging.info("Simulated user terminated conversation.")
+                        break
+                    current_prompt = next_response
+                else:
+                    break
+
+        if last_result:
+            self._finalize_scenario(
+                scenario,
+                last_result,
+                conversation_history,
+                accumulated_tools,
+                eval_result,
+                job_id,
+                metadata
+            )
+
+    def _log_cli_result(self, turn: int, max_turns: int, result: subprocess.CompletedProcess):
+        logging.info(f"Turn {turn + 1}/{max_turns} - Gemini CLI exit code: {result.returncode}")
+        logging.info(f"Turn {turn + 1}/{max_turns} - Gemini CLI stdout: {result.stdout}")
+        logging.info(f"Turn {turn + 1}/{max_turns} - Gemini CLI stderr: {result.stderr}")
+
+    def _finalize_scenario(
+        self,
+        scenario: Dict[str, Any],
+        last_result: subprocess.CompletedProcess,
+        conversation_history: List[Dict[str, str]],
+        accumulated_tools: List[str],
+        eval_result: Any,
+        job_id: str,
+        metadata: Dict[str, Any]
+    ):
+        """Finalizes the scenario by scoring and appending results."""
+        # Prepare intermediate eval_output with all necessary data for scoring
+        eval_output_data = {
+            "eval_id": scenario["id"],
+            "stdout": last_result.stdout,
+            "stderr": last_result.stderr,
+            "returncode": last_result.returncode,
+            "prompt_generator_error": None,
+            "generated_error": None,
+            "sql_generator_error": None,
+            "golden_error": None,
+            "generated_sql": "skipped",
+            "prompt": scenario["starting_prompt"],
+            "conversation_history": json.dumps(conversation_history, indent=2),
+            "scenario": scenario,
+            "accumulated_tools": accumulated_tools,
+            "job_id": job_id,
+            "metadata": metadata
+        }
+
+        score_work = AgentScoreWork(
+            config=metadata,
+            eval_output=eval_output_data,
+            scoring_results=eval_result.scoring_results
+        )
+        score_work.run()
+
+        eval_result.agent_results.append(eval_output_data)

evalbench/evaluator/simulateduser.py

@@ -0,0 +1,53 @@
+import generators.models as models
+import generators.prompts as prompts
+import threading
+import logging
+
+
+class SimulatedUser:
+    def __init__(self, config):
+        self.config = config
+
+        global_models = {
+            "lock": threading.Lock(),
+            "registered_models": {}
+        }
+
+        # Expect 'simulated_user_model_config' path in config
+        model_config_path = config.get("simulated_user_model_config")
+
+        self.prompt_generator = prompts.get_generator(
+            None,
+            {"prompt_generator": "SimulatedUserPromptGenerator"},
+            "SimulatedUserPromptGenerator"
+        )
+
+        self.model_generator = None
+        if model_config_path:
+            try:
+                self.model_generator = models.get_generator(
+                    global_models, model_config_path, None
+                )
+            except Exception as e:
+                logging.warning(f"Failed to load simulated user model from {model_config_path}: {e}")
+        else:
+            logging.warning("No 'simulated_user_model_config' provided. SimulatedUser will not be able to generate responses.")
+
+    def get_next_response(self, conversation_plan: str, history: list, last_agent_reply: str) -> str:
+        if not self.model_generator:
+            logging.error("Model generator not initialized.")
+            return "TERMINATE"
+
+        payload = {
+            "conversation_plan": conversation_plan,
+            "history": history,
+            "last_agent_reply": last_agent_reply
+        }
+
+        # Generate prompt
+        self.prompt_generator.generate(payload)
+        prompt = payload["prompt"]
+
+        # Call model
+        response = self.model_generator.generate(prompt)
+        return response

evalbench/generators/models/gemini_cli.py

@@ -80,3 +80,37 @@ def _run_gemini_cli(self, cli_cmd: CLICommand):
             command,
             env=env,
         )
+
+    def parse_response(self, stdout: str) -> dict:
+        """Parses the JSON output from Gemini CLI."""
+        try:
+            return json.loads(stdout)
+        except json.JSONDecodeError:
+            return {}
+
+    def extract_tools(self, stdout: str) -> list[str]:
+        """Extracts the list of tools used from the CLI output."""
+        output_json = self.parse_response(stdout)
+        if (
+            "stats" in output_json
+            and "tools" in output_json["stats"]
+            and "byName" in output_json["stats"]["tools"]
+        ):
+            return list(output_json["stats"]["tools"]["byName"].keys())
+        return []
+
+    def safe_generate(self, cli_cmd: CLICommand) -> subprocess.CompletedProcess:
+        """Runs the generation and handles empty responses."""
+        result = self.generate(cli_cmd)
+        if isinstance(result, str) and not result:
+            return subprocess.CompletedProcess(
+                args=[cli_cmd.cli],
+                returncode=1,
+                stdout="",
+                stderr="Error: Generator returned empty response (possibly resource exhausted).",
+            )
+        return result
+
+    def create_command(self, cli: str, prompt: str, env: dict = None, resume: bool = False) -> CLICommand:
+        """Creates a CLICommand object."""
+        return CLICommand(cli=cli, prompt=prompt, env=env, resume=resume)

evalbench/generators/prompts/__init__.py

@@ -2,6 +2,7 @@
 from .passthrough import NOOPGenerator
 from .interactsystem import InteractSystemGenerator
 from .interactuser import InteractUserGenerator
+from .simulateduser import SimulatedUserPromptGenerator
 from .dataagentinteractuser import DataAgentInteractUserGenerator
 
 
@@ -17,6 +18,8 @@ def get_generator(db, promptgenerator_config, generator_name=None):
         return InteractSystemGenerator(db, promptgenerator_config)
     if promptgenerator_config["prompt_generator"] == "InteractUserGenerator":
         return InteractUserGenerator(db, promptgenerator_config)
+    if promptgenerator_config["prompt_generator"] == "SimulatedUserPromptGenerator":
+        return SimulatedUserPromptGenerator(db, promptgenerator_config)
     if promptgenerator_config["prompt_generator"] == "DataAgentInteractUserGenerator":
         return DataAgentInteractUserGenerator(db, promptgenerator_config)
     raise ValueError("Prompt Generator not Supported")

evalbench/generators/prompts/simulateduser.py

@@ -0,0 +1,45 @@
+from .generator import PromptGenerator
+
+SIMULATED_USER_PROMPT = """You are a simulated user interacting with Gemini CLI agent.
+Your objective is to follow the conversation plan provided below and engage with the agent naturally to achieve the goals.
+
+# Conversation Plan:
+[[conversation_plan]]
+
+# Interaction History:
+[[history]]
+
+# Last Agent Reply:
+[[last_agent_reply]]
+
+Based on the plan and the agent's last reply, provide your next input to the CLI agent.
+If the plan is fully completed or you cannot proceed, reply with "TERMINATE".
+Only provide the text command or response. Do not include markdown formatting or explanations unless necessary for the command.
+"""
+
+
+class SimulatedUserPromptGenerator(PromptGenerator):
+    def __init__(self, db, promptgenerator_config):
+        super().__init__(db, promptgenerator_config)
+        self.prompt_template = SIMULATED_USER_PROMPT
+
+    def setup(self):
+        pass
+
+    def generate(self, item):
+        # item is the payload dictionary
+        plan = item.get("conversation_plan", "")
+        history_list = item.get("history", [])
+        last_reply = item.get("last_agent_reply", "")
+
+        # Format history
+        history_str = ""
+        for turn in history_list:
+            history_str += f"User: {turn['user']}\nAgent: {turn['agent']}\n"
+
+        prompt = self.prompt_template.replace("[[conversation_plan]]", str(plan))
+        prompt = prompt.replace("[[history]]", history_str)
+        prompt = prompt.replace("[[last_agent_reply]]", str(last_reply))
+
+        item["prompt"] = prompt
+        return item