feat: 增强版的SubAgent功能#7108
Open
elecvoid243 wants to merge 65 commits intoAstrBotDevs:masterfrom
Open
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request introduces an enhanced dynamic subagent management system, allowing the main agent to create, manage, and communicate with specialized subagents. Key additions include a DynamicSubAgentManager for lifecycle and shared context handling, a dedicated SubAgentLogger, and integration into the main agent's toolset and prompt construction. Feedback focuses on improving error handling by avoiding broad exception silences, ensuring safe string formatting for system prompts, and refining the logic for detecting subagent creation failures. A minor typo in the subagent capability prompt was also identified.
Soulter
force-pushed
the
master
branch
2 times, most recently
from
faf411f to
0068960
Compare
Author
|
@sourcery-ai review |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 6 issues, and left some high level feedback:
- In SubAgentManager.remove_subagent you directly index session.subagent_status[agent_name] before checking for
sessionexistence or whetheragent_nameis present (and even whenagent_name == 'all'), which can raise KeyError/AttributeError; guard against a missing session and use.get()or early returns before accessing this dict. - In FunctionToolExecutor._execute_handoff the timeout branch calls
_save_subagent_history(agent_name, runner_messages, umo)but the method signature is(umo, runner_messages, agent_name), so history will be written under the wrong key; swap the argument order to match the signature. - WaitForSubagentTool.call assumes
resultis non-None in some error paths (e.g., FAILED branchelsecase usesresult.errorwhenresultmay be None), which will raise; add a separate branch forresult is Noneand avoid dereferencing it there.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In SubAgentManager.remove_subagent you directly index session.subagent_status[agent_name] before checking for `session` existence or whether `agent_name` is present (and even when `agent_name == 'all'`), which can raise KeyError/AttributeError; guard against a missing session and use `.get()` or early returns before accessing this dict.
- In FunctionToolExecutor._execute_handoff the timeout branch calls `_save_subagent_history(agent_name, runner_messages, umo)` but the method signature is `(umo, runner_messages, agent_name)`, so history will be written under the wrong key; swap the argument order to match the signature.
- WaitForSubagentTool.call assumes `result` is non-None in some error paths (e.g., FAILED branch `else` case uses `result.error` when `result` may be None), which will raise; add a separate branch for `result is None` and avoid dereferencing it there.
## Individual Comments
### Comment 1
<location path="astrbot/core/astr_agent_tool_exec.py" line_range="414" />
<code_context>
+ )
+ except asyncio.TimeoutError:
+ # 若超时，保存已产生的部分历史
+ cls._save_subagent_history(agent_name, runner_messages, umo)
+ error_msg = f"SubAgent '{agent_name}' execution timeout after {execution_timeout:.1f} seconds."
+ logger.warning(f"[SubAgent:Timeout] {error_msg}")
</code_context>
<issue_to_address>
**issue (bug_risk):** Arguments to `_save_subagent_history` are passed in the wrong order in the timeout path.
`_save_subagent_history` is defined as `(umo: str, runner_messages: list[Message], agent_name: str)`, but in the timeout branch you call `cls._save_subagent_history(agent_name, runner_messages, umo)`. This swaps `umo` and `agent_name`, so history is written under the wrong key. Match the normal path and call `cls._save_subagent_history(umo, runner_messages, agent_name)` here.
</issue_to_address>
### Comment 2
<location path="astrbot/core/subagent_manager.py" line_range="1078-164" />
<code_context>
+ return session.subagent_background_results[agent_name].get(task_id, None)
+
+ @classmethod
+ def has_subagent_result(
+ cls, session_id: str, agent_name: str, task_id: str | None = None
+ ) -> bool:
+ """检查 SubAgent 是否有结果
+
+ Args:
+ session_id: Session ID
+ agent_name: SubAgent 名称
+ task_id: 任务ID，如果为None则检查是否有任何已完成的任务
+ """
+ session = cls.get_session(session_id)
+ task_store = cls._ensure_task_store(session, agent_name)
+ if not session or not task_store:
</code_context>
<issue_to_address>
**issue (bug_risk):** `has_subagent_result` can raise when session is missing due to calling `_ensure_task_store` on `None`.
`session` is retrieved before `_ensure_task_store` is called, but if `cls.get_session(session_id)` returns `None`, `_ensure_task_store` is invoked with `None` and can raise before the `if not session or not task_store` check runs. Please guard against `session is None` and return early, then call `_ensure_task_store` only when a valid session exists, consistent with other accessors.
</issue_to_address>
### Comment 3
<location path="astrbot/core/subagent_manager.py" line_range="1167-1170" />
<code_context>
+ "type": "string",
+ "description": "Subagent system_prompt",
+ },
+ "tools": {
+ "type": "array",
+ "items": {"type": "string"},
+ "description": "Tools available to subagent, can be empty.",
+ },
+ "skills": {
</code_context>
<issue_to_address>
**nitpick (bug_risk):** The `tools`/`skills` defaults in `CreateSubAgentTool.call` don't align with the JSON schema types.
In `CreateSubAgentTool.call`, `tools = kwargs.get("tools", {})` and `skills = kwargs.get("skills", {})` conflict with the JSON schema, which defines both as arrays. While `set({})` currently yields an empty set, using `{}` as the default is type-misleading and brittle. Prefer `None`/`[]` and normalize before `set`, e.g. `tools = set(kwargs.get("tools") or [])`, to align with the schema and avoid subtle type issues if callers pass unexpected shapes.
</issue_to_address>
### Comment 4
<location path="astrbot/core/subagent_manager.py" line_range="901" />
<code_context>
+ return list(session.handoff_tools.values())
+
+ @classmethod
+ def create_pending_subagent_task(cls, session_id: str, agent_name: str) -> str:
+ """为 SubAgent 创建一个 pending 任务，返回 task_id
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the background-task dict handling into a dedicated BackgroundTaskStore helper and moving path-safety checks into a shared utility to simplify SubAgentManager and CreateSubAgentTool without changing behavior.
You can reduce complexity in a few high‑impact, low‑risk spots without changing behavior. Two concrete refactors:
---
### 1. Encapsulate background task dict juggling in a `BackgroundTaskStore`
All of these methods manually juggle nested dicts on `session.subagent_background_results` / `session.background_task_counters`:
- `create_pending_subagent_task`
- `_ensure_task_store`
- `_is_task_completed`
- `get_pending_subagent_tasks`
- `get_latest_task_id`
- `store_subagent_result`
- `get_subagent_result`
- `has_subagent_result`
- `clear_subagent_result`
You can move the dict logic into a small object per agent and keep `SubAgentManager` methods thin. This keeps functionality but drops a lot of low-level dict manipulation from the manager.
Example (core idea, minimal change):
```python
# new helper, can live in same module initially
@dataclass
class BackgroundTaskStore:
results: dict[str, SubAgentExecutionResult] = field(default_factory=dict)
counter: int = 0
def create_pending(self, agent_name: str) -> str:
self.counter += 1
task_id = str(self.counter)
self.results[task_id] = SubAgentExecutionResult(
task_id=task_id,
agent_name=agent_name,
success=False,
result=None,
created_at=time.time(),
metadata={},
)
return task_id
def pending_ids(self) -> list[str]:
return [
tid for tid, res in self.results.items()
if res.completed_at <= 0 and res.error is None
]
def latest_id(self) -> str | None:
if not self.results:
return None
tid, _ = max(self.results.items(), key=lambda x: x[1].created_at)
return tid
def store_result(
self,
agent_name: str,
success: bool,
result: str,
task_id: str | None = None,
error: str | None = None,
execution_time: float = 0.0,
metadata: dict | None = None,
) -> None:
if task_id is None:
pending = self.pending_ids()
if not pending:
return
task_id = pending[-1]
if task_id not in self.results:
self.results[task_id] = SubAgentExecutionResult(
task_id=task_id,
agent_name=agent_name,
success=False,
result="",
created_at=time.time(),
metadata=metadata or {},
)
r = self.results[task_id]
r.success = success
r.result = result
r.error = error
r.execution_time = execution_time
r.completed_at = time.time()
if metadata:
r.metadata.update(metadata)
def latest_completed(self) -> SubAgentExecutionResult | None:
completed = [
r for r in self.results.values()
if r.result != "" or r.completed_at > 0
]
if not completed:
return None
return max(completed, key=lambda r: r.created_at)
def get(self, task_id: str) -> SubAgentExecutionResult | None:
return self.results.get(task_id)
def has_result(self, task_id: str | None = None) -> bool:
if task_id is None:
return any(r.result != "" or r.completed_at > 0 for r in self.results.values())
r = self.results.get(task_id)
return bool(r and (r.result != "" or r.completed_at > 0))
def clear(self, task_id: str | None = None) -> None:
if task_id is None:
self.results.clear()
self.counter = 0
else:
self.results.pop(task_id, None)
```
Then in `SubAgentSession`:
```python
@dataclass
class SubAgentSession:
...
subagent_background_results: dict[str, BackgroundTaskStore] = field(default_factory=dict)
```
And `SubAgentManager` becomes mostly delegations:
```python
@classmethod
def _get_task_store(cls, session: SubAgentSession, agent_name: str) -> BackgroundTaskStore:
if agent_name not in session.subagent_background_results:
session.subagent_background_results[agent_name] = BackgroundTaskStore()
return session.subagent_background_results[agent_name]
@classmethod
def create_pending_subagent_task(cls, session_id: str, agent_name: str) -> str:
session = cls._get_or_create_session(session_id)
if session.subagent_status[agent_name] == "RUNNING":
return f"__PENDING_TASK_CREATE_FAILED__: Subagent {agent_name} already running"
store = cls._get_task_store(session, agent_name)
return store.create_pending(agent_name)
@classmethod
def get_pending_subagent_tasks(cls, session_id: str, agent_name: str) -> list[str]:
session = cls.get_session(session_id)
if not session:
return []
return cls._get_task_store(session, agent_name).pending_ids()
@classmethod
def store_subagent_result(...):
session = cls._get_or_create_session(session_id)
store = cls._get_task_store(session, agent_name)
store.store_result(agent_name, success, result, task_id, error, execution_time, metadata)
@classmethod
def get_subagent_result(...):
session = cls.get_session(session_id)
if not session or agent_name not in session.subagent_background_results:
return None
store = cls._get_task_store(session, agent_name)
return store.get(task_id) if task_id is not None else store.latest_completed()
@classmethod
def has_subagent_result(...):
session = cls.get_session(session_id)
if not session:
return False
store = cls._get_task_store(session, agent_name)
return store.has_result(task_id)
@classmethod
def clear_subagent_result(...):
session = cls.get_session(session_id)
if not session:
return
store = cls._get_task_store(session, agent_name)
store.clear(task_id)
```
This keeps the same external API and behavior but substantially shrinks the manager’s internal complexity around background tasks.
---
### 2. Extract `_check_path_safety` into a shared utility
`CreateSubAgentTool._check_path_safety` mixes OS‑specific logic into the tool class and duplicates environment concerns in this already-large module. You can move it to a separate helper (same module or `utils/path_safety.py`) and reuse it elsewhere if needed.
Example:
```python
# new file: astrbot/core/utils/path_safety.py
import os
import platform
def is_safe_workdir(path_str: str) -> bool:
if not path_str or not isinstance(path_str, str):
return False
if not os.path.isabs(path_str):
return False
try:
resolved = os.path.realpath(path_str)
except (OSError, ValueError):
return False
path_parts = {part.lower() for part in os.path.normpath(resolved).split(os.sep)}
system = platform.system().lower()
if system == "windows":
dangerous = {
"windows", "system32", "syswow64", "boot", "recovery",
"programdata", "$recycle.bin", "system volume information",
}
if path_parts & dangerous:
return False
elif system == "linux":
for prefix in ["/etc", "/bin", "/sbin", "/lib", "/lib64",
"/boot", "/dev", "/proc", "/sys", "/root"]:
if resolved == prefix or resolved.startswith(prefix + os.sep):
return False
elif system == "darwin":
for prefix in ["/System", "/Library", "/private/var", "/usr"]:
if resolved == prefix or resolved.startswith(prefix + os.sep):
return False
if ".." in path_str:
return False
if not os.path.exists(resolved):
return False
return True
```
Then `CreateSubAgentTool` becomes simpler and more testable:
```python
from astrbot.core.utils.path_safety import is_safe_workdir
@dataclass
class CreateSubAgentTool(FunctionTool):
...
async def call(self, context, **kwargs) -> str:
...
workdir = kwargs.get("workdir")
if not workdir or not is_safe_workdir(workdir):
workdir = get_astrbot_temp_path()
...
```
This removes system/path concerns from the tool class and centralizes safety logic in one place.
</issue_to_address>
### Comment 5
<location path="astrbot/core/astr_agent_tool_exec.py" line_range="370" />
<code_context>
- tool_call_timeout=run_context.tool_call_timeout,
- stream=stream,
+
+ # 获取子代理的历史上下文
+ subagent_history, agent_name = cls._load_subagent_history(umo, tool)
+ # 如果有历史上下文，合并到 contexts 中
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting shared timeout handling and SubAgentManager access into small helpers and making `_save_subagent_history` keyword-only to simplify subagent orchestration logic and avoid subtle call bugs.
You can reduce the new complexity around subagent orchestration without changing behavior by:
1. **Collapsing duplicated timeout logic**
`_execute_handoff` and `_do_handoff_background` both wrap inner coroutines with `asyncio.wait_for`, handle `TimeoutError`, and do subagent-specific reporting. Extract that into a small shared helper so each caller reads linearly.
For example:
```python
@staticmethod
async def _run_subagent_with_timeout(
coro: T.Coroutine[T.Any, T.Any, T.Any],
*,
timeout: float,
on_timeout: T.Callable[[], None],
):
if timeout > 0:
try:
return await asyncio.wait_for(coro, timeout=timeout)
except asyncio.TimeoutError:
on_timeout()
raise
else:
return await coro
```
Then in `_execute_handoff`:
```python
execution_timeout = cls._get_subagent_execution_timeout()
async def _run_subagent():
return await ctx.tool_loop_agent(
event=event,
chat_provider_id=prov_id,
prompt=input_,
image_urls=image_urls,
system_prompt=subagent_system_prompt,
tools=toolset,
contexts=contexts,
max_steps=agent_max_step,
tool_call_timeout=run_context.tool_call_timeout,
stream=stream,
runner_messages=runner_messages,
)
def _on_timeout():
cls._save_subagent_history(umo=umo, runner_messages=runner_messages, agent_name=agent_name)
error_msg = f"SubAgent '{agent_name}' execution timeout after {execution_timeout:.1f} seconds."
logger.warning(f"[SubAgent:Timeout] {error_msg}")
cls._handle_subagent_timeout(umo=umo, agent_name=agent_name)
try:
llm_resp = await cls._run_subagent_with_timeout(
_run_subagent(), timeout=execution_timeout, on_timeout=_on_timeout
)
except asyncio.TimeoutError:
yield mcp.types.CallToolResult(
content=[mcp.types.TextContent(type="text", text=f"error: SubAgent '{agent_name}' execution timeout after {execution_timeout:.1f} seconds.")]
)
return
```
And in `_do_handoff_background`:
```python
execution_timeout = cls._get_subagent_execution_timeout()
async def _run():
nonlocal result_text
async for r in cls._execute_handoff(
tool,
run_context,
image_urls_prepared=True,
**tool_args,
):
if isinstance(r, mcp.types.CallToolResult):
for content in r.content:
if isinstance(content, mcp.types.TextContent):
result_text += content.text + "\n"
def _on_timeout():
nonlocal error_text, result_text
error_text = f"Execution timeout after {execution_timeout:.1f} seconds."
result_text = f"error: Background SubAgent '{agent_name}' {error_text}"
logger.warning(f"[SubAgent:BackgroundTask] {error_text}")
try:
await cls._run_subagent_with_timeout(
_run(), timeout=execution_timeout, on_timeout=_on_timeout
)
except asyncio.TimeoutError:
# _on_timeout already set error_text/result_text
pass
```
This keeps all features (history save, status update, logging) but removes duplicated timeout/control-flow branches and makes the core logic easier to follow.
2. **Tighten the `SubAgentManager` coupling behind small helpers**
Several helpers repetitively reach into `SubAgentManager` (`_register_subagent_task`, `_is_managed_subagent`, `_handle_subagent_timeout`, parts of `_execute_handoff_background`). You can introduce a tiny, *local* façade instead of scattering direct imports:
```python
class _SubagentRuntime:
def __init__(self, umo: str, agent_name: str | None):
from astrbot.core.subagent_manager import SubAgentManager
self._mgr = SubAgentManager
self.umo = umo
self.agent_name = agent_name
def is_managed(self) -> bool:
if not self.agent_name:
return False
session = self._mgr.get_session(self.umo)
return bool(session and self.agent_name in session.subagents)
def set_status(self, status: str) -> None:
if self.agent_name:
self._mgr.set_subagent_status(
session_id=self.umo, agent_name=self.agent_name, status=status
)
def create_pending_task(self) -> str | None:
# inner logic from _register_subagent_task, simplified…
...
```
Then `_execute_handoff_background`, `_do_handoff_background`, `_handle_subagent_timeout`, `_is_managed_subagent`, and parts of `_handle_subagent_background_result` can operate through `_SubagentRuntime` instead of each doing their own session lookups/imports. This reduces the “spiderweb” coupling without changing behavior.
3. **Make `_save_subagent_history` harder to misuse**
In the timeout branch of `_execute_handoff` you currently call:
```python
cls._save_subagent_history(agent_name, runner_messages, umo)
```
while the signature is:
```python
def _save_subagent_history(umo: str, runner_messages: list[Message], agent_name: str) -> None:
```
To both fix argument order and reduce future complexity/bugs, make the helper keyword-only and always call it with names:
```python
@staticmethod
def _save_subagent_history(
*, umo: str, runner_messages: list[Message], agent_name: str
) -> None:
if agent_name and runner_messages:
from astrbot.core.subagent_manager import SubAgentManager
SubAgentManager.update_subagent_history(umo, agent_name, runner_messages)
```
Usage:
```python
cls._save_subagent_history(
umo=umo, runner_messages=runner_messages, agent_name=agent_name
)
```
This keeps functionality identical but removes a subtle source of bugs and cognitive load.
</issue_to_address>
### Comment 6
<location path="astrbot/core/agent/runners/tool_loop_agent_runner.py" line_range="1020" />
<code_context>
if not req.func_tool:
return
+ # Prefer dynamic tools when available
+ func_tool = self._resolve_dynamic_subagent_tool(func_tool_name)
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting small helper methods to centralize tool resolution, session/handoff access, and interruption-message building so the main loop and abort logic stay simpler and more readable.
You can keep all the new behavior but reduce complexity and duplication by extracting a few small helpers and unifying the tool resolution path.
### 1. Unify tool resolution (avoid double-resolution + branching in-loop)
Right now `_execute_tool_calls` effectively computes `func_tool` multiple times:
- You first call `_resolve_dynamic_subagent_tool`.
- Then you re-run the old `skills_like` / `req.func_tool.get_tool` logic, potentially overwriting the dynamic result.
This both increases mental load and risks subtle bugs. A small helper that encapsulates all resolution rules keeps the loop simpler:
```python
def _get_func_tool(self, func_tool_name: str):
# Prefer dynamic tools when available
func_tool = self._resolve_dynamic_subagent_tool(func_tool_name)
if func_tool is not None:
return func_tool
# Fallback to existing tool schema modes
if (
self.tool_schema_mode == "skills_like"
and self._skill_like_raw_tool_set
):
return self._skill_like_raw_tool_set.get_tool(func_tool_name)
if self.req.func_tool:
return self.req.func_tool.get_tool(func_tool_name)
return None
```
Then in the main loop:
```python
if not req.func_tool:
return
func_tool = self._get_func_tool(func_tool_name)
if func_tool is None:
# keep existing error/reporting behavior
...
```
This keeps dynamic routing logic out of the loop body and centralizes the policy.
### 2. Deduplicate session / handoff access for dynamic tools
Both `_resolve_dynamic_subagent_tool` and `_maybe_register_dynamic_tool_from_result` manually walk `run_context.context.event` and import `SubAgentManager`. That coupling can be kept but hidden behind a narrow helper so the runner doesn’t repeat low-level details:
```python
def _get_session_handoffs(self):
run_context_context = getattr(self.run_context, "context", None)
event = getattr(run_context_context, "event", None) if run_context_context else None
session_id = getattr(event, "unified_msg_origin", None) if event else None
if not session_id:
return None, []
try:
from astrbot.core.subagent_manager import SubAgentManager
handoffs = SubAgentManager.get_handoff_tools_for_session(session_id)
except Exception as e:
logger.warning(f"[SubAgent] Failed to load dynamic handoffs: {e}")
return session_id, []
return session_id, handoffs
```
Use it in both places:
```python
def _resolve_dynamic_subagent_tool(self, func_tool_name: str):
session_id, dynamic_handoffs = self._get_session_handoffs()
if not session_id:
return None
for h in dynamic_handoffs:
if h.name == func_tool_name or f"transfer_to_{h.name}" == func_tool_name:
return h
return None
```
```python
def _maybe_register_dynamic_tool_from_result(self, result_content: str) -> None:
parse_result = self._parse_dynamic_tool_creation(result_content)
if not parse_result:
return
new_tool_name, new_tool_obj_name = parse_result
session_id, handoffs = self._get_session_handoffs()
if not session_id:
return
for handoff in handoffs:
if (
handoff.name == new_tool_obj_name
or handoff.name == new_tool_name.replace("transfer_to_", "")
):
if self.req.func_tool:
self.req.func_tool.add_tool(handoff)
logger.info(f"[SubAgent] Added {handoff.name} to func_tool set")
break
```
And the parser stays small and focused:
```python
def _parse_dynamic_tool_creation(self, result_content: str) -> tuple[str, str] | None:
if not result_content.startswith("__DYNAMIC_TOOL_CREATED__:"):
return None
parts = result_content.split(":", 3)
if len(parts) < 4:
return None
return parts[1], parts[2] # new_tool_name, new_tool_obj_name
```
This keeps the magic-string protocol and `SubAgentManager` knowledge in one place, reducing complexity elsewhere.
### 3. Simplify abort-finalization branching
`_finalize_aborted_step` now mixes logging, message selection, and state transitions. The only real variability is the interruption message. You can make that explicit:
```python
def _build_interruption_message(self, manual_stop: bool) -> str:
if manual_stop:
return (
"[SYSTEM: SubAgent was manually stopped by main agent. "
"Partial output before interruption is preserved.]"
)
return self.USER_INTERRUPTION_MESSAGE
```
Then `_finalize_aborted_step` becomes easier to scan:
```python
async def _finalize_aborted_step(
self,
llm_resp: LLMResponse | None = None,
manual_stop: bool = False,
) -> AgentResponse:
logger.info(
"SubAgent execution was manually stopped by main agent."
if manual_stop
else "Agent execution was requested to stop by user."
)
if llm_resp is None:
llm_resp = LLMResponse(role="assistant", completion_text="")
if llm_resp.role != "assistant":
llm_resp = LLMResponse(
role="assistant",
completion_text=self._build_interruption_message(manual_stop),
)
self.final_llm_resp = llm_resp
self._aborted = True
self._transition_state(AgentState.DONE)
self.stats.end_time = time.time()
...
```
This preserves the manual vs user-stop behavior while localizing the branching logic, so the method’s main flow (log → normalize llm_resp → finalize) is clearer.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| return list(session.handoff_tools.values()) | ||
|
|
||
| @classmethod | ||
| def create_pending_subagent_task(cls, session_id: str, agent_name: str) -> str: |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider extracting the background-task dict handling into a dedicated BackgroundTaskStore helper and moving path-safety checks into a shared utility to simplify SubAgentManager and CreateSubAgentTool without changing behavior.
You can reduce complexity in a few high‑impact, low‑risk spots without changing behavior. Two concrete refactors:
1. Encapsulate background task dict juggling in a BackgroundTaskStore
All of these methods manually juggle nested dicts on session.subagent_background_results / session.background_task_counters:
create_pending_subagent_task_ensure_task_store_is_task_completedget_pending_subagent_tasksget_latest_task_idstore_subagent_resultget_subagent_resulthas_subagent_resultclear_subagent_result
You can move the dict logic into a small object per agent and keep SubAgentManager methods thin. This keeps functionality but drops a lot of low-level dict manipulation from the manager.
Example (core idea, minimal change):
# new helper, can live in same module initially
@dataclass
class BackgroundTaskStore:
results: dict[str, SubAgentExecutionResult] = field(default_factory=dict)
counter: int = 0
def create_pending(self, agent_name: str) -> str:
self.counter += 1
task_id = str(self.counter)
self.results[task_id] = SubAgentExecutionResult(
task_id=task_id,
agent_name=agent_name,
success=False,
result=None,
created_at=time.time(),
metadata={},
)
return task_id
def pending_ids(self) -> list[str]:
return [
tid for tid, res in self.results.items()
if res.completed_at <= 0 and res.error is None
]
def latest_id(self) -> str | None:
if not self.results:
return None
tid, _ = max(self.results.items(), key=lambda x: x[1].created_at)
return tid
def store_result(
self,
agent_name: str,
success: bool,
result: str,
task_id: str | None = None,
error: str | None = None,
execution_time: float = 0.0,
metadata: dict | None = None,
) -> None:
if task_id is None:
pending = self.pending_ids()
if not pending:
return
task_id = pending[-1]
if task_id not in self.results:
self.results[task_id] = SubAgentExecutionResult(
task_id=task_id,
agent_name=agent_name,
success=False,
result="",
created_at=time.time(),
metadata=metadata or {},
)
r = self.results[task_id]
r.success = success
r.result = result
r.error = error
r.execution_time = execution_time
r.completed_at = time.time()
if metadata:
r.metadata.update(metadata)
def latest_completed(self) -> SubAgentExecutionResult | None:
completed = [
r for r in self.results.values()
if r.result != "" or r.completed_at > 0
]
if not completed:
return None
return max(completed, key=lambda r: r.created_at)
def get(self, task_id: str) -> SubAgentExecutionResult | None:
return self.results.get(task_id)
def has_result(self, task_id: str | None = None) -> bool:
if task_id is None:
return any(r.result != "" or r.completed_at > 0 for r in self.results.values())
r = self.results.get(task_id)
return bool(r and (r.result != "" or r.completed_at > 0))
def clear(self, task_id: str | None = None) -> None:
if task_id is None:
self.results.clear()
self.counter = 0
else:
self.results.pop(task_id, None)Then in SubAgentSession:
@dataclass
class SubAgentSession:
...
subagent_background_results: dict[str, BackgroundTaskStore] = field(default_factory=dict)And SubAgentManager becomes mostly delegations:
@classmethod
def _get_task_store(cls, session: SubAgentSession, agent_name: str) -> BackgroundTaskStore:
if agent_name not in session.subagent_background_results:
session.subagent_background_results[agent_name] = BackgroundTaskStore()
return session.subagent_background_results[agent_name]
@classmethod
def create_pending_subagent_task(cls, session_id: str, agent_name: str) -> str:
session = cls._get_or_create_session(session_id)
if session.subagent_status[agent_name] == "RUNNING":
return f"__PENDING_TASK_CREATE_FAILED__: Subagent {agent_name} already running"
store = cls._get_task_store(session, agent_name)
return store.create_pending(agent_name)
@classmethod
def get_pending_subagent_tasks(cls, session_id: str, agent_name: str) -> list[str]:
session = cls.get_session(session_id)
if not session:
return []
return cls._get_task_store(session, agent_name).pending_ids()
@classmethod
def store_subagent_result(...):
session = cls._get_or_create_session(session_id)
store = cls._get_task_store(session, agent_name)
store.store_result(agent_name, success, result, task_id, error, execution_time, metadata)
@classmethod
def get_subagent_result(...):
session = cls.get_session(session_id)
if not session or agent_name not in session.subagent_background_results:
return None
store = cls._get_task_store(session, agent_name)
return store.get(task_id) if task_id is not None else store.latest_completed()
@classmethod
def has_subagent_result(...):
session = cls.get_session(session_id)
if not session:
return False
store = cls._get_task_store(session, agent_name)
return store.has_result(task_id)
@classmethod
def clear_subagent_result(...):
session = cls.get_session(session_id)
if not session:
return
store = cls._get_task_store(session, agent_name)
store.clear(task_id)This keeps the same external API and behavior but substantially shrinks the manager’s internal complexity around background tasks.
2. Extract _check_path_safety into a shared utility
CreateSubAgentTool._check_path_safety mixes OS‑specific logic into the tool class and duplicates environment concerns in this already-large module. You can move it to a separate helper (same module or utils/path_safety.py) and reuse it elsewhere if needed.
Example:
# new file: astrbot/core/utils/path_safety.py
import os
import platform
def is_safe_workdir(path_str: str) -> bool:
if not path_str or not isinstance(path_str, str):
return False
if not os.path.isabs(path_str):
return False
try:
resolved = os.path.realpath(path_str)
except (OSError, ValueError):
return False
path_parts = {part.lower() for part in os.path.normpath(resolved).split(os.sep)}
system = platform.system().lower()
if system == "windows":
dangerous = {
"windows", "system32", "syswow64", "boot", "recovery",
"programdata", "$recycle.bin", "system volume information",
}
if path_parts & dangerous:
return False
elif system == "linux":
for prefix in ["/etc", "/bin", "/sbin", "/lib", "/lib64",
"/boot", "/dev", "/proc", "/sys", "/root"]:
if resolved == prefix or resolved.startswith(prefix + os.sep):
return False
elif system == "darwin":
for prefix in ["/System", "/Library", "/private/var", "/usr"]:
if resolved == prefix or resolved.startswith(prefix + os.sep):
return False
if ".." in path_str:
return False
if not os.path.exists(resolved):
return False
return TrueThen CreateSubAgentTool becomes simpler and more testable:
from astrbot.core.utils.path_safety import is_safe_workdir
@dataclass
class CreateSubAgentTool(FunctionTool):
...
async def call(self, context, **kwargs) -> str:
...
workdir = kwargs.get("workdir")
if not workdir or not is_safe_workdir(workdir):
workdir = get_astrbot_temp_path()
...This removes system/path concerns from the tool class and centralizes safety logic in one place.
Author
|
重构实现，将原版和改版的subagent管理进行了统一，并通过配置逐项开关功能，满足不同任务下的定制化使用方式：临时的用完即弃型subagent / 复用于多轮对话的长任务subagent / 静态的预定义型subagent / 运行时动态创建的subagent等 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Modifications / 改动点
一、概述
增强版SubAgent对AstrBot的Subagent进行了功能扩充，支持两种使用方式：
subagent_orchestrator.agents在配置文件中预定义子代理（原版实现）所有子代理（无论静态还是动态）都由统一的
SubAgentManager管理，支持以下高级功能：独立上下文记忆、独立工具隔离、工作目录隔离、技能隔离、上下文共享区域、超时限制、后台任务模式等。resolved #6954
resolved #7605
核心特性
SubAgentManager统一管理create_subagent在运行时动态创建子代理wait_for_subagent阻塞式等待输出二、配置说明
2.1 配置项说明
在
cmd_config.json中通过subagent_orchestrator配置块进行设置：2.2 配置项详解
main_enablefalseremove_main_duplicate_toolsfalserouter_system_prompt""agents[]enabledfalsemax_dynamic_subagent_count3auto_cleanup_per_turntruetools_blacklisttools_inherenthistory_enabledtruetrue可以保留跨多轮对话的记忆shared_context_enabledtrueshared_context_maxlen300max_subagent_history300execution_timeout1200.0三、核心文件
3.1 subagent_manager.py
文件路径：
astrbot/core/subagent_manager.py核心数据结构：
核心类：
SubAgentManagerconfigure()get_execution_timeout()is_auto_cleanup_per_turn()is_shared_context_enabled()register_blacklisted_tool(tool_name)register_inherent_tool(tool_name)create_subagent(session_id, config, protected=False)register_static_subagent(session_id, handoff_tool, skills, workdir)remove_subagent(session_id, agent_name)cleanup_session_turn_end(session_id)protect_subagent(session_id, agent_name)is_protected(session_id, agent_name)update_subagent_history(session_id, agent_name, current_messages)get_subagent_history(session_id, agent_name)clear_subagent_history(session_id, agent_name)build_static_subagent_prompts(session_id, agent_name)build_dynamic_subagent_prompts(session_id, agent_name, runtime)build_task_router_prompt(session_id)get_subagent_tools(session_id, agent_name)add_shared_context(session_id, sender, context_type, content, target)get_shared_context(session_id, filter_by_agent)_build_shared_context_prompt(session_id, agent_name)create_pending_subagent_task(session_id, agent_name)get_pending_subagent_tasks(session_id, agent_name)get_latest_task_id(session_id, agent_name)store_subagent_result(session_id, agent_name, success, result, task_id, error, execution_time, metadata)get_subagent_result(session_id, agent_name, task_id)has_subagent_result(session_id, agent_name, task_id)clear_subagent_result(session_id, agent_name, task_id)get_subagent_status(session_id, agent_name)get_all_subagent_status(session_id)set_subagent_status(session_id, agent_name, status)cleanup_shared_context_by_agent(session_id, agent_name)clear_shared_context(session_id)set_shared_context_enabled(session_id, enabled)get_handoff_tools_for_session(session_id)cleanup_session(session_id)四、功能实现
4.1 静态子代理
功能描述
静态子代理是原版的子代理实现，由配置文件定义人格。
为了实现统一管理，静态子代理在
build_main_agent()时自动注册到SubAgentManager，可以享受历史记忆、公共上下文、Skills调用等增强功能（如果开启）。实现方式
astrbot/core/subagent_orchestrator.pySubAgentOrchestrator.register_static_subagents_to_manager()astr_main_agent.py的_apply_subagent_tools()中注册流程
build_main_agent()构建主Agent时调用_apply_subagent_tools()_apply_subagent_tools()调用so.register_static_subagents_to_manager(session_id)SubAgentOrchestrator.handoffs中的所有静态 handoff 工具SubAgentManager.register_static_subagent()register_static_subagent()内部：HandoffTool提取 agent 信息SubAgentConfigcreate_subagent(session_id, config, protected=True)4.2 动态子代理
功能描述
当动态子代理功能开启时，主Agent可以通过
create_subagent工具动态创建子代理，每个子代理拥有独立的人设、工具、技能和工作目录配置。随后系统会创建对应的transfer_to_xxx工具实现方式
astrbot/core/subagent_manager.pyCreateSubAgentToolname: 子代理名称system_prompt: 子代理的人设和系统提示tools: 可用工具列表（字符串名称）skills: 可用技能列表（字符串名称）workdir: 子代理工作目录（绝对路径，可选）使用示例
名称验证规则
^[a-zA-Z][a-zA-Z0-9_]{0,31}$）创建后流程
astrbot_execute_shell,astrbot_execute_python。若公共上下文启用，还会添加send_shared_context）SubAgentConfig配置对象Agent和HandoffTool对象SubAgentManager__DYNAMIC_TOOL_CREATED__标记的消息，触发工具schema刷新工作目录安全检查
CreateSubAgentTool._check_path_safety()对传入的workdir进行安全验证：..windows,system32,syswow64,boot等/etc,/bin,/sbin,/root等/System,/Library,/private/var,/usr等验证失败时，工作目录回退到
get_astrbot_temp_path()。工具黑名单与固有工具
4.3 动态子代理委派 (transfer_to_xxx)
功能描述
创建子代理后，主Agent使用
transfer_to_xxx工具将任务委派给对应子代理。实现方式
astrbot/core/astr_agent_tool_exec.pyFunctionToolExecutor._execute_handoff()/_execute_handoff_background()子代理System Prompt构建
将子代理的system_prompt构建分为静态部分和动态部分，尽可能增加缓存命中率
静态提示词
build_static_subagent_prompts()：动态提示词
build_dynamic_subagent_prompts()：委派执行流程
transfer_to_xxx工具ToolLoopAgentRunner._handle_function_tools()在SubAgentManager中查找handoff工具HandoffTool对象FunctionToolExecutor._execute_handoff()执行委派send_shared_context工具到子代理工具集provider_id使用不同的LLM提供商_get_subagent_execution_timeout()获取超时时间asyncio.wait_for()添加执行超时控制_handle_subagent_timeout()将状态设为 "FAILED"历史上下文注入流程
subagent_histories获取历史消息Message对象begin_dialogs之前runner_messages（Agent运行期间的所有消息）追加到历史max_subagent_history（默认300条），超出时保留最新的。（此处仅用于约束数组长度，实际上下文长度仍由astrbot的truncate管理）执行超时控制
4.4 后台任务等待
功能描述
当主Agent认为某个任务耗时很长时，可以让SubAgent以后台模式运行，主Agent不会被阻塞，可继续执行其他可并行的任务
原版Subagent结束任务时，会通过
_wake_main_agent_for_background_result唤醒主Agent如果Subagent耗时很长，一些聪明的Agent可能会执行
time.sleep()的python代码进行等待，但我们要假设Agent是愚蠢的。事实上，大多数情况下，主Agent会觉得迟迟拿不到结果而试图自己执行，导致同一任务完成两遍。因此引入了一个
wait_for_subagent工具，主Agent需要拿到Subagent结果时，可进行主动的阻塞式等待，避免此类情况发生实现方式
astrbot/core/astr_agent_tool_exec.py_execute_handoff_background()- 执行后台委派_do_handoff_background()- 后台任务执行逻辑_register_subagent_task()- 注册SubAgent任务_handle_subagent_background_result()- 处理SubAgent结果_maybe_wake_main_agent_after_background()- 智能唤醒主Agent。Subagent完成任务时，如果主Agent已经结束运行，才执行_wake_main_agent_for_background_result，否则把结果存到subagent_background_results里执行流程
主Agent状态检测
后台任务完成后，需要决定是否通知用户。关键逻辑：
这依赖
AstrAgentContext.extra字段，在build_main_agent()中注入：主动等待工具
WaitForSubagentTool 轮询逻辑
task_id，获取最新的 pending 任务IDLE→ 返回错误：子代理未在运行任务COMPLETED→ 返回执行结果FAILED→ 返回失败信息RUNNING→ 继续等待后台Subagent任务管理
每个subagent都会储存结果到
subagent_background_results中。同一个subagent可以有多个任务，通过task_id来区分（每次创建任务递增）创建任务
获取结果
4.5 子代理历史记忆
功能描述
history_enabled=true时，子代理可以保留跨轮对话的历史上下文，实现连续对话能力。实现方式
SubAgentSession.subagent_historiesmax_subagent_history（默认300条）历史管理机制
runner_messages（Agent运行期间的消息）追加到历史role=system的消息_MAX_TOOL_RESULT_LEN（2000字符）的tool结果截断，附加...[truncated]max_subagent_history时，保留最新的消息Message对象，合并到contexts历史清理
reset_subagent工具主动清除指定子代理的历史4.6 动态子代理工作目录隔离（软约束）
功能描述
每个子代理可配置独立的工作目录，未指定时默认使用
get_astrbot_temp_path()。实现方式
SubAgentConfig.workdirCreateSubAgentTool._check_path_safety()_build_workdir_prompt()注入的工作目录提示
4.7 动态子代理行为规范注入
功能描述
子代理自动注入安全模式、输出规范和时间信息等行为约束。
注入内容
角色定义
_build_subagent_system_prompt()：# Role Your name is {agent_name}(used for tool calling) {base_instructions}静态行为规范
_build_rule_prompt()：动态时间信息
_build_time_prompt()：# Current Time 2026-04-14 23:09 (CST)4.8 Skills隔离
功能描述
修复了Subagent无法使用skills的问题，每个子代理可分配不同的Skills，相互隔离。
注入逻辑
SubAgentConfig.skills）SkillManager.list_skills()获取所有可用技能build_skills_prompt()生成提示词build_dynamic_subagent_prompts()注入到子代理的system_prompt4.9 公共上下文
功能描述
当
shared_context_enabled=true时，维护一个所有子代理共享的、实时更新的群聊区域。在SubAgent每次调用LLM之前，公共上下文的内容会被注入到System Prompt中。公共上下文中每条信息的格式
上下文类型
statusmessagesystem公共上下文注入方式
按类型和优先级分组注入到子代理的 System Prompt 中，让Agent可以更清晰地获取共享上下文的信息，并知道哪些需要优先处理。
实现方式
astrbot/core/subagent_manager.pySubAgentManager._build_shared_context_prompt()Prompt结构
容量管理
shared_context_maxlen时，保留最近90%的消息公共上下文特点
4.10 主Agent路由提示
功能描述
当
dynamic_agents.enabled=true时，在主Agent的System Prompt中注入动态SubAgent能力说明，包括动态子代理配额信息、创建指南和生命周期说明。如果dynamic_agents.enabled=false，则仍然使用router_system_prompt动态子代理路由的注入内容
核心方法：
SubAgentManager.build_task_router_prompt()配额耗尽时：仍然显示代理能力说明和生命周期，但告知无法创建新子代理，仍可委派已有的子代理。
4.11 动态子代理的自动清理
功能描述
当
auto_cleanup_per_turn=true时，每轮对话结束后，自动清理产生的动态子代理。auto_cleanup_per_turn=true，但又想要保留某个子代理留作后续使用，可以由主agent调用protect_subagent管理工具为子代理添加保护，防止被自动清理。主Agent也可以根据需要，手动通过
unprotect_subagent移除保护，或直接通过remove_subagent工具清理某个子代理。清理规则
remove_subagent()进行清理（移除配置、handoff工具、历史、后台结果）触发时机
在
internal.py的 agent 结束流程中：4.12 动态子代理数量限制
功能描述
限制单个会话中子代理的最大数量，防止资源耗尽。
注意：替换已存在的同名子代理不增加计数。
4.13 执行超时控制
功能描述
除了原有的工具调用超时外，为每个子代理设置总的执行超时时间，避免无限等待。这是一项全局设置，对所有子代理有效
实现方式
SubAgentConfig.execution_timeout（默认1200秒）subagent_orchestrator.execution_timeoutSubAgentManager.get_execution_timeout()超时处理流程
五、工具列表
5.1 主Agent新增可用工具
create_subagentremove_subagentlist_subagentsreset_subagentprotect_subagentunprotect_subagentview_shared_contextsend_shared_context_for_main_agentwait_for_subagent5.2 子代理可用工具
send_shared_contexttools_inherent给出六、修改文件清单
astrbot/core/subagent_manager.pyDynamicSubAgentManager→SubAgentManager，DynamicSubAgentConfig→SubAgentConfig，DynamicSubAgentSession→SubAgentSession；工具名create_dynamic_subagent→create_subagent，remove_dynamic_subagent→remove_subagent，list_dynamic_subagents→list_subagentsastrbot/core/astr_agent_tool_exec.py_execute_handoff()（静态/动态Prompt构建、历史注入、超时控制）；_execute_handoff_background()（pending任务创建）；新增_do_handoff_background()（结果存储与智能唤醒逻辑）；新增多个工具方法（_load_subagent_history、_build_subagent_system_prompt、_save_subagent_history、_register_enhanced_subagent_task等）astrbot/core/astr_main_agent.py_ensure_persona_and_skills()新增_apply_subagent_manager_tools()，仅在enable_dynamic=True时使用。在MainAgentBuildConfig中修改配置字段，在build_main_agent()中注入main_agent_runner到AstrAgentContext.extraastrbot/core/subagent_orchestrator.pyregister_static_subagents_to_manager()方法；静态subagent自动注册到SubAgentManager享受增强功能astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.pyastrbot/core/agent/runners/tool_loop_agent_runner.py_handle_function_tools()中集成动态SubAgent工具查找；新增_resolve_dynamic_subagent_tool()和_maybe_register_dynamic_tool_from_result()方法astrbot/core/config/default.pyastrbot/core/astr_agent_context.pyextra字段的类型改成dict[str, Any]，使得可以记录任意类型的extra信息astrbot/core/star/context.pytool_loop_agent()支持runner_messages参数，用于记录SubAgent历史七、使用流程
7.1 静态子代理使用流程
与原版一致，但子代理获得增强
7.2 动态子代理使用流程
7.3 带保护的多轮对话流程
7.4 带公共上下文的协作流程
7.5 后台任务并行处理流程
7.6 后台任务与共享上下文结合
八、API 参考
8.1 SubAgentManager 核心方法
配置与生命周期
configure(max_dynamic_subagent_count, auto_cleanup_per_turn, shared_context_enabled, shared_context_maxlen, max_subagent_history, tools_blacklist, tools_whitelist, execution_timeout)get_execution_timeout() -> floatis_auto_cleanup_per_turn() -> boolis_shared_context_enabled() -> boolregister_blacklisted_tool(tool_name)register_inherent_tool(tool_name)子代理管理
create_subagent(session_id, config, protected=False) -> tupleregister_static_subagent(session_id, handoff_tool, skills, workdir) -> tupleremove_subagent(session_id, agent_name) -> strprotect_subagent(session_id, agent_name)is_protected(session_id, agent_name) -> boolset_subagent_status(session_id, agent_name, status)cleanup_session_turn_end(session_id) -> dictcleanup_session(session_id) -> dictget_handoff_tools_for_session(session_id) -> listget_subagent_tools(session_id, agent_name) -> list | None历史管理
update_subagent_history(session_id, agent_name, current_messages)get_subagent_history(session_id, agent_name) -> listclear_subagent_history(session_id, agent_name) -> strPrompt构建
build_task_router_prompt(session_id) -> strbuild_static_subagent_prompts(session_id, agent_name) -> strbuild_dynamic_subagent_prompts(session_id, agent_name, runtime) -> str后台任务管理
create_pending_subagent_task(session_id, agent_name) -> strget_pending_subagent_tasks(session_id, agent_name) -> list[str]get_latest_task_id(session_id, agent_name) -> str | Nonestore_subagent_result(session_id, agent_name, success, result, task_id, error, execution_time, metadata)get_subagent_result(session_id, agent_name, task_id) -> SubAgentExecutionResult | Nonehas_subagent_result(session_id, agent_name, task_id) -> boolclear_subagent_result(session_id, agent_name, task_id)get_subagent_status(session_id, agent_name) -> strget_all_subagent_status(session_id) -> dict公共上下文管理
add_shared_context(session_id, sender, context_type, content, target)get_shared_context(session_id, filter_by_agent) -> listset_shared_context_enabled(session_id, enabled)cleanup_shared_context_by_agent(session_id, agent_name)clear_shared_context(session_id)8.2 SubAgentExecutionResult 数据结构
8.3 SubAgentConfig 数据结构
8.4 SubAgentSession 数据结构
Screenshots or Test Results / 运行截图或测试结果
测试1：动态Agent创建和保护机制
测试流程：动态建立若干个SubAgent，给其中一些加入保护，另一些不加保护，观察清理情况，以及跨轮对话的记忆
测试1——动态创建子Agent和保护机制.md
测试2：子Agent历史上下文
测试流程：建立一个SubAgent，并使其跨多轮对话，查看其是否有多轮对话的记忆。
测试2——子Agent历史上下文.md
测试3：子Agent共享上下文
测试流程
send_shared_context发送工具向共享上下文中添加内容测试3——子Agent共享上下文.md
测试4：异步与等待
测试流程
设计一个同时包含并行与串行、前台与后台模式的任务，观察工作流程
测试4——异步与等待.md
测试5：超长上下文工作实例
测试流程
让agent阅读一个大型项目的全部代码，并生成细致的介绍文档
测试5——动态处理超长上下文示例.md
Checklist / 检查清单
😊 If there are new features added in the PR, I have discussed it with the authors through issues/emails, etc.
/ 如果 PR 中有新加入的功能，已经通过 Issue / 邮件等方式和作者讨论过。
👀 My changes have been well-tested, and "Verification Steps" and "Screenshots" have been provided above.
/ 我的更改经过了良好的测试，并已在上方提供了“验证步骤”和“运行截图”。
🤓 I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations in
requirements.txtandpyproject.toml./ 我确保没有引入新依赖库，或者引入了新依赖库的同时将其添加到
requirements.txt和pyproject.toml文件相应位置。😮 My changes do not introduce malicious code.
/ 我的更改没有引入恶意代码。
Summary by Sourcery
Introduce a unified SubAgentManager that manages both static and dynamically created sub-agents, adds enhanced subagent runtime features and background task controls, and exposes corresponding configuration, tooling, and dashboard UI to manage them.
New Features:
Enhancements:
Tests: