feat: 添加有向无环图(DAG)编排器，自动处理数据依赖

Author: elecvoid243

astrbot/core/subagent_dag.py

@@ -244,6 +244,7 @@ async def _execute_single_node(
                     else:
                         node.status = DAGNodeStatus.FAILED
                         node.error = result.error or "Unknown error"
+                        node.result = result.result or ""
                     return
 
                 session = SubAgentManager.get_session(session_id)
@@ -406,9 +407,9 @@ def _format_dag_result(
                     f"  ✓ {nid} ({node.agent_name}) — {node.execution_time:.1f}s"
                 )
                 if node.result:
-                    preview = node.result[:300]
-                    if len(node.result) > 300:
-                        preview += "..."
+                    preview = node.result[:1000]
+                    if len(node.result) > 1000:
+                        preview += "...[truncated]"
                     lines.append(f"     {preview}")
             elif node.status == DAGNodeStatus.FAILED:
                 lines.append(
@@ -417,6 +418,11 @@ def _format_dag_result(
                 )
                 if node.error:
                     lines.append(f"     Error: {node.error}")
+                if node.result:
+                    preview = node.result[:1000]
+                    if len(node.result) > 1000:
+                        preview += "...[truncated]"
+                    lines.append(f"     Output: {preview}")
             elif node.status == DAGNodeStatus.SKIPPED:
                 lines.append(
                     f"  ⊘ {nid} ({node.agent_name}) — skipped (dependency failed)"

astrbot/core/subagent_manager.py

@@ -259,7 +259,7 @@ def cleanup_session_turn_end(cls, session_id: str) -> dict:
             if not session.subagents and not session.protected_agents:
                 # 所有subagent都被清理，清除公共上下文
                 cls.clear_shared_context(session_id)
-                logger.debug(
+                logger.info(
                     "[SubAgent:SharedContext] All subagents cleaned, cleared shared context"
                 )
             else:
@@ -279,7 +279,6 @@ def cleanup_session_turn_end(cls, session_id: str) -> dict:
 
         # 每轮结束时顺便清理全局过期会话
         cls.cleanup_expired_sessions()
-
         return {"status": "cleaned", "cleaned_agents": cleaned}
 
     @classmethod
@@ -734,27 +733,21 @@ def _build_time_prompt(cls) -> str:
 
     _TASK_STATUS_PROMPT = (
         "# Task Status Reporting\n"
-        "At the end of your task, you MUST self-audit before reporting status.\n"
-        "\n"
-        "SUCCESS protocol — verify ALL:\n"
-        "- No tool returned an error or empty result unexpectedly\n"
-        "- The output directly answers the task prompt\n"
-        "- You are confident the result is not a guess or placeholder\n"
-        "- If files were created: they exist on disk and their content is "
-        "correct and complete enough for another agent to continue the work\n"
-        "<task_status><result>SUCCESS</result></task_status>\n"
-        "\n"
-        "FAILURE protocol — if ANY tool failed or you cannot complete:\n"
-        "<task_status><result>FAILURE</result>"
-        "<reason>specific reason</reason></task_status>\n"
-        "\n"
-        "Common mistakes to avoid:\n"
-        "- Do NOT report SUCCESS when a command returned an error\n"
-        "- Do NOT report SUCCESS when you only completed part of the task\n"
-        "- Do NOT report SUCCESS when generated files are missing or incomplete\n"
-        "- Do NOT guess status — if uncertain, use FAILURE\n"
-        "\n"
-        "Put the XML block first, then your result or error details.\n"
+        "At the end of your task, self-audit before giving your final answer.\n"
+        "## SUCCESS — use only when ALL of these are true:\n"
+        "- Every tool call succeeded; no unexpected error or empty result\n"
+        "- Your output directly answers the task you were assigned\n"
+        "- You are confident the result is accurate, not a guess or placeholder\n"
+        "- If you created files: ensure they exist on disk, and their content is correct and complete\n"
+        "If all pass, put this EXACT line FIRST, then your result:\n"
+        "[TASK RESULT: SUCCESS]\n"
+        "## FAILURE — use if ANY tool failed, or you cannot complete the task:\n"
+        "[TASK RESULT: FAILURE]\n"
+        "[FAILURE REASON: <one-line explanation>]\n"
+        "## Reporting Marker Rules\n"
+        "- The marker MUST be exactly `[TASK RESULT: SUCCESS]` or `[TASK RESULT: FAILURE]` — do not change it\n"
+        "- The marker MUST be on its own line, at the very top of your response\n"
+        "- When uncertain between success and failure, choose failure\n"
     )
 
     @classmethod
@@ -771,7 +764,10 @@ def _build_rule_prompt(cls) -> str:
                 "- Mark all generated code/documents with your name and timestamp (if given).\n"
             )
         )
-        return base + cls._TASK_STATUS_PROMPT
+        if cls._dag_enabled:
+            return base + cls._TASK_STATUS_PROMPT
+        else:
+            return base
 
     @classmethod
     def cleanup_shared_context_by_agent(cls, session_id: str, agent_name: str) -> None:

astrbot/core/subagent_tools.py

@@ -740,15 +740,28 @@ async def _launch_dag_node(node, _sid, injected_context, task_id):
                                 result_text += c.text + chr(10)
 
                 # Detect task status from subagent output.
-                # Priority: 1) XML task_status  2) error: prefix  3) empty
+                # Priority: 1) [TASK RESULT: ...] marker  2) error: prefix  3) empty
                 success = True
+                error_reason = None
                 stripped = result_text.strip()
                 status_match = re.search(
-                    r"<task_status>\s*<result>\s*(SUCCESS|FAILURE)\s*</result>",
+                    r"\[TASK\s*RESULT\s*:\s*(SUCCESS|FAILURE)\]",
                     stripped,
+                    re.IGNORECASE,
                 )
                 if status_match:
-                    success = status_match.group(1) == "SUCCESS"
+                    success = status_match.group(1).upper() == "SUCCESS"
+                    if not success:
+                        # Extract failure reason for concise error reporting
+                        reason_match = re.search(
+                            r"\[FAILURE\s*REASON\s*:\s*(.+?)\]",
+                            stripped,
+                            re.IGNORECASE,
+                        )
+                        if reason_match:
+                            error_reason = reason_match.group(1).strip()
+                        else:
+                            error_reason = "No reason provided"
                 elif not stripped or stripped.lower().startswith("error:"):
                     success = False
 
@@ -759,6 +772,7 @@ async def _launch_dag_node(node, _sid, injected_context, task_id):
                     result_text,
                     task_id=task_id,
                     execution_time=0.0,
+                    error=error_reason,
                 )
             except Exception as e:
                 logger.error(f"[SubAgent:DAG] Launch error for {node.agent_name}: {e}")
@@ -776,6 +790,7 @@ async def _launch_dag_node(node, _sid, injected_context, task_id):
             result = await SubAgentDAGEngine.execute_dag(
                 ctx=dag_ctx,
                 session_id=session_id,
+                max_inject_length=cfg.get("dag_max_inject_length", 4000),
                 launch_fn=_launch_dag_node,
             )
             dag_ctx.status = "COMPLETED" if result["failed"] == 0 else "FAILED"