feat: 动态创建子Agent支持在运行时产生的事件直接转发到父流程的事件流中的能力

Author: bochencwx

docs/mkdocs/en/sub_agent.md

@@ -21,6 +21,19 @@ A **short-lived sub-agent** is a natural fit for these problems: a fresh context
 
 The difference is *who defines the role*: the developer (Spawn) or the LLM (Dynamic).
 
+### How this differs from other multi-agent mechanisms
+
+The framework already offers several ways to compose agents (see [Multi Agents](multi_agents.md)). Spawned Sub-Agents solve a different problem:
+
+| Mechanism | Agents involved | Who decides when to invoke | Context | Typical use |
+| --- | --- | --- | --- | --- |
+| **Chain / Parallel / Cycle Agent** | **Pre-built** fixed agent instances | **Deterministic** orchestration — run in list order / in parallel / in a loop, regardless of input | Each agent independent | Fixed multi-step workflows |
+| **Sub Agents (transfer)** | Pre-registered agents | Parent **transfers control** at runtime; the sub-agent then takes over the conversation | Shared session | **Hand off** the whole conversation to a better-suited agent |
+| **AgentTool** | Wraps **an existing agent instance** as a tool | Parent LLM calls it on demand | Shares/syncs state & artifacts back to parent | Reuse a **specific, already-built** agent |
+| **Spawned Sub-Agents** | **Created on the fly** per call, destroyed after | Parent LLM calls it on demand | **Strictly isolated**: fresh ephemeral session, history/state not shared by default | Delegate a **one-off** subtask while keeping the parent context clean |
+
+In one line: **Chain/Parallel/Cycle** deterministically orchestrate a fixed set of agents; **transfer** hands the conversation off; **AgentTool** reuses one existing agent as a tool; while **Spawned Sub-Agents** create an **isolated, short-lived** sub-agent on the spot for a single task and discard it afterward — the emphasis is on **on-demand runtime creation** and **context isolation**, not reusing an existing agent or transferring control.
+
 ## Quick Start
 
 ```python
@@ -173,8 +186,19 @@ class SubAgentConfig:
 
     max_turns: int | None = None
     """Max LLM calls the sub-agent may make. None = unlimited."""
+
+    forward_events: bool = False
+    """Whether to forward the sub-agent's execution events to the parent
+    runner's consumer as progress updates.
+
+    True: the orchestrator can display the sub-agent's execution live (model
+    output, tool calls, tool results); the parent agent's LLM still receives
+    only the sub-agent's final result. False (default): the sub-agent runs
+    silently and only its final result is returned."""
 ```
 
+Forwarded events reach the consumer as progress events; they are **not** written to the parent session and **never** enter the parent agent's LLM context. Consumers identify them via `tool_progress=True` on `event.custom_metadata` and read the execution from `payload` (`author` / `partial` / `content`, plus optional `error` / `usage`).
+
 ## Usage
 
 ### SpawnSubAgentTool
