Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request migrates the project's documentation from Sphinx to MkDocs Material, introducing comprehensive quickstart guides, core concepts, and benchmark results. It also adds standalone inference examples, a sandbox demo, and updates training recipes to utilize verl's V1 unified trainer in colocate_async mode. The review feedback highlights a critical security issue with hardcoded Modal API credentials in the inference runtime environment. Additionally, several technical improvements are requested, such as lazily initializing asyncio.Semaphore in Ray actors to prevent runtime errors, implementing the stubbed run method in MiniSweAgentAgent to fix failing tests, using with blocks for safe file operations in task_runner.py, utilizing async with for OpenAICompatibleChatModel in the ReAct agent, and avoiding hardcoded paths in training shell scripts.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| MODAL_TOKEN_ID: "ak-MWuXc0JB73Ll3y75lnTl1K" | ||
| MODAL_TOKEN_SECRET: "as-b3iq9Xf3igKAb3DfPQVNwG" |
There was a problem hiding this comment.
Hardcoded Modal API credentials (MODAL_TOKEN_ID and MODAL_TOKEN_SECRET) have been committed to the repository. This poses a significant security risk.
Please replace these credentials with placeholders (similar to examples/quickstart/inference/runtime_env.yaml) and load them via environment variables or a local configuration file.
MODAL_TOKEN_ID: "<modal-token-id>"
MODAL_TOKEN_SECRET: "<modal-token-secret>"| class InferenceActor: | ||
| _semaphore = asyncio.Semaphore(max(1, GLOBAL_CONCURRENCY // NUM_WORKERS)) | ||
|
|
||
| async def run_single(self, sample: dict, task_defaults: dict) -> dict: | ||
| async with self._semaphore: |
There was a problem hiding this comment.
Creating an asyncio.Semaphore at class definition time (outside of any running event loop) is problematic. When this class is imported or instantiated on Ray workers, it can lead to RuntimeError: Task got Future attached to a different loop because the semaphore binds to the loop active during class loading (or no loop at all).
Instead, initialize the semaphore lazily inside the async method or within __init__ once the event loop is running.
class InferenceActor:
def __init__(self) -> None:
self._semaphore = None
async def run_single(self, sample: dict, task_defaults: dict) -> dict:
if self._semaphore is None:
self._semaphore = asyncio.Semaphore(max(1, GLOBAL_CONCURRENCY // NUM_WORKERS))
async with self._semaphore:| class TestEvalActor: | ||
| _semaphore = asyncio.Semaphore(max(1, GLOBAL_CONCURRENCY // NUM_WORKERS)) | ||
|
|
||
| async def run_single(self, sample: dict) -> dict: | ||
| async with self._semaphore: | ||
| task_config = sample["extra_info"]["tools_kwargs"]["task"] |
There was a problem hiding this comment.
Creating an asyncio.Semaphore at class definition time (outside of any running event loop) is problematic. When this class is imported or instantiated on Ray workers, it can lead to RuntimeError: Task got Future attached to a different loop because the semaphore binds to the loop active during class loading (or no loop at all).
Instead, initialize the semaphore lazily inside the async method or within __init__ once the event loop is running.
@ray.remote
class TestEvalActor:
def __init__(self) -> None:
self._semaphore = None
async def run_single(self, sample: dict) -> dict:
if self._semaphore is None:
self._semaphore = asyncio.Semaphore(max(1, GLOBAL_CONCURRENCY // NUM_WORKERS))
async with self._semaphore:| async def run( | ||
| self, | ||
| *, | ||
| sandbox: Sandbox, | ||
| messages: list[dict[str, Any]], | ||
| ) -> AgentResult: | ||
| pass |
There was a problem hiding this comment.
The run method for MiniSweAgentAgent is currently implemented as a stub (pass). However, a comprehensive test suite tests/uni_agent/agents/test_mini_swe_agent_agent.py has been added in this PR that asserts a full implementation of run returning an AgentResult.
Running the test suite will currently fail with an AssertionError because run returns None. Please implement the run method or mark the tests as skipped/expected to fail until the agent is fully implemented.
| entries = raw if isinstance(raw, list) else [raw] | ||
| index: dict[str, dict[str, Any]] = {} | ||
| for entry in entries: |
There was a problem hiding this comment.
The file is opened using open(path) without explicitly closing it or using a with statement. This can lead to unclosed file descriptor leaks.
Additionally, it is highly recommended to specify encoding="utf-8" when opening text files to ensure platform-independent behavior.
import yaml
with open(path, "r", encoding="utf-8") as f:
raw = yaml.safe_load(f)| toolbox = Toolbox.from_specs(cfg.tools, sandbox=sandbox) | ||
| model = OpenAICompatibleChatModel( | ||
| base_url=cfg.model.base_url, | ||
| api_key=cfg.model.api_key, | ||
| model_name=cfg.model.model_name, | ||
| sampling_params={ | ||
| "temperature": cfg.model.temperature, | ||
| "top_p": cfg.model.top_p, | ||
| "top_k": cfg.model.top_k, | ||
| }, | ||
| tools_schemas=toolbox.schemas(), | ||
| ) | ||
|
|
||
| transcript: list[dict[str, Any]] = list(messages) | ||
| for message in messages: | ||
| logger.info(f"{str(message.get('role', '')).upper()} PROMPT:\n{message.get('content', '')}") | ||
| trajectory_info: dict[str, Any] = { | ||
| "steps": 0, | ||
| "num_tool_calls": 0, | ||
| "timeouts": 0, | ||
| "errors": 0, | ||
| "total_tokens": 0, | ||
| "exit_reason": "unknown", | ||
| } | ||
| try: | ||
| async with toolbox.entered(retry=3, timeout=60): | ||
| for step_idx in range(1, cfg.max_steps + 1): | ||
| trajectory_info["steps"] = step_idx | ||
| stop_reason = await self.step(cfg, model, toolbox, transcript, trajectory_info) | ||
| if stop_reason != "completed": | ||
| trajectory_info["exit_reason"] = stop_reason | ||
| break | ||
| else: # loop ran the full step budget without an early stop | ||
| trajectory_info["exit_reason"] = "max_steps" | ||
| logger.warning(f"Reached max steps ({cfg.max_steps}) without finishing.") | ||
| except Exception as exc: # keep the partial transcript; the task buckets the failure | ||
| logger.exception("react loop failed at step %s", trajectory_info["steps"]) | ||
| trajectory_info["exit_reason"] = "unknown_error" | ||
| trajectory_info["error"] = f"{type(exc).__name__}: {exc}" | ||
| finally: | ||
| await model.aclose() # release the reused HTTP session |
There was a problem hiding this comment.
Since OpenAICompatibleChatModel implements __aenter__ and __aexit__ to manage its underlying HTTP session lifecycle, it is much more idiomatic and safer to use an async with block. This ensures that the model session is always closed cleanly, even if an exception is raised during initialization or before entering the main try block, preventing potential resource leaks.
toolbox = Toolbox.from_specs(cfg.tools, sandbox=sandbox)
async with OpenAICompatibleChatModel(
base_url=cfg.model.base_url,
api_key=cfg.model.api_key,
model_name=cfg.model.model_name,
sampling_params={
"temperature": cfg.model.temperature,
"top_p": cfg.model.top_p,
"top_k": cfg.model.top_k,
},
tools_schemas=toolbox.schemas(),
) as model:
transcript: list[dict[str, Any]] = list(messages)
for message in messages:
logger.info(f"{str(message.get('role', '')).upper()} PROMPT:\n{message.get('content', '')}")
trajectory_info: dict[str, Any] = {
"steps": 0,
"num_tool_calls": 0,
"timeouts": 0,
"errors": 0,
"total_tokens": 0,
"exit_reason": "unknown",
}
try:
async with toolbox.entered(retry=3, timeout=60):
for step_idx in range(1, cfg.max_steps + 1):
trajectory_info["steps"] = step_idx
stop_reason = await self.step(cfg, model, toolbox, transcript, trajectory_info)
if stop_reason != "completed":
trajectory_info["exit_reason"] = stop_reason
break
else: # loop ran the full step budget without an early stop
trajectory_info["exit_reason"] = "max_steps"
logger.warning(f"Reached max steps ({cfg.max_steps}) without finishing.")
except Exception as exc: # keep the partial transcript; the task buckets the failure
logger.exception("react loop failed at step %s", trajectory_info["steps"])
trajectory_info["exit_reason"] = "unknown_error"
trajectory_info["error"] = f"{type(exc).__name__}: {exc}"|
|
||
| project_name='Uni-Agent-SWE-Agent' | ||
| exp_name='GRPO-Qwen3-30B-R2E-Fully-Async' | ||
| RAY_DATA_HOME=/mnt/hdfs/yyding |
There was a problem hiding this comment.
Hardcoding RAY_DATA_HOME=/mnt/hdfs/yyding at the top of the script prevents users from overriding this path via environment variables, as the subsequent check RAY_DATA_HOME=${RAY_DATA_HOME:-...} will always see the hardcoded value.
Consider using the standard shell parameter expansion to allow environment overrides.
| RAY_DATA_HOME=/mnt/hdfs/yyding | |
| RAY_DATA_HOME=${RAY_DATA_HOME:-/mnt/hdfs/yyding} |
| # | ||
| # Launch from the repo root so Ray packages both `verl/` and `uni_agent/`. | ||
|
|
||
| RAY_DATA_HOME=/mnt/hdfs/yyding |
There was a problem hiding this comment.
Hardcoding RAY_DATA_HOME=/mnt/hdfs/yyding at the top of the script prevents users from overriding this path via environment variables, as the subsequent check RAY_DATA_HOME=${RAY_DATA_HOME:-...} will always see the hardcoded value.
Consider using the standard shell parameter expansion to allow environment overrides.
| RAY_DATA_HOME=/mnt/hdfs/yyding | |
| RAY_DATA_HOME=${RAY_DATA_HOME:-/mnt/hdfs/yyding} |
2 participants
What does this PR do?
This PR refactors Uni-Agent around a unified set of composable runtime abstractions for agent inference, data synthesis, evaluation, and reinforcement learning.
New docs: https://uni-agent.readthedocs.io/en/latest/
Key changes:
Task,Agent,Sandbox,Tool, andToolboxinterfaces.Related:
Checklist Before Starting
Test
The following targeted checks were run:
API and Usage Example
A workload is now defined through one Task Config:
Tasks can also be instantiated directly:
Task configuration is resolved as:
This allows sample-wise customization while keeping model endpoints and credentials bound to the live runtime.
Design & Code Changes
Core abstractions
Taskowns the episode lifecycle, prompt, metadata, Sandbox, Agent, and reward computation.Agentdefines the solving strategy and supports white-box loops or black-box harnesses.Sandboxprovides a shared lifecycle and execution/data-plane interface across Local, Modal, veFaaS, and Seed backends.Tooldefines one model-visible action;Toolboxbinds stateful Tool instances to one Sandbox.All major abstractions use typed Pydantic configurations, registries, and lazy module loading.
Gateway and verl integration
The verl-managed rollout path is:
The Gateway supports OpenAI and Anthropic-compatible sessions and captures token-level trajectories for the training pipeline.
Inference workflows
Two inference modes are provided:
Examples live under:
Reliability and extensibility
Documentation
Breaking Changes
This PR intentionally replaces several legacy interfaces:
AgentEnv, interaction, deployment, and global reward-registry architecture.Existing integrations will need to migrate to the new registries and configuration format.
Checklist Before Submitting
pre-commit run --all-files --show-diff-on-failure --color=always