Enhance configuration and logging features; add server experiment directory retrieval and log empty content messages

Author: binary-husky

ajet/default_config/ajet_default.yaml

@@ -356,3 +356,9 @@ ajet:
   # DO NOT EDIT, FOR ROBOT TESTING PURPOSE ONLY. NOT FOR HUMAN.
   execute_test: False        # DO NOT EDIT, FOR ROBOT TESTING PURPOSE ONLY. NOT FOR HUMAN.
   execute_testing_lambda: "" # DO NOT EDIT, FOR ROBOT TESTING PURPOSE ONLY. NOT FOR HUMAN.
+
+
+# ------------------ hydra runtime ------------------
+hydra:
+  run:
+    dir: saved_experiments/hydra_outputs/${now:%Y-%m-%d}/${now:%H-%M-%S}

ajet/tuner_lib/experimental/oai_model_server.py

@@ -39,6 +39,7 @@
 from openai.types.chat.chat_completion_chunk import ChoiceDelta, ChoiceDeltaToolCall, ChoiceDeltaToolCallFunction
 
 from ajet.utils.networking import get_host_ip
+from ajet.utils.message_utils import log_empty_content_messages
 from ajet.tuner_lib.experimental.interchange_utils import EpisodeStatus
 from ajet.tuner_lib.experimental.interchange_utils import DEBUG, VERBOSE, API_KEY_PREFIX
 
@@ -288,6 +289,9 @@ async def chat_completions(request: Request, authorization: str = Header(None)):
                 logger.warning(f"First message role is '{first_msg.get('role')}', expected 'system'. Adding default system prompt.")
                 new_req.messages.insert(0, {"role": "system", "content": "You are a helpful assistant, your name is AgentJet."})
 
+        # Detect empty-content messages in the inbound request
+        log_empty_content_messages(new_req.messages, episode_uuid=episode_uuid)
+
         # Create timeline UUID
         timeline_uuid = uuid.uuid4().hex
 

ajet/tuner_lib/experimental/swarm_client.py

@@ -747,6 +747,21 @@ def stop_engine(self):
             raise RuntimeError("Failed to stop training engine")
         self._wait_until_status_change_to(desired_status="ENGINE.OFFLINE")
 
