diff --git a/astrbot/core/astr_main_agent.py b/astrbot/core/astr_main_agent.py
index 1c4fd400a0..8bb8f801a0 100644
--- a/astrbot/core/astr_main_agent.py
+++ b/astrbot/core/astr_main_agent.py
@@ -238,6 +238,20 @@ async def _get_session_conv(
return conversation
+def _wrap_system_block(name: str, content: str | None) -> str:
+ """Wrap a system prompt fragment with clear XML-like boundaries.
+
+ This keeps persona, skills, runtime, and tool-use instructions visually
+ separated even though the provider receives them as one system prompt.
+ """
+ if not content:
+ return ""
+ content = str(content).strip()
+ if not content:
+ return ""
+ return f"\n<{name}>\n{content}\n{name}>\n"
+
+
async def _apply_kb(
event: AstrMessageEvent,
req: ProviderRequest,
@@ -256,8 +270,9 @@ async def _apply_kb(
if not kb_result:
return
if req.system_prompt is not None:
- req.system_prompt += (
- f"\n\n[Related Knowledge Base Results]:\n{kb_result}"
+ req.system_prompt += _wrap_system_block(
+ "related_knowledge_base_results",
+ kb_result,
)
except Exception as exc: # noqa: BLE001
logger.error("Error occurred while retrieving knowledge base: %s", exc)
@@ -359,12 +374,13 @@ def _apply_workspace_extra_prompt(
if not extra_prompt:
return
- req.system_prompt = (
- f"{req.system_prompt or ''}\n"
- "[Workspace Extra Prompt]\n"
- "The following instructions are loaded from the current workspace "
- "`EXTRA_PROMPT.md` file.\n"
- f"{extra_prompt}\n"
+ req.system_prompt = (req.system_prompt or "") + _wrap_system_block(
+ "workspace_extra_prompt",
+ (
+ "The following instructions are loaded from the current workspace "
+ "`EXTRA_PROMPT.md` file.\n"
+ f"{extra_prompt}"
+ ),
)
@@ -378,7 +394,11 @@ def _apply_local_env_tools(req: ProviderRequest, plugin_context: Context) -> Non
req.func_tool.add_tool(tool_mgr.get_builtin_tool(FileWriteTool))
req.func_tool.add_tool(tool_mgr.get_builtin_tool(FileEditTool))
req.func_tool.add_tool(tool_mgr.get_builtin_tool(GrepTool))
- req.system_prompt = f"{req.system_prompt or ''}\n{_build_local_mode_prompt()}\n"
+ req.system_prompt = f"{req.system_prompt or ''}"
+ req.system_prompt += _wrap_system_block(
+ "runtime_environment",
+ _build_local_mode_prompt(),
+ )
def _build_local_mode_prompt() -> str:
@@ -460,11 +480,17 @@ async def _ensure_persona_and_skills(
if persona:
# Inject persona system prompt
if prompt := persona["prompt"]:
- req.system_prompt += f"\n# Persona Instructions\n\n{prompt}\n"
+ req.system_prompt += _wrap_system_block(
+ "persona_instructions",
+ prompt,
+ )
if begin_dialogs := copy.deepcopy(persona.get("_begin_dialogs_processed")):
req.contexts[:0] = begin_dialogs
elif use_webchat_special_default:
- req.system_prompt += CHATUI_SPECIAL_DEFAULT_PERSONA_PROMPT
+ req.system_prompt += _wrap_system_block(
+ "persona_instructions",
+ CHATUI_SPECIAL_DEFAULT_PERSONA_PROMPT,
+ )
# Inject skills prompt
runtime = cfg.get("computer_use_runtime", "local")
@@ -480,12 +506,18 @@ async def _ensure_persona_and_skills(
allowed = set(persona["skills"])
skills = [skill for skill in skills if skill.name in allowed]
if skills:
- req.system_prompt += f"\n{build_skills_prompt(skills)}\n"
+ req.system_prompt += _wrap_system_block(
+ "available_skills",
+ build_skills_prompt(skills),
+ )
if runtime == "none":
- req.system_prompt += (
- "User has not enabled the Computer Use feature. "
- "You cannot use shell or Python to perform skills. "
- "If you need to use these capabilities, ask the user to enable Computer Use in the AstrBot WebUI -> Config."
+ req.system_prompt += _wrap_system_block(
+ "computer_use_disabled_notice",
+ (
+ "User has not enabled the Computer Use feature. "
+ "You cannot use shell or Python to perform skills. "
+ "If you need to use these capabilities, ask the user to enable Computer Use in the AstrBot WebUI -> Config."
+ ),
)
tmgr = plugin_context.get_llm_tool_manager()
@@ -567,7 +599,10 @@ async def _ensure_persona_and_skills(
.get("router_system_prompt", "")
).strip()
if router_prompt:
- req.system_prompt += f"\n{router_prompt}\n"
+ req.system_prompt += _wrap_system_block(
+ "subagent_routing_instructions",
+ router_prompt,
+ )
try:
event.trace.record(
"sel_persona",
@@ -1004,7 +1039,13 @@ async def _handle_webchat(
def _apply_llm_safety_mode(config: MainAgentBuildConfig, req: ProviderRequest) -> None:
if config.safety_mode_strategy == "system_prompt":
- req.system_prompt = f"{LLM_SAFETY_MODE_SYSTEM_PROMPT}\n\n{req.system_prompt}"
+ req.system_prompt = (
+ _wrap_system_block(
+ "safety_instructions",
+ LLM_SAFETY_MODE_SYSTEM_PROMPT,
+ ).lstrip()
+ + f"\n{req.system_prompt or ''}"
+ )
else:
logger.warning(
"Unsupported llm_safety_mode strategy: %s.",
@@ -1043,23 +1084,27 @@ def _apply_sandbox_tools(
if booter == "shipyard_neo":
# Neo-specific path rule: filesystem tools operate relative to sandbox
# workspace root. Do not prepend "/workspace".
- req.system_prompt += (
- "\n[Shipyard Neo File Path Rule]\n"
- "When using sandbox filesystem tools (upload/download/read/write/list/delete), "
- "always pass paths relative to the sandbox workspace root. "
- "Example: use `baidu_homepage.png` instead of `/workspace/baidu_homepage.png`.\n"
+ req.system_prompt += _wrap_system_block(
+ "shipyard_neo_file_path_rule",
+ (
+ "When using sandbox filesystem tools (upload/download/read/write/list/delete), "
+ "always pass paths relative to the sandbox workspace root. "
+ "Example: use `baidu_homepage.png` instead of `/workspace/baidu_homepage.png`."
+ ),
)
- req.system_prompt += (
- "\n[Neo Skill Lifecycle Workflow]\n"
- "When user asks to create/update a reusable skill in Neo mode, use lifecycle tools instead of directly writing local skill folders.\n"
- "Preferred sequence:\n"
- "1) Use `astrbot_create_skill_payload` to store canonical payload content and get `payload_ref`.\n"
- "2) Use `astrbot_create_skill_candidate` with `skill_key` + `source_execution_ids` (and optional `payload_ref`) to create a candidate.\n"
- "3) Use `astrbot_promote_skill_candidate` to release: `stage=canary` for trial; `stage=stable` for production.\n"
- "For stable release, set `sync_to_local=true` to sync `payload.skill_markdown` into local `SKILL.md`.\n"
- "Do not treat ad-hoc generated files as reusable Neo skills unless they are captured via payload/candidate/release.\n"
- "To update an existing skill, create a new payload/candidate and promote a new release version; avoid patching old local folders directly.\n"
+ req.system_prompt += _wrap_system_block(
+ "neo_skill_lifecycle_workflow",
+ (
+ "When user asks to create/update a reusable skill in Neo mode, use lifecycle tools instead of directly writing local skill folders.\n"
+ "Preferred sequence:\n"
+ "1) Use `astrbot_create_skill_payload` to store canonical payload content and get `payload_ref`.\n"
+ "2) Use `astrbot_create_skill_candidate` with `skill_key` + `source_execution_ids` (and optional `payload_ref`) to create a candidate.\n"
+ "3) Use `astrbot_promote_skill_candidate` to release: `stage=canary` for trial; `stage=stable` for production.\n"
+ "For stable release, set `sync_to_local=true` to sync `payload.skill_markdown` into local `SKILL.md`.\n"
+ "Do not treat ad-hoc generated files as reusable Neo skills unless they are captured via payload/candidate/release.\n"
+ "To update an existing skill, create a new payload/candidate and promote a new release version; avoid patching old local folders directly."
+ ),
)
# Determine sandbox capabilities from an already-booted session.
@@ -1093,22 +1138,28 @@ def _apply_sandbox_tools(
req.func_tool.add_tool(tool_mgr.get_builtin_tool(SyncSkillReleaseTool))
if booter == "cua":
- req.system_prompt += (
- "\n[CUA Desktop Control]\n"
- "Use `astrbot_execute_shell` with `background=true` to launch GUI apps. "
- 'Use Firefox for browser tasks, for example `firefox "https://example.com"`. '
- "After each visible step, call `astrbot_cua_screenshot` with "
- "`send_to_user=true` and `return_image_to_llm=true` so the user can "
- "monitor progress. When typing, inspect the screenshot first and confirm "
- "the target field is focused and empty or safe to append to. Use "
- "`astrbot_cua_mouse_click` for coordinates and `astrbot_cua_keyboard_type` "
- "for text input; use text=`\\n` for Enter.\n"
+ req.system_prompt += _wrap_system_block(
+ "cua_desktop_control_instructions",
+ (
+ "Use `astrbot_execute_shell` with `background=true` to launch GUI apps. "
+ 'Use Firefox for browser tasks, for example `firefox "https://example.com"`. '
+ "After each visible step, call `astrbot_cua_screenshot` with "
+ "`send_to_user=true` and `return_image_to_llm=true` so the user can "
+ "monitor progress. When typing, inspect the screenshot first and confirm "
+ "the target field is focused and empty or safe to append to. Use "
+ "`astrbot_cua_mouse_click` for coordinates and `astrbot_cua_keyboard_type` "
+ "for text input; use text=`\\n` for Enter."
+ ),
)
req.func_tool.add_tool(tool_mgr.get_builtin_tool(CuaScreenshotTool))
req.func_tool.add_tool(tool_mgr.get_builtin_tool(CuaMouseClickTool))
req.func_tool.add_tool(tool_mgr.get_builtin_tool(CuaKeyboardTypeTool))
- req.system_prompt = f"{req.system_prompt or ''}\n{SANDBOX_MODE_PROMPT}\n"
+ req.system_prompt = f"{req.system_prompt or ''}"
+ req.system_prompt += _wrap_system_block(
+ "sandbox_mode_instructions",
+ SANDBOX_MODE_PROMPT,
+ )
def _proactive_cron_job_tools(req: ProviderRequest, plugin_context: Context) -> None:
@@ -1507,18 +1558,27 @@ async def build_main_agent(
)
if config.computer_use_runtime == "local":
- tool_prompt += (
- f"\nCurrent workspace you can use: "
- f"`{_get_workspace_path_for_umo(event.unified_msg_origin)}`\n"
- "Unless the user explicitly specifies a different directory, "
- "perform all file-related operations in this workspace.\n"
+ req.system_prompt = (req.system_prompt or "") + _wrap_system_block(
+ "workspace_runtime_rule",
+ (
+ f"Current workspace you can use: "
+ f"`{_get_workspace_path_for_umo(event.unified_msg_origin)}`\n"
+ "Unless the user explicitly specifies a different directory, "
+ "perform all file-related operations in this workspace."
+ ),
)
- req.system_prompt += f"\n{tool_prompt}\n"
+ req.system_prompt = (req.system_prompt or "") + _wrap_system_block(
+ "tool_use_instructions",
+ tool_prompt,
+ )
action_type = event.get_extra("action_type")
if action_type == "live":
- req.system_prompt += f"\n{LIVE_MODE_SYSTEM_PROMPT}\n"
+ req.system_prompt = (req.system_prompt or "") + _wrap_system_block(
+ "live_mode_instructions",
+ LIVE_MODE_SYSTEM_PROMPT,
+ )
reset_coro = agent_runner.reset(
provider=provider,
diff --git a/tests/unit/test_astr_main_agent.py b/tests/unit/test_astr_main_agent.py
index db729a23ba..8987814417 100644
--- a/tests/unit/test_astr_main_agent.py
+++ b/tests/unit/test_astr_main_agent.py
@@ -343,8 +343,9 @@ async def test_apply_kb_without_agentic_mode(self, mock_event, mock_context):
):
await module._apply_kb(mock_event, req, mock_context, config)
- assert "[Related Knowledge Base Results]:" in req.system_prompt
+ assert "" in req.system_prompt
assert "KB result" in req.system_prompt
+ assert "" in req.system_prompt
@pytest.mark.asyncio
async def test_apply_kb_with_agentic_mode(self, mock_event, mock_context):
@@ -1788,7 +1789,9 @@ def test_apply_llm_safety_mode_prepends_safety_prompt(self):
module._apply_llm_safety_mode(config, req)
- assert req.system_prompt.startswith("You are running in Safe Mode")
+ assert req.system_prompt.startswith("")
+ assert "You are running in Safe Mode" in req.system_prompt
+ assert "" in req.system_prompt
assert "My custom prompt" in req.system_prompt
def test_apply_llm_safety_mode_with_none_system_prompt(self):