interchange zmq client to async

Author: binary-husky

ajet/backbone/warm_up.py

@@ -6,7 +6,11 @@
 import asyncio
 import logging
 import os
-from ajet.utils.async_utils import apply_httpx_aclose_patch, suppress_httpx_aclose_exception
+from ajet.utils.async_utils import (
+    apply_httpx_aclose_patch,
+    silence_hermes_tool_parser_loggers,
+    suppress_httpx_aclose_exception,
+)
 apply_httpx_aclose_patch()
 suppress_httpx_aclose_exception()
 
@@ -36,10 +40,7 @@ def init_parallel_rollout_logger(experiment_name, experiment_dir):
         debug=False,
     )
 
-    target_logger = logging.getLogger("vllm.entrypoints.openai.tool_parsers.hermes_tool_parser")
-    target_logger.setLevel(logging.CRITICAL)
-    target_logger = logging.getLogger("vllm.tool_parsers.hermes_tool_parser")
-    target_logger.setLevel(logging.CRITICAL)
+    silence_hermes_tool_parser_loggers()
     logging.getLogger("httpx").setLevel(logging.WARNING)
 
 

ajet/task_rollout/async_llm_bridge.py

@@ -12,6 +12,9 @@
     from vllm.entrypoints.openai.tool_parsers.hermes_tool_parser import Hermes2ProToolParser
 except:
     from vllm.tool_parsers.hermes_tool_parser import Hermes2ProToolParser   # vllm 0.17.x moved this class elsewhere
+
+from ajet.utils.async_utils import silence_hermes_tool_parser_loggers
+silence_hermes_tool_parser_loggers()
 from verl.workers.rollout.replica import TokenOutput
 from agentscope.model import ChatResponse as AgentScopeChatResponse
 from openai.types.chat.chat_completion import ChatCompletion as OpenAIChatCompletion

ajet/utils/async_utils.py

@@ -118,6 +118,19 @@ def filter(self, record):
     logging.getLogger('asyncio').addFilter(HttpxAcloseFilter())
 
 
+def silence_hermes_tool_parser_loggers():
+    """Silence the noisy vllm Hermes tool-parser loggers (import path varies across vllm versions)."""
+    for name in (
+        "vllm.entrypoints.openai.tool_parsers.hermes_tool_parser",
+        "vllm.tool_parsers.hermes_tool_parser",
+    ):
+        lg = logging.getLogger(name)
+        lg.setLevel(logging.CRITICAL)
+        for handler in lg.handlers:
+            handler.setLevel(logging.CRITICAL)
+        lg.propagate = False
+
+
 class IterationSafeDict(dict):
     """A dict subclass that creates snapshots during iteration to avoid RuntimeError when dict changes size."""