Merge branch 'dev' of github.com:JamesKrW/vagen into dev (65b1f7b3) · Commits · 张泽凯 / VAGEN

tmpn40k8k1n

deleted100644 → 0

+0 −172

Original line number	Diff line number	Diff line

		Section "Device"
		Identifier "Device0"
		Driver "nvidia"
		VendorName "NVIDIA Corporation"
		BusID "PCI:5:0:0"
		EndSection


		Section "Screen"
		Identifier "Screen0"
		Device "Device0"
		DefaultDepth 24
		Option "AllowEmptyInitialConfiguration" "True"
		SubSection "Display"
		Depth 24
		Virtual 1024 768
		EndSubSection
		EndSection


		Section "Device"
		Identifier "Device1"
		Driver "nvidia"
		VendorName "NVIDIA Corporation"
		BusID "PCI:6:0:0"
		EndSection


		Section "Screen"
		Identifier "Screen1"
		Device "Device1"
		DefaultDepth 24
		Option "AllowEmptyInitialConfiguration" "True"
		SubSection "Display"
		Depth 24
		Virtual 1024 768
		EndSubSection
		EndSection


		Section "Device"
		Identifier "Device2"
		Driver "nvidia"
		VendorName "NVIDIA Corporation"
		BusID "PCI:9:0:0"
		EndSection


		Section "Screen"
		Identifier "Screen2"
		Device "Device2"
		DefaultDepth 24
		Option "AllowEmptyInitialConfiguration" "True"
		SubSection "Display"
		Depth 24
		Virtual 1024 768
		EndSubSection
		EndSection


		Section "Device"
		Identifier "Device3"
		Driver "nvidia"
		VendorName "NVIDIA Corporation"
		BusID "PCI:10:0:0"
		EndSection


		Section "Screen"
		Identifier "Screen3"
		Device "Device3"
		DefaultDepth 24
		Option "AllowEmptyInitialConfiguration" "True"
		SubSection "Display"
		Depth 24
		Virtual 1024 768
		EndSubSection
		EndSection


		Section "Device"
		Identifier "Device4"
		Driver "nvidia"
		VendorName "NVIDIA Corporation"
		BusID "PCI:141:0:0"
		EndSection


		Section "Screen"
		Identifier "Screen4"
		Device "Device4"
		DefaultDepth 24
		Option "AllowEmptyInitialConfiguration" "True"
		SubSection "Display"
		Depth 24
		Virtual 1024 768
		EndSubSection
		EndSection


		Section "Device"
		Identifier "Device5"
		Driver "nvidia"
		VendorName "NVIDIA Corporation"
		BusID "PCI:142:0:0"
		EndSection


		Section "Screen"
		Identifier "Screen5"
		Device "Device5"
		DefaultDepth 24
		Option "AllowEmptyInitialConfiguration" "True"
		SubSection "Display"
		Depth 24
		Virtual 1024 768
		EndSubSection
		EndSection


		Section "Device"
		Identifier "Device6"
		Driver "nvidia"
		VendorName "NVIDIA Corporation"
		BusID "PCI:145:0:0"
		EndSection


		Section "Screen"
		Identifier "Screen6"
		Device "Device6"
		DefaultDepth 24
		Option "AllowEmptyInitialConfiguration" "True"
		SubSection "Display"
		Depth 24
		Virtual 1024 768
		EndSubSection
		EndSection


		Section "Device"
		Identifier "Device7"
		Driver "nvidia"
		VendorName "NVIDIA Corporation"
		BusID "PCI:146:0:0"
		EndSection


		Section "Screen"
		Identifier "Screen7"
		Device "Device7"
		DefaultDepth 24
		Option "AllowEmptyInitialConfiguration" "True"
		SubSection "Display"
		Depth 24
		Virtual 1024 768
		EndSubSection
		EndSection


		Section "ServerLayout"
		Identifier "Layout0"
		Screen 0 "Screen0" 0 0
		Screen 1 "Screen1" 0 0
		Screen 2 "Screen2" 0 0
		Screen 3 "Screen3" 0 0
		Screen 4 "Screen4" 0 0
		Screen 5 "Screen5" 0 0
		Screen 6 "Screen6" 0 0
		Screen 7 "Screen7" 0 0
		EndSection

vagen/env/README.md

+34 −0

Original line number	Diff line number	Diff line
		## General Server
		```
		# Start a Server
		python vagen/env/server.py
		```
		### Navigation
		```
		# Additional dependencies:
		@@ -21,3 +26,32 @@ python vagen/env/navigation/startx.py 1
		pip install gymnasium
		pip install "gymnasium[toy-text]"
		```

		### SVG
		```
		# Additional dependencies:
		pip install bs4
		pip install svgpathtools
		pip install cairosvg

		# Then run experiment of SVG simply copy the code below
		bash vagen/examples/debug_svg_vision_grpo/run.sh
		```

		### SVGDino
		```
		# Additional dependencies:
		pip install bs4
		pip install svgpathtools
		pip install cairosvg
		pip install flask

		# create server for reward model
		python vagen/env/svgdino/reward_model_server.py

		# test whether reward model server functions well
		python vagen/examples/debug_reward_model_server/reward_server_benchmark.py --concurrency 128 --environments 1000 --interactions 3

		# Then run experiment of SVG simply copy the code below
		bash vagen/examples/debug_svgdino_vision_grpo/run.sh
		```
		No newline at end of file

vagen/env/init.py

+14 −1

Original line number	Diff line number	Diff line
		from .sokoban import SokobanEnv,SokobanConfig
		from .frozenlake import FrozenLakeEnv,FrozenLakeConfig
		from .frozenlake import FrozenLakeEnv,FrozenLakeConfig, FrozenLakeService
		from .navigation import NavigationEnv, NavigationConfig
		from .svgdino import SVGDINOEnv, SVGDINOConfig
		from .svg import SVGEnv, SVGConfig, SVGService

		REGISTERED_ENV = {
		"sokoban": {
		"env_cls": SokobanEnv,
		@@ -9,9 +12,19 @@ REGISTERED_ENV = {
		"frozenlake": {
		"env_cls": FrozenLakeEnv,
		"config_cls": FrozenLakeConfig,
		"service_cls": FrozenLakeService
		},
		"navigation": {
		"env_cls": NavigationEnv,
		"config_cls": NavigationConfig
		},
		"svg": {
		"env_cls": SVGEnv,
		"config_cls": SVGConfig,
		"service_cls": SVGService
		},
		"svgdino": {
		"env_cls": SVGDINOEnv,
		"config_cls": SVGDINOConfig,
		}
		}
		No newline at end of file

vagen/env/base_service.py

0 → 100644

+127 −0

Original line number	Diff line number	Diff line
		from abc import ABC, abstractmethod
		from typing import List, Dict, Tuple, Optional, Any, Union
		import uuid
		from concurrent.futures import ThreadPoolExecutor

		class BaseService(ABC):
		"""
		Abstract base class for environment services.
		Implements batch operations for efficient parallel processing.
		Single environment operations are provided as convenience methods
		that invoke the corresponding batch methods.
		"""

		@abstractmethod
		def create_environments_batch(self, ids2configs: Dict[str, Any]) -> None:
		"""
		Create multiple environments in parallel.

		Args:
		ids2configs (Dict[Any, Any]):
		A dictionary where each key is an environment ID and the corresponding
		value is the configuration for that environment.

		Returns:
		None

		Note:
		The implementation should create all environments concurrently.
		It should gracefully handle errors and perform cleanup of any partially created environments.
		"""
		pass

		@abstractmethod
		def reset_batch(self, ids2seeds: Dict[str, Any]) -> Dict[str, Tuple[Any, Any]]:
		"""
		Reset multiple environments in parallel.

		Args:
		ids2seeds (Dict[Any, Any]):
		A dictionary where each key is an environment ID and the corresponding
		value is a seed value (or None for using default seeding behavior).

		Returns:
		Dict[Any, Tuple[Any, Any]]:
		A dictionary mapping environment IDs to tuples of the form (observation, info),
		where 'observation' is the initial state after reset, and 'info' contains additional details.

		Note:
		For environments with a None seed, the default seeding behavior should be applied.
		"""
		pass

		@abstractmethod
		def step_batch(self, ids2actions: Dict[str, Any]) -> Dict[str, Tuple[Dict, float, bool, Dict]]:
		"""
		Step through multiple environments in parallel.

		Args:
		ids2actions (Dict[Any, Any]):
		A dictionary where each key is an environment ID and the corresponding
		value is the action to execute in that environment.

		Returns:
		Dict[Any, Tuple[Dict, float, bool, Dict]]:
		A dictionary mapping environment IDs to tuples of the form
		(observation, reward, done, info), where:
		- 'observation' is the new state of the environment after the action,
		- 'reward' is a float representing the reward received,
		- 'done' is a boolean indicating whether the environment is finished,
		- 'info' contains additional information or context.

		Note:
		The implementation should process all steps in parallel while ensuring that
		each action is correctly applied to its corresponding environment.
		"""
		pass

		@abstractmethod
		def compute_reward_batch(self, env_ids: List[str]) -> Dict[str, float]:
		"""
		Compute the total reward for multiple environments in parallel.

		Args:
		env_ids (List[str]): A list of environment IDs.

		Returns:
		Dict[Any, float]:
		A dictionary mapping each environment ID to its computed total reward.

		Note:
		The implementation should compute rewards concurrently.
		"""
		pass

		@abstractmethod
		def get_system_prompts_batch(self, env_ids: List[str]) -> Dict[str, str]:
		"""
		Retrieve system prompts for multiple environments in parallel.

		Args:
		env_ids (List[str]): A list of environment IDs.

		Returns:
		Dict[Any, str]:
		A dictionary mapping each environment ID to its corresponding system prompt string.

		Note:
		The implementation should retrieve all system prompts concurrently.
		"""
		pass

		@abstractmethod
		def close_batch(self, env_ids: Optional[List[str]] = None) -> None:
		"""
		Close multiple environments and clean up resources in parallel.

		Args:
		env_ids (Optional[List[str]]):
		A list of environment IDs to close. If None, all environments should be closed.

		Returns:
		None

		Note:
		The implementation should perform cleanup concurrently and handle any errors gracefully.
		"""
		pass

vagen/env/client.py

0 → 100644

+313 −0

Original line number	Diff line number	Diff line
		from typing import Dict, List, Tuple, Optional, Any, Union
		import requests
		import time
		from vagen.utils.serial import deserialize_observation, deserialize_step_result

		class BatchEnvClient:
		"""
		Client for interacting with the batch environment server.
		Uses dictionary-based interface to match the server API and service interface.
		"""

		def __init__(self, base_url: str, timeout: int = 60, max_workers: int = 10):
		"""
		Initialize the BatchEnvClient.

		Args:
		base_url: Base URL of the environment server
		timeout: Timeout for HTTP requests in seconds
		max_workers: Maximum number of worker threads for parallel processing
		"""
		self.base_url = base_url.rstrip('/')
		self.timeout = timeout
		self.max_workers = max_workers
		self.env_configs = {} # Store configs for each environment for reference

		def _make_request(self, endpoint: str, method: str = "POST", data: Any = None) -> Any:
		"""
		Make an HTTP request to the environment server.

		Args:
		endpoint: API endpoint to call
		method: HTTP method (GET, POST, etc.)
		data: Data to send with the request

		Returns:
		Response data from the server

		Raises:
		ConnectionError: If the request fails
		"""
		url = f"{self.base_url}/{endpoint}"
		headers = {"Content-Type": "application/json"}

		try:
		if method.upper() == "GET":
		response = requests.get(url, headers=headers, timeout=self.timeout)
		elif method.upper() == "POST":
		response = requests.post(url, headers=headers, json=data, timeout=self.timeout)
		elif method.upper() == "DELETE":
		response = requests.delete(url, headers=headers, json=data, timeout=self.timeout)
		else:
		raise ValueError(f"Unsupported HTTP method: {method}")

		response.raise_for_status() # Raise an exception for 4XX/5XX responses
		return response.json()

		except Exception as e:
		print(f"Exception in _make_request: {str(e)}")
		raise

		def check_server_health(self) -> Dict[str, Any]:
		"""
		Check the health of the server.

		Returns:
		Health status information
		"""
		try:
		return self._make_request("health", method="GET")
		except Exception as e:
		return {"status": "error", "message": str(e)}

		def wait_for_server(self, max_retries: int = 10, retry_delay: float = 1.0) -> bool:
		"""
		Wait for the server to become available.

		Args:
		max_retries: Maximum number of retries
		retry_delay: Delay between retries in seconds

		Returns:
		True if server is available, False otherwise
		"""
		for i in range(max_retries):
		try:
		health = self.check_server_health()
		if health.get("status") == "ok":
		print(f"Server available at {self.base_url}")
		return True
		except Exception:
		pass

		print(f"Waiting for server (attempt {i+1}/{max_retries})...")
		time.sleep(retry_delay)

		print(f"Server not available after {max_retries} attempts")
		return False

		def create_environments_batch(self, ids2configs: Dict[Any, Any]) -> None:
		"""
		Create multiple environments based on the provided configurations.
		Implements BaseService.create_environments_batch interface.

		Args:
		ids2configs: Dictionary mapping environment IDs to their configurations
		"""
		response = self._make_request("environments", "POST", {"ids2configs": ids2configs})
		if response.get("success") != True:
		raise Exception(f"Failed to create environments: {response.get('error', 'Unknown error')}")

		# Store the configs for reference
		for env_id in ids2configs:
		self.env_configs[env_id] = ids2configs[env_id]

		def reset_batch(self, ids2seeds: Dict[str, Any]) -> Dict[str, Tuple[Dict, Dict]]:
		"""
		Reset multiple environments in batch.

		Args:
		ids2seeds: Dictionary mapping environment IDs to seeds

		Returns:
		Dictionary mapping environment IDs to (observation, info) tuples
		"""
		response = self._make_request("batch/reset", "POST", {"ids2seeds": ids2seeds})
		results = response.get("results", {})

		# Deserialize observations
		deserialized_results = {}
		for env_id, (observation, info) in results.items():
		deserialized_results[env_id] = (deserialize_observation(observation), info)

		return deserialized_results

		def step_batch(self, ids2actions: Dict[str, str]) -> Dict[str, Tuple[Dict, float, bool, Dict]]:
		"""
		Step multiple environments in batch.

		Args:
		ids2actions: Dictionary mapping environment IDs to actions

		Returns:
		Dictionary mapping environment IDs to (observation, reward, done, info) tuples
		"""
		response = self._make_request("batch/step", "POST", {"ids2actions": ids2actions})
		results = response.get("results", {})

		# Deserialize observations
		deserialized_results = {}
		for env_id, serialized_result in results.items():
		deserialized_results[env_id] = deserialize_step_result(serialized_result)

		return deserialized_results

		def compute_reward_batch(self, env_ids: List[str]) -> Dict[str, float]:
		"""
		Compute rewards for multiple environments in batch.

		Args:
		env_ids: List of environment IDs

		Returns:
		Dictionary mapping environment IDs to reward values
		"""
		response = self._make_request("batch/reward", "POST", {"env_ids": env_ids})
		return response.get("rewards", {})

		def get_system_prompts_batch(self, env_ids: List[str]) -> Dict[str, str]:
		"""
		Get system prompts for multiple environments in batch.

		Args:
		env_ids: List of environment IDs

		Returns:
		Dictionary mapping environment IDs to system prompt strings
		"""
		response = self._make_request("batch/system_prompt", "POST", {"env_ids": env_ids})
		return response.get("system_prompts", {})

		def close_batch(self, env_ids: Optional[List[str]] = None) -> None:
		"""
		Close multiple environments and clean up resources.

		Args:
		env_ids: Optional list of environment IDs to close. If None, close all environments.
		"""
		# If no env_ids provided, close all known environments
		if env_ids is None:
		env_ids = list(self.env_configs.keys())

		self._make_request("batch/close", "POST", {"env_ids": env_ids})

		# Remove closed environments from tracking
		for env_id in env_ids:
		self.env_configs.pop(env_id, None)

		# Convenience methods for single-environment operations

		def reset(self, env_id: str, seed: Any = None) -> Tuple[Dict, Dict]:
		"""
		Reset a single environment.

		Args:
		env_id: Environment ID
		seed: Optional seed for resetting

		Returns:
		Tuple of (observation, info)
		"""
		results = self.reset_batch({env_id: seed})
		return results.get(env_id, ({}, {"error": "Reset failed"}))

		def step(self, env_id: str, action: str) -> Tuple[Dict, float, bool, Dict]:
		"""
		Take a step in a single environment.

		Args:
		env_id: Environment ID
		action: Action to take

		Returns:
		Tuple of (observation, reward, done, info)
		"""
		results = self.step_batch({env_id: action})
		return results.get(env_id, ({}, 0.0, True, {"error": "Step failed"}))

		def compute_reward(self, env_id: str) -> float:
		"""
		Compute reward for a single environment.

		Args:
		env_id: Environment ID

		Returns:
		Reward value
		"""
		results = self.compute_reward_batch([env_id])
		return results.get(env_id, 0.0)

		def get_system_prompt(self, env_id: str) -> str:
		"""
		Get system prompt for a single environment.

		Args:
		env_id: Environment ID

		Returns:
		System prompt string
		"""
		results = self.get_system_prompts_batch([env_id])
		return results.get(env_id, "")

		def close(self, env_id: str) -> None:
		"""
		Close a single environment.

		Args:
		env_id: Environment ID
		"""
		self.close_batch([env_id])


		if __name__ == "__main__":
		# Example usage of the client
		client = BatchEnvClient(base_url="http://localhost:5000", timeout=10)

		# Wait for server to be available
		if client.wait_for_server():
		try:
		# Create environments
		configs = [
		{
		"env_name": "frozenlake",
		"env_config": {"is_slippery": False, "size": 4, "render_mode": "text"}
		},
		{
		"env_name": "frozenlake",
		"env_config": {"is_slippery": True, "size": 8, "render_mode": "vision"}
		}
		]

		print("Creating environments...")
		env_ids = client.create_environments_batchs(configs)
		print(f"Created {len(env_ids)} environments: {env_ids}")

		# Reset environments
		print("Resetting environments...")
		ids2seeds = {env_id: i*42 for i, env_id in enumerate(env_ids)}
		results = client.reset_batch(ids2seeds)

		# Get system prompts
		print("Getting system prompts...")
		prompts = client.get_system_prompts_batch(env_ids)

		# Step environments
		print("Stepping environments...")
		ids2actions = {
		env_ids[0]: "<think>Let me try going right first.</think><answer>Right</answer>",
		env_ids[1]: "<think>I'll start by going down.</think><answer>Down</answer>"
		}
		results = client.step_batch(ids2actions)

		# Close environments
		print("Closing environments...")
		client.close_batch(env_ids)

		print("Done!")

		except Exception as e:
		print(f"Error: {str(e)}")
		else:
		print("Server not available")
		No newline at end of file

Admin message