@@ -270,3 +294,4 @@ orchestrator = LlmAgent(
 - **Session isolation**: sub-agents run in a fresh ephemeral session. Parent history is not shared by default; opt in via `include_parent_history=True`.
 - **Nesting**: 1-level hard cap. Sub-agents cannot spawn further sub-agents.
 - **Result shape**: the sub-agent's final text is returned as the tool result string.
+- **Live execution (`forward_events`)**: set `SubAgentConfig(forward_events=True)` to stream the sub-agent's execution to the parent runner's consumer for display. Forwarded events are progress events — they never enter the parent LLM's context, which still receives only the final result. Consumers detect them via `tool_progress=True` on `event.custom_metadata` and read `payload`. See `examples/dynamic_subagent` and `examples/spawn_subagent` for a working consumer.

docs/mkdocs/zh/sub_agent.md

@@ -21,6 +21,19 @@
 
 区别在于**谁定义角色**：开发者（Spawn）还是 LLM（Dynamic）。
 
+### 与框架其他多 agent 机制的区别
+
+框架已有几种组合 agent 的方式（详见 [Multi Agents](multi_agents.md)）。Spawned Sub-Agents 与它们解决的是不同问题：
+
+| 机制 | 参与的 agent | 谁决定何时调用 | 上下文 | 典型用途 |
+| --- | --- | --- | --- | --- |
+| **Chain / Parallel / Cycle Agent** | **预先构建**的固定 agent 实例 | **确定性**编排——按列表顺序/并行/循环执行，与输入无关 | 各 agent 独立 | 固定的多步工作流 |
+| **Sub Agents（transfer）** | 预先注册的 agent | 父 agent 运行时**转移控制权**，之后由子 agent 接管对话 | 共享同一会话 | 把整段对话**移交**给更合适的 agent |
+| **AgentTool** | 把**某个已有 agent 实例**包成工具 | 父 LLM 按需调用 | 会共享/同步 state 与 artifact 回父 | 复用一个**具体的、已存在的** agent |
+| **Spawned Sub-Agents** | **调用时临时创建**、用完即销毁 | 父 LLM 按需调用 | **严格隔离**：全新临时会话，默认不共享历史/state | 委派**一次性**子任务，保持父上下文干净 |
+
+一句话概括：**Chain/Parallel/Cycle** 是"确定性编排一组固定 agent"；**transfer** 是"把对话交出去"；**AgentTool** 是"把一个已有 agent 当工具复用"；而 **Spawned Sub-Agents** 是"为单次任务**现场造一个隔离的、短命的**子 agent，跑完就丢"——强调的是**运行时按需创建**与**上下文隔离**，而非复用既有 agent 或转移控制。
+
 ## Quick Start
 
 ```python
@@ -173,8 +186,17 @@ class SubAgentConfig:
 
     max_turns: int | None = None
     """子 agent 最多可发起的 LLM 调用次数。None = 不限制。"""
+
+    forward_events: bool = False
+    """是否将子 agent 的执行事件转发给父 runner 的消费者，作为进度更新。
+
+    True：编排层可实时展示子 agent 的执行（模型输出、工具调用、工具结果），
+    父 agent 的 LLM 仍只收到子 agent 的最终结果。
+    False（默认）：子 agent 静默执行，只回传最终结果。"""
 ```
 
+转发的事件以进度事件形式到达消费者，**不会**写入父会话、也**不会**进入父 agent 的 LLM 上下文；消费者通过 `event.custom_metadata` 上的 `tool_progress=True` 识别它们，并从 `payload` 读取执行内容（`author` / `partial` / `content` / 可选 `error` / `usage`）。
+
 ## 使用方式
 
 ### SpawnSubAgentTool
@@ -270,3 +292,4 @@ orchestrator = LlmAgent(
 - **会话隔离**：子 agent 在全新临时会话中运行，默认不共享父会话历史。通过 `include_parent_history=True` 可注入。
 - **嵌套限制**：1 层硬限，子 agent 无法再次 spawn。
 - **结果形态**：子 agent 的最终文本作为 tool result 字符串返回。
+- **实时执行（`forward_events`）**：设置 `SubAgentConfig(forward_events=True)` 可将子 agent 的执行流转发给父 runner 的消费者用于展示。转发事件为进度事件——它们不会进入父 agent 的 LLM 上下文（父仍只收到最终结果）。消费者通过 `event.custom_metadata` 上的 `tool_progress=True` 识别它们并读取 `payload`。可运行示例见 `examples/dynamic_subagent` 与 `examples/spawn_subagent`。

examples/dynamic_subagent/agent/agent.py

@@ -45,7 +45,12 @@ def create_minimal_agent() -> LlmAgent:
             temperature=0.7,
             max_output_tokens=2000,
         ),
-        tools=workspace_tools + [DynamicSubAgentTool()],
+        tools=workspace_tools + [
+            DynamicSubAgentTool(
+                # Stream the sub-agent's execution to the parent consumer.
+                agent_config=SubAgentConfig(forward_events=True),
+            ),
+        ],
     )
 
 
@@ -69,6 +74,8 @@ def create_bounded_agent() -> LlmAgent:
                         temperature=0.3,
                         max_output_tokens=1000,
                     ),
+                    # Stream the sub-agent's execution to the parent consumer.
+                    forward_events=True,
                 ),
             ),
         ],

examples/dynamic_subagent/run_agent.py

@@ -33,14 +33,45 @@
 
 def _truncate(text: str, max_len: int = 200) -> str:
     """Truncate long tool output for display."""
-    return text
     if not isinstance(text, str):
         text = str(text)
     if len(text) <= max_len:
         return text
     return text[:max_len] + f"\n... (truncated, total {len(text)} chars)"
 
 