+    def server_experiment_dir(self) -> str:
+        """
+        Fetch the absolute experiment directory from the Swarm server.
+        Returns None if the engine has not started yet (no experiment dir is set).
+        """
+        try:
+            resp = self._http_client.get(
+                f"{self.server_url}/get_server_experiment_dir",
+                timeout=10
+            )
+            raise_for_status_with_detail(resp)
+            return resp.json().get("server_experiment_dir", None)
+        except Exception as e:
+            return "saved_experiments"
+
     def get_rollout_stat(self) -> CurrentBatchRolloutPoolInformation:
         """
         Get the current batch rollout pool information from the Swarm server.

ajet/tuner_lib/experimental/swarm_server.py

@@ -389,6 +389,7 @@ def override_param_callback(config):
                 backbone=backbone,
                 override_param_callback=override_param_callback,
             )
+            shared_mem_dict["server_experiment_dir"] = exe_exp_base
 
             # Setup environment variables
             env, exp_config = setup_environment_vars(args, exp_config, main_yaml_fp)
@@ -491,6 +492,11 @@ async def get_engine_status():
             "global_step": global_step,
         }
 
+    @app.get("/get_server_experiment_dir")
+    async def get_server_experiment_dir():
+        """Return the absolute experiment directory once the engine has started."""
+        return {"server_experiment_dir": shared_mem_dict.get("server_experiment_dir", None)}
+
     # --- episode status ---
     @app.post("/register_episode", response_model=BoolResponse)
     async def register_episode(req: RegisterEpisodeRequest):

ajet/utils/core_env_vars.py

@@ -42,6 +42,7 @@ def get_runtime_env(config, is_trinity: bool = False) -> dict:
             "TRINITY_PLUGIN_DIRS": str((Path(__file__).parent.parent / "backbone").resolve()),
             # "VLLM_ALLOW_RUNTIME_LORA_UPDATING": "true",
             "SWANLAB_API_KEY": os.getenv("SWANLAB_API_KEY", ""),
+            "SWANLAB_LOG_DIR": os.getenv("SWANLAB_LOG_DIR", "saved_experiments/swanlog"),
             "AJET_CONFIG_REDIRECT": os.getenv("AJET_CONFIG_REDIRECT", ""),
             "AJET_DAT_INTERCHANGE_PORT": os.getenv("AJET_DAT_INTERCHANGE_PORT", data_interchange_port),
             "MASTER_NODE_IP": os.getenv("MASTER_NODE_IP", master_node_ip),

ajet/utils/message_utils.py

@@ -1,6 +1,23 @@
 import copy
 from typing import Dict, List
 
+from loguru import logger
+
+
+def log_empty_content_messages(messages: List[Dict], episode_uuid: str = "") -> None:
+    """Scan an OpenAI-compatible message list and log an error for any message
+    whose content is empty/None and which carries no tool_calls.
+    """
+    for idx, m in enumerate(messages or []):
+        content = m.get("content")
+        tool_calls = m.get("tool_calls") or []
+        if content in (None, "") and not tool_calls:
+            logger.error(
+                f"[{episode_uuid}] Empty content in inbound message "
+                f"index={idx} role={m.get('role')} tool_call_id={m.get('tool_call_id')!r} "
+                f"content={content!r} tool_calls={tool_calls}"
+            )
+
 
 # apply chat_template to a message, and then convert back to message
 def convert_tool_to_user_message(tool_message, tokenizer, format="qwen"):

tutorial/opencode_build_aime/agent_roll_v3.py

@@ -24,12 +24,13 @@
 
 REMOTE_MODEL_PATH = os.getenv("REMOTE_MODEL_PATH", "/mnt/data_cpfs/xielipeng.xlp/models/Qwen3-14B")
 BATCH_SIZE = 16
-PPO_EPOCH = 4
+PPO_EPOCH = 2
 NUM_REPEAT = 8
-MINI_BATCH_NUM = 1
+MINI_BATCH_NUM = 2
 ajet_job = AgentJetJob(
+    ensure_new_experiment=True,
     algorithm="grpo",
-    experiment_name="aime_swarm_14b_v33_ppoepoch4",
+    experiment_name="aime_swarm_14b_v33_ppoepoch4_v3",
     max_env_worker=128,
     n_gpu=8,
     model=REMOTE_MODEL_PATH,
@@ -148,6 +149,8 @@ def run_eval(self, n_global_step: int):
         """Run evaluation on AIME-2024 test set."""
         if not self.eval_tasks:
             return
+        eval_log_path = os.path.join(self.swarm_worker.server_experiment_dir(), "eval_results.log")
+        print(eval_log_path)
 
         k = self.EVAL_K
         total_rollouts = len(self.eval_tasks) * k
@@ -182,7 +185,6 @@ def run_eval(self, n_global_step: int):
                 f"n_tasks={len(per_task_rewards)}  n_rollouts={len(flat)}"
             )
             print(summary)
-            eval_log_path = os.path.join(os.path.dirname(__file__), "eval_results.log")
             with open(eval_log_path, "a") as f:
                 f.write(summary + "\n")
         else:
@@ -193,7 +195,7 @@ def run_eval(self, n_global_step: int):
     def train(self):
         """Main training loop."""
         # Run eval once before training starts (baseline)
-        # self.run_eval(0)
+        self.run_eval(0)
 
         task_count = 0
         max_parallel = 64

tutorial/opencode_build_aime/agent_run_v3.py

@@ -393,7 +393,7 @@ async def run(self, messages: list[dict], sampling_params: dict) -> tuple[str, l
                     except Exception as e:
                         tool_response = {"text": f"Error executing tool: {e}"}
 
-                    truncated_text = self._truncate_response(tool_response.get("text", ""))
+                    truncated_text = self._truncate_response(tool_response.get("text", "")) or "(no output)"
                     formatted_messages.append({
                         "role": "tool",
                         "content": truncated_text,