diff --git a/dashscope/finetune/agentic_rl.py b/dashscope/finetune/agentic_rl.py index 61b9c184..aed51aff 100644 --- a/dashscope/finetune/agentic_rl.py +++ b/dashscope/finetune/agentic_rl.py @@ -3,6 +3,7 @@ # Copyright (c) Alibaba, Inc. and its affiliates. +import time from typing import Union, List, Optional, ClassVar, Dict, Any from typing_extensions import Self @@ -43,6 +44,7 @@ from dashscope.finetune.reinforcement.common.errors import ( RegistrationError, ValidationError, + InstanceQueryError, RuntimeErrorWithCode, ValueErrorWithCode, DatasetsError, @@ -426,7 +428,28 @@ async def test_functions( functype: FunctionType, input_data: Dict[str, Any], api_key: str = None, + pull_logs: bool = False, + log_page_size: int = 100, + start_time: Optional[int] = None, + end_time: Optional[int] = None, ): + """Test a deployed function instance with custom input data. + + Args: + instance_id: Target function instance ID. + functype: Function type (ROLLOUT/REWARD/GROUP_REWARD). + input_data: Test input payload. + api_key: DashScope API key (uses DASHSCOPE_API_KEY env var + if omitted). + pull_logs: If True, pull all logs of the function instance + (with pagination) after verification and print them + between separator markers. + log_page_size: Page size used when pulling logs. + start_time: Optional start time filter (in seconds) for log + pulling. Defaults to 24 hours before ``end_time``. + end_time: Optional end time filter (in seconds) for log + pulling. Defaults to the current time (now). + """ try: set_api_key(api_key) @@ -450,13 +473,197 @@ async def test_functions( }, ) - return await AgenticRLFunctionComponent.verify_function( + result = await AgenticRLFunctionComponent.verify_function( value, instance_id, ) + if pull_logs: + await cls._pull_function_instance_logs( + function_instance_id=instance_id, + page_size=log_page_size, + start_time=start_time, + end_time=end_time, + ) + + return result + except Exception as e: raise ValidationError( "Function test failed", error_code=3008, ) from e + + @classmethod + async def query_function_instance_logs( + cls, + function_instance_id: str, + page_number: int = 1, + page_size: int = 100, + start_time: Optional[int] = None, + end_time: Optional[int] = None, + keywords: Optional[List[str]] = None, + api_key: str = None, + ) -> Dict[str, Any]: + """Query one page of logs for a function (faas) instance. + + Args: + function_instance_id: Target function instance ID. + page_number: Page number, starting from 1. + page_size: Number of log entries per page. + start_time: Optional start time filter (in seconds). + end_time: Optional end time filter (in seconds). + keywords: Optional keyword filters for log messages. + api_key: DashScope API key (uses DASHSCOPE_API_KEY env var + if omitted). + + Returns: + Raw response dict of the log query API. + """ + try: + set_api_key(api_key) + + fc_component = AgenticRLFunctionComponent + return await fc_component.query_function_instance_logs( + function_instance_id=function_instance_id, + page_number=page_number, + page_size=page_size, + start_time=start_time, + end_time=end_time, + keywords=keywords, + ) + + except Exception as e: + if hasattr(e, "error_code"): + raise + raise InstanceQueryError( + "Function instance log query failed", + error_code=3009, + ) from e + + @classmethod + async def query_all_function_instance_logs( + cls, + function_instance_id: str, + page_size: int = 100, + start_time: Optional[int] = None, + end_time: Optional[int] = None, + keywords: Optional[List[str]] = None, + api_key: str = None, + ) -> List[str]: + """Fetch all logs of a function (faas) instance. + + Pages are fetched one by one via the paginated log query until + all entries are collected. + + Args: + function_instance_id: Target function instance ID. + page_size: Number of log entries per page. + start_time: Optional start time filter (in seconds). + end_time: Optional end time filter (in seconds). + keywords: Optional keyword filters for log messages. + api_key: DashScope API key (uses DASHSCOPE_API_KEY env var + if omitted). + + Returns: + Aggregated list of log messages across all pages. + """ + try: + set_api_key(api_key) + + fc_component = AgenticRLFunctionComponent + return await fc_component.query_all_function_instance_logs( + function_instance_id=function_instance_id, + page_size=page_size, + start_time=start_time, + end_time=end_time, + keywords=keywords, + ) + + except Exception as e: + if hasattr(e, "error_code"): + raise + raise InstanceQueryError( + "Function instance log query failed", + error_code=3010, + ) from e + + @classmethod + async def delete_function_instance( + cls, + function_instance_id: str, + api_key: str = None, + ) -> Dict[str, Any]: + """Delete a function (faas) runtime instance. + + Args: + function_instance_id: Target function instance ID. + api_key: DashScope API key (uses DASHSCOPE_API_KEY env var + if omitted). + + Returns: + Raw response dict of the delete API, containing + ``data.status`` (e.g. "deleted") and + ``data.sandbox_code``. + + Raises: + InstanceQueryError: If the deletion fails. + """ + try: + set_api_key(api_key) + + fc_component = AgenticRLFunctionComponent + return await fc_component.delete_function_instance( + function_instance_id=function_instance_id, + ) + + except Exception as e: + if hasattr(e, "error_code"): + raise + raise InstanceQueryError( + "Function instance deletion failed", + error_code=3011, + ) from e + + @classmethod + async def _pull_function_instance_logs( + cls, + function_instance_id: str, + page_size: int = 100, + start_time: Optional[int] = None, + end_time: Optional[int] = None, + ) -> None: + """Pull all logs of a function instance and print them between + separator markers (best effort, never raises). + + When ``start_time``/``end_time`` are not provided, defaults to + pulling logs from 24 hours ago until now. + """ + if end_time is None: + end_time = int(time.time()) + if start_time is None: + start_time = end_time - 24 * 60 * 60 + logger.info( + "************start query log " + f"(function_instance_id={function_instance_id})*******", + ) + try: + logs = await cls.query_all_function_instance_logs( + function_instance_id=function_instance_id, + page_size=page_size, + start_time=start_time, + end_time=end_time, + ) + for entry in logs: + logger.info(f"[function instance log] {entry}") + logger.info( + f"Pulled {len(logs)} log entries for function instance " + f"{function_instance_id}", + ) + except Exception as e: + logger.warning( + f"Failed to pull logs for function instance " + f"{function_instance_id}: {e}", + ) + finally: + logger.info("************end query log *******") diff --git a/dashscope/finetune/reinforcement/__init__.py b/dashscope/finetune/reinforcement/__init__.py index b524d976..cbe8d459 100644 --- a/dashscope/finetune/reinforcement/__init__.py +++ b/dashscope/finetune/reinforcement/__init__.py @@ -16,11 +16,13 @@ DASHSCOPE_HTTP_BASE_URL, LOG_LEVEL, FC_API_KEY, + FC_DELETE_API, FC_FILES_START, FC_LOAD_API, FC_PYPI_LIB, FC_PYPI_REPO, FC_QUERY_API, + FC_QUERY_LOG_API, FC_REGISTER_REWARD_API, FC_REGISTER_ROLLOUT_API, FC_REGISTER_GROUP_REWARD_API, @@ -154,11 +156,13 @@ def __getattr__(name): "DASHSCOPE_HTTP_BASE_URL", "LOG_LEVEL", "FC_API_KEY", + "FC_DELETE_API", "FC_FILES_START", "FC_LOAD_API", "FC_PYPI_LIB", "FC_PYPI_REPO", "FC_QUERY_API", + "FC_QUERY_LOG_API", "FC_REGISTER_REWARD_API", "FC_REGISTER_ROLLOUT_API", "FC_REGISTER_GROUP_REWARD_API", diff --git a/dashscope/finetune/reinforcement/common/constants.py b/dashscope/finetune/reinforcement/common/constants.py index ae4c5784..47c1eba5 100644 --- a/dashscope/finetune/reinforcement/common/constants.py +++ b/dashscope/finetune/reinforcement/common/constants.py @@ -83,6 +83,14 @@ def get_int_env(env_var: str, default: int) -> int: ) FC_LOAD_API = os.environ.get("FC_LOAD_API", f"{FC_BASE_ENDPOINT}/online/faas") FC_QUERY_API = os.environ.get("FC_QUERY_API", f"{FC_BASE_ENDPOINT}/query/faas") +FC_QUERY_LOG_API = os.environ.get( + "FC_QUERY_LOG_API", + f"{FC_BASE_ENDPOINT}/query/faas/log", +) +FC_DELETE_API = os.environ.get( + "FC_DELETE_API", + f"{FC_BASE_ENDPOINT}/delete/faas", +) FC_LAYER_CREATE_API = os.environ.get( # POST "FC_LAYER_CREATE_API", f"{FC_BASE_ENDPOINT}/create/faas/layer", diff --git a/dashscope/finetune/reinforcement/common/model.py b/dashscope/finetune/reinforcement/common/model.py index 7c50ee9a..7819adf6 100644 --- a/dashscope/finetune/reinforcement/common/model.py +++ b/dashscope/finetune/reinforcement/common/model.py @@ -25,8 +25,10 @@ # Local Application from dashscope.finetune.reinforcement.common.constants import ( FC_API_KEY, + FC_DELETE_API, FC_LOAD_API, FC_QUERY_API, + FC_QUERY_LOG_API, FC_REGISTER_REWARD_API, FC_REGISTER_ROLLOUT_API, FC_REGISTER_GROUP_REWARD_API, @@ -1056,6 +1058,239 @@ async def verify_function( error_code=2047, ) from e + @staticmethod + def _extract_log_page( + result: Dict[str, Any], + ) -> Tuple[List[str], Optional[int]]: + """Extract log messages and total count from a log query response. + + Two payload shapes are supported. The DashScope gateway returns + snake_case keys under "output":: + + { + "request_id": "...", + "output": { + "page_number": 1, + "page_size": 100, + "total_count": 86, + "log_entries": [ + {"instance_id": "...", "message": "...", + "timestamp": 1699051200} + ] + } + } + + while the backend service returns camelCase keys under "data":: + + { + "success": true, + "data": { + "pageNumber": 1, + "pageSize": 50, + "totalCount": 128, + "logEntries": [ + {"instanceId": "...", "message": "...", + "timestamp": 1699051200} + ] + } + } + + Only the "message" field of each log entry is kept; other fields + are ignored. + """ + data = result.get("output") + if not isinstance(data, Dict): + data = result.get("data") + if not isinstance(data, Dict): + return [], None + + entries = data.get("log_entries") or data.get("logEntries") or [] + logs = [ + entry.get("message", "") if isinstance(entry, Dict) else str(entry) + for entry in entries + ] + + total = data.get("total_count") + if total is None: + total = data.get("totalCount") + if not isinstance(total, int): + total = None + + return logs, total + + @classmethod + async def query_function_instance_logs( + cls, + function_instance_id: str, + page_number: int = 1, + page_size: int = 100, + start_time: Optional[int] = None, + end_time: Optional[int] = None, + keywords: Optional[List[str]] = None, + ) -> Dict[str, Any]: + """Query logs of a function (faas) instance (single page). + + Args: + function_instance_id: Target function instance ID. + page_number: Page number, starting from 1. + page_size: Number of log entries per page. + start_time: Optional start time filter (in seconds). + end_time: Optional end time filter (in seconds). + keywords: Optional keyword filters for log messages. + + Returns: + Raw response dict of the log query API. + """ + if not function_instance_id: + raise InputError( + "No function instance ID available for log query", + error_code=2071, + ) + + url = f"{FC_QUERY_LOG_API}/{function_instance_id}" + request_data: Dict[str, Any] = { + "pageNumber": page_number, + "pageSize": page_size, + } + if start_time is not None: + request_data["startTime"] = start_time + if end_time is not None: + request_data["endTime"] = end_time + if keywords: + request_data["keywords"] = keywords + + result = await client_fc(FC_API_KEY, url, request_data) + status = result.get("status", {}) + if isinstance(status, Dict) and status.get("code", 200) != 200: + raise InstanceQueryError( + f"Log query failed: {result}", + error_code=2072, + instance_id=function_instance_id, + ) + if result.get("success") is False: + raise InstanceQueryError( + f"Log query failed: {result.get('message') or result}", + error_code=2073, + instance_id=function_instance_id, + ) + + logger.debug( + f"Log query completed | FunctionInstanceID: " + f"{function_instance_id} | " + f"Page: {page_number} | Size: {page_size}", + ) + return result + + @classmethod + async def query_all_function_instance_logs( + cls, + function_instance_id: str, + page_size: int = 100, + start_time: Optional[int] = None, + end_time: Optional[int] = None, + keywords: Optional[List[str]] = None, + max_pages: int = 100, + ) -> List[str]: + """Fetch all logs of a function (faas) instance with pagination. + + Iterates pages starting from page 1 until all entries are + collected (based on the reported total count, an empty page, or + a short page). + + Args: + function_instance_id: Target function instance ID. + page_size: Number of log entries per page. + start_time: Optional start time filter (in seconds). + end_time: Optional end time filter (in seconds). + keywords: Optional keyword filters for log messages. + max_pages: Safety limit on the number of pages to fetch. + + Returns: + Aggregated list of log messages across all pages. + """ + all_logs: List[str] = [] + page_number = 1 + while page_number <= max_pages: + result = await cls.query_function_instance_logs( + function_instance_id=function_instance_id, + page_number=page_number, + page_size=page_size, + start_time=start_time, + end_time=end_time, + keywords=keywords, + ) + logs, total = cls._extract_log_page(result) + all_logs.extend(logs) + + if total is not None and len(all_logs) >= total: + break + if len(logs) < page_size: + break + page_number += 1 + + logger.debug( + f"All logs fetched | FunctionInstanceID: " + f"{function_instance_id} | " + f"Pages: {page_number} | Entries: {len(all_logs)}", + ) + return all_logs + + @classmethod + async def delete_function_instance( + cls, + function_instance_id: str, + ) -> Dict[str, Any]: + """Delete a function (faas) runtime instance. + + Args: + function_instance_id: Target function instance ID. + + Returns: + Raw response dict of the delete API, e.g.:: + + { + "code": 0, + "message": "success", + "data": { + "status": "deleted", + "sandbox_code": "sandbox-xxx" + } + } + """ + if not function_instance_id: + raise InputError( + "No function instance ID available for deletion", + error_code=2074, + ) + + url = f"{FC_DELETE_API}/{function_instance_id}" + result = await client_fc(FC_API_KEY, url, {}) + status = result.get("status", {}) + if isinstance(status, Dict) and status.get("code", 200) != 200: + raise InstanceQueryError( + f"Function instance deletion failed: {result}", + error_code=2075, + instance_id=function_instance_id, + ) + code = result.get("code") + if result.get("success") is False or ( + code is not None and code not in (0, 200) + ): + raise InstanceQueryError( + "Function instance deletion failed: " + f"{result.get('message') or result}", + error_code=2075, + instance_id=function_instance_id, + ) + + data = result.get("data", {}) + status_value = data.get("status") if isinstance(data, Dict) else data + logger.info( + f"Function instance deleted | FunctionInstanceID: " + f"{function_instance_id} | Status: {status_value}", + ) + return result + class RolloutFunctionComponent(AgenticRLFunctionComponent): """Rollout function component with type fixed as ROLLOUT.""" diff --git a/dashscope/finetune/reinforcement/common/utils.py b/dashscope/finetune/reinforcement/common/utils.py index f51c2b21..ff3ea453 100644 --- a/dashscope/finetune/reinforcement/common/utils.py +++ b/dashscope/finetune/reinforcement/common/utils.py @@ -520,11 +520,20 @@ def _sync_upload_to_oss(signed_url: str, zip_path: str) -> int: file_size = os.path.getsize(zip_path) size_mb = file_size / (1024 * 1024) if file_size > FC_OSS_FILE_SIZE_WARNING: - logger.warning( - f"Uploading large file: {zip_path} ({size_mb:.2f}MB) to OSS", + max_mb = FC_OSS_FILE_SIZE_WARNING / (1024 * 1024) + reason = ( + f"Package too large: {zip_path} ({size_mb:.2f}MB) " + f"exceeds the maximum allowed size of {max_mb:.0f}MB " + f"for OSS upload. The upload was not attempted. " + f"Reduce the package size (e.g. exclude large files via " + f"env var FC_ZIP_EXCLUDE_PATTERNS) or raise the limit via " + f"env var FC_OSS_FILE_SIZE_WARNING (current: " + f"{FC_OSS_FILE_SIZE_WARNING} bytes)." ) + logger.error(reason) raise OSSUploadError( - f"Uploading large file: {zip_path} ({size_mb:.2f}MB) to OSS", + reason, + file_size=file_size, ) logger.debug( @@ -541,11 +550,18 @@ def _sync_upload_to_oss(signed_url: str, zip_path: str) -> int: if response.status_code != 200: error_msg = response.text + logger.error( + f"OSS upload failed | File: {zip_path} " + f"({size_mb:.2f}MB), Status: {response.status_code}, " + f"Response: {error_msg}", + ) raise OSError( f"OSS upload failed ({response.status_code}): {error_msg}", ) return response.status_code + except OSSUploadError: + raise except Exception as e: raise RuntimeErrorWithCode( "OSS upload error", @@ -663,6 +679,7 @@ async def to_bailian_data(files: List[FileSpec]) -> List[str]: timeout=BAILIAN_FILE_TIMEOUT, retry_times=1, ) + logger.info(f"File upload result: {result}") # Handle errors if result.get("status", {}).get("code", 200) != 200: @@ -674,7 +691,7 @@ async def to_bailian_data(files: List[FileSpec]) -> List[str]: data = result.get("data", {}) if "failed_uploads" in data and data["failed_uploads"]: failed_files = ", ".join( - [f["name"] for f in data["failed_uploads"]], + [f"{f.get('name')}: {f}" for f in data["failed_uploads"]], ) raise OutputError( f"Partial upload failed: {failed_files}", diff --git a/dashscope/finetune/reinforcement/component/data/base_data_model.py b/dashscope/finetune/reinforcement/component/data/base_data_model.py index b09b4f96..319d7d0f 100644 --- a/dashscope/finetune/reinforcement/component/data/base_data_model.py +++ b/dashscope/finetune/reinforcement/component/data/base_data_model.py @@ -26,6 +26,17 @@ def _generate_ro_id(length: int) -> str: return "ro-" + hashlib.sha1(uuid.uuid4().bytes).hexdigest()[:length] +def _normalize_error_code(v: Any) -> Any: + """Coerce ``int`` error codes to ``str``; pass other values through. + + The protocol keeps ``error_code`` as a string, so passing an int no + longer hard-fails validation and downstream always receives a str. + """ + if isinstance(v, int) and not isinstance(v, bool): + return str(v) + return v + + # ========================================================================== # # Enum Definitions # # ========================================================================== # diff --git a/dashscope/finetune/reinforcement/component/data/reward_output.py b/dashscope/finetune/reinforcement/component/data/reward_output.py index eec6013d..5ff5ab9b 100644 --- a/dashscope/finetune/reinforcement/component/data/reward_output.py +++ b/dashscope/finetune/reinforcement/component/data/reward_output.py @@ -5,11 +5,12 @@ Data model definitions for Reward processor output results. """ -from typing import Dict, Optional -from pydantic import BaseModel, Field +from typing import Any, Dict, Optional +from pydantic import BaseModel, Field, field_validator from dashscope.finetune.reinforcement.component.data.base_data_model import ( TaskStatus, + _normalize_error_code, ) @@ -55,6 +56,16 @@ class RewardOutput(BaseModel): default=None, description="Error details if the reward computation failed.", ) + error_code: Optional[str] = Field( + default=None, + description="Error code if the reward computation failed.", + ) + + @field_validator("error_code", mode="before") + @classmethod + def normalize_error_code(cls, v: Any) -> Any: + """Normalize int error codes to str (protocol stays string).""" + return _normalize_error_code(v) class Config: extra = "allow" diff --git a/dashscope/finetune/reinforcement/component/data/rollout_output.py b/dashscope/finetune/reinforcement/component/data/rollout_output.py index 9e97eaee..c50b4ba9 100644 --- a/dashscope/finetune/reinforcement/component/data/rollout_output.py +++ b/dashscope/finetune/reinforcement/component/data/rollout_output.py @@ -6,12 +6,13 @@ Corresponds to the RolloutOutput structure in protocol.py. """ -from typing import Optional -from pydantic import BaseModel, Field +from typing import Any, Optional +from pydantic import BaseModel, Field, field_validator from dashscope.finetune.reinforcement.component.data.base_data_model import ( TaskStatus, AgentOutput, + _normalize_error_code, ) @@ -40,6 +41,16 @@ class RolloutOutput(BaseModel): default=None, description="Error details (when failed).", ) + error_code: Optional[str] = Field( + default=None, + description="Error code (when failed).", + ) + + @field_validator("error_code", mode="before") + @classmethod + def normalize_error_code(cls, v: Any) -> Any: + """Normalize int error codes to str (protocol stays string).""" + return _normalize_error_code(v) class Config: extra = "allow" diff --git a/tests/unit/test_agentic_rl_components.py b/tests/unit/test_agentic_rl_components.py index b248abe9..a6c22530 100644 --- a/tests/unit/test_agentic_rl_components.py +++ b/tests/unit/test_agentic_rl_components.py @@ -279,6 +279,22 @@ def test_rollout_output_failure(self): assert output.error == "Model inference timeout" assert output.agent_output is None + def test_rollout_output_error_code_int_normalized(self): + output = RolloutOutput( + status=TaskStatus.FAILED, + error="Model inference timeout", + error_code=3001, + ) + assert output.error_code == "3001" + assert isinstance(output.error_code, str) + + def test_rollout_output_error_code_str_passthrough(self): + output = RolloutOutput( + status=TaskStatus.FAILED, + error_code="RATE_LIMIT", + ) + assert output.error_code == "RATE_LIMIT" + def test_rollout_output_defaults(self): output = RolloutOutput() assert output.status == TaskStatus.SUCCESS @@ -335,6 +351,15 @@ def test_reward_output_failure(self): ) assert output.status == TaskStatus.FAILED + def test_reward_output_error_code_int_normalized(self): + output = RewardOutput( + reward=Reward(reward_score=0.0), + status=TaskStatus.FAILED, + error_code=3002, + ) + assert output.error_code == "3002" + assert isinstance(output.error_code, str) + def test_reward_output_missing_reward(self): with pytest.raises(ValidationError): RewardOutput()