+def _print_subagent_progress(payload: dict) -> None:
+    """Render one forwarded sub-agent execution event.
+
+    ``payload`` is a :class:`SubAgentProgress` dict: ``author`` / ``partial``,
+    the framework-native ``content`` dump (``parts`` with ``function_call`` /
+    ``function_response`` / ``text`` / ``thought``), and optional ``error`` /
+    ``usage``. Indented under the parent output so the sub-agent's steps are
+    visually distinct from the orchestrator's.
+    """
+    name = payload.get("author") or "subagent"
+    # Errors first — always surface them, even on an otherwise-partial event.
+    err = payload.get("error")
+    if err:
+        print(f"\n   \U0001F9E9 [{name}] !! error {err.get('code')}: {err.get('message')}")
+    if payload.get("partial"):
+        # Skip streaming text deltas to keep the demo output readable; the
+        # non-partial steps below already summarize the sub-agent's work.
+        return
+    parts = (payload.get("content") or {}).get("parts") or []
+    has_calls = any(p.get("function_call") or p.get("function_response") for p in parts)
+    for p in parts:
+        fc = p.get("function_call")
+        if fc:
+            print(f"\n   \U0001F9E9 [{name}] -> tool {fc.get('name')}({_truncate(fc.get('args'))})")
+        fr = p.get("function_response")
+        if fr:
+            print(f"   \U0001F9E9 [{name}] <- {_truncate(fr.get('response'))}")
+        text = p.get("text")
+        if text and not p.get("thought") and not has_calls:
+            print(f"\n   \U0001F9E9 [{name}] {_truncate(text)}")
+
+
 _QUERIES = {
     "minimal": [
         # Simple task: orchestrator may call word_count directly.
@@ -93,6 +124,18 @@ async def run_demo(mode: str):
             session_id=current_session_id,
             new_message=user_content,
         ):
+            # Forwarded sub-agent execution events (SubAgentConfig
+            # forward_events=True). These are partial progress events carrying
+            # the sub-agent's own steps under custom_metadata.payload; they
+            # never reach the parent LLM's context. This demo registers only the
+            # sub-agent tool, so tool_progress alone identifies them; an app with
+            # several progress tools would also branch on meta["tool_name"].
+            meta = event.custom_metadata or {}
+            payload = meta.get("payload")
+            if meta.get("tool_progress") and isinstance(payload, dict):
+                _print_subagent_progress(payload)
+                continue
+
             if event.content and event.content.parts and event.author != "user":
                 if event.partial:
                     for part in event.content.parts:

examples/spawn_subagent/agent/agent.py

@@ -25,6 +25,7 @@
 from trpc_agent_sdk.agents.sub_agent import EXPLORE_AGENT
 from trpc_agent_sdk.agents.sub_agent import PLAN_AGENT
 from trpc_agent_sdk.agents.sub_agent import SubAgentArchetype
+from trpc_agent_sdk.agents.sub_agent import SubAgentConfig
 from trpc_agent_sdk.tools import SpawnSubAgentTool
 from trpc_agent_sdk.models import LLMModel
 from trpc_agent_sdk.models import OpenAIModel
@@ -52,7 +53,11 @@ def create_default_agent() -> LlmAgent:
         description="Coding assistant with spawn_subagent in zero-config mode.",
         model=_create_model(),
         instruction=INSTRUCTION,
-        tools=[ReadTool(), GlobTool(), GrepTool(), SpawnSubAgentTool()],
+        tools=[ReadTool(), GlobTool(), GrepTool(),
+               SpawnSubAgentTool(
+                   # Stream the sub-agent's execution to the parent consumer.
+                   agent_config=SubAgentConfig(forward_events=True),
+               )],
     )
 
 
@@ -88,7 +93,11 @@ def create_code_agent() -> LlmAgent:
         instruction=INSTRUCTION,
         tools=[
             ReadTool(), GlobTool(), GrepTool(),
-            SpawnSubAgentTool(agents=[_SECURITY_AUDITOR, EXPLORE_AGENT, PLAN_AGENT]),
+            SpawnSubAgentTool(
+                agents=[_SECURITY_AUDITOR, EXPLORE_AGENT, PLAN_AGENT],
+                # Stream the sub-agent's execution to the parent consumer.
+                agent_config=SubAgentConfig(forward_events=True),
+            ),
         ],
     )
 
@@ -109,7 +118,11 @@ def create_md_agent() -> LlmAgent:
         instruction=INSTRUCTION,
         tools=[
             ReadTool(), GlobTool(), GrepTool(),
-            SpawnSubAgentTool(agents=[EXPLORE_AGENT, PLAN_AGENT], agent_paths=[_AGENTS_PATH]),
+            SpawnSubAgentTool(
+                agents=[EXPLORE_AGENT, PLAN_AGENT], agent_paths=[_AGENTS_PATH],
+                # Stream the sub-agent's execution to the parent consumer.
+                agent_config=SubAgentConfig(forward_events=True),
+            ),
         ],
     )
 

examples/spawn_subagent/run_agent.py

@@ -45,6 +45,38 @@ def _truncate(text: str, max_len: int = 200) -> str:
     return text[:max_len] + f"\n... (truncated, total {len(text)} chars)"
 
 
+def _print_subagent_progress(payload: dict) -> None:
+    """Render one forwarded sub-agent execution event.
+
+    ``payload`` is a :class:`SubAgentProgress` dict: ``author`` / ``partial``,
+    the framework-native ``content`` dump (``parts`` with ``function_call`` /
+    ``function_response`` / ``text`` / ``thought``), and optional ``error`` /
+    ``usage``. Indented under the parent output so the sub-agent's steps are
+    visually distinct from the assistant's.
+    """
+    name = payload.get("author") or "subagent"
+    # Errors first — always surface them, even on an otherwise-partial event.
+    err = payload.get("error")
+    if err:
+        print(f"\n   \U0001F9E9 [{name}] !! error {err.get('code')}: {err.get('message')}")
+    if payload.get("partial"):
+        # Skip streaming text deltas to keep the demo output readable; the
+        # non-partial steps below already summarize the sub-agent's work.
+        return
+    parts = (payload.get("content") or {}).get("parts") or []
+    has_calls = any(p.get("function_call") or p.get("function_response") for p in parts)
+    for p in parts:
+        fc = p.get("function_call")
+        if fc:
+            print(f"\n   \U0001F9E9 [{name}] -> tool {fc.get('name')}({_truncate(fc.get('args'))})")
+        fr = p.get("function_response")
+        if fr:
+            print(f"   \U0001F9E9 [{name}] <- {_truncate(fr.get('response'))}")
+        text = p.get("text")
+        if text and not p.get("thought") and not has_calls:
+            print(f"\n   \U0001F9E9 [{name}] {_truncate(text)}")
+
+
 # Queries per mode — simple tasks (parent handles directly) vs complex
 # tasks (delegated to a sub-agent).
 _SHARED_AGENT_QUERIES = [
@@ -111,6 +143,18 @@ async def run_demo(mode: str):
             session_id=current_session_id,
             new_message=user_content,
         ):
+            # Forwarded sub-agent execution events (SubAgentConfig
+            # forward_events=True). These are partial progress events carrying
+            # the sub-agent's own steps under custom_metadata.payload; they
+            # never reach the parent LLM's context. This demo registers only the
+            # sub-agent tool, so tool_progress alone identifies them; an app with
+            # several progress tools would also branch on meta["tool_name"].
+            meta = event.custom_metadata or {}
+            payload = meta.get("payload")
+            if meta.get("tool_progress") and isinstance(payload, dict):
+                _print_subagent_progress(payload)
+                continue
+
             if event.content and event.content.parts and event.author != "user":
                 if event.partial:
                     for part in event.content.parts:

tests/agents/sub_agent/test_dynamic_sub_agent_tool.py

@@ -39,6 +39,57 @@ def test_constructor_with_config() -> None:
 def test_constructor_skip_summarization() -> None:
     t = DynamicSubAgentTool(skip_summarization=True)
     assert t._skip_summarization is True
+    # Exposed as a property for the progress-streaming execution path.
+    assert t.skip_summarization is True
+
+
+def test_is_progress_streaming_default_off() -> None:
+    """Without forward_events, the tool runs on the non-streaming path."""
+    assert DynamicSubAgentTool().is_progress_streaming is False
+    assert DynamicSubAgentTool(agent_config=SubAgentConfig()).is_progress_streaming is False
+
+
+def test_is_progress_streaming_on_when_forwarding() -> None:
+    t = DynamicSubAgentTool(agent_config=SubAgentConfig(forward_events=True))
+    assert t.is_progress_streaming is True
+
+
+@pytest.mark.asyncio
+async def test_run_streaming_empty_prompt_yields_error() -> None:
+    """run_streaming surfaces the prompt validation error as its only value."""
+    t = DynamicSubAgentTool(agent_config=SubAgentConfig(forward_events=True))
+    ctx = MagicMock()
+    yielded = [v async for v in t.run_streaming(tool_context=ctx, args={"prompt": "  "})]
+    assert len(yielded) == 1
+    assert yielded[0]["status"] == "error"
+    assert "prompt" in yielded[0]["message"]
+
+
+@pytest.mark.asyncio
+async def test_run_streaming_forwards_projections_then_result() -> None:
+    """run_streaming delegates to run_subagent_streaming, yielding its values in order."""
+    from unittest.mock import patch
+
+    t = DynamicSubAgentTool(agent_config=SubAgentConfig(forward_events=True))
+    ctx = MagicMock()
+
+    async def _fake_stream(**kwargs):
+        yield {"author": "subagent_dynamic", "partial": True, "content": {"parts": [{"text": "step 1"}]}}
+        yield "final result"
+
+    with patch(
+        "trpc_agent_sdk.agents.sub_agent._dynamic_sub_agent_tool.run_subagent_streaming",
+        _fake_stream,
+    ):
+        yielded = [v async for v in t.run_streaming(
+            tool_context=ctx,
+            args={"instruction": "You are a helper.", "prompt": "do it"},
+        )]
+
+    assert yielded == [
+        {"author": "subagent_dynamic", "partial": True, "content": {"parts": [{"text": "step 1"}]}},
+        "final result",
+    ]
 
 
 def test_constructor_custom_name() -> None: