defmodule CMDC.Tool.Task do @moduledoc """ 子代理工具 — 在独立进程中启动子 Agent 执行委托任务。 对标 DeepAgents 的 `task` 工具,利用 OTP 进程实现**真正的隔离与并发**: 子 Agent crash 不影响父 Agent,多个子代理可同时运行。 ## 调用示例 %{ "name" => "task", "arguments" => %{ "description" => "搜索项目中所有 TODO 注释并整理成清单", "subagent_type" => "general-purpose" } } ## 执行流程 1. 解析 `subagent_type` → 匹配 `Options.subagents` 中的 `SubAgent` 规格 2. `SubAgent.to_options/2` 合并父 Agent 配置(nil 字段继承父级) 3. 启动子 Agent 进程(`CMDC.Agent.start_link/1`) 4. 发送 HumanMessage(任务描述),子 Agent 独立运行 5. 通过 EventBus 订阅子 Agent 事件,收集流式输出与工具调用日志 6. 子 Agent 完成后,取最后一条 AI 回复文本作为 ToolMessage 返回父 Agent 7. 广播 `:subagent_start` / `:subagent_end` 事件 8. 终止子 Agent 进程 ## 深度限制 子代理的工具列表自动排除 `CMDC.Tool.Task`,防止递归生成。 ## 上下文隔离 - **继承**:working_dir、sandbox、model(若未覆盖)、工具集(若未覆盖)、user_data(若未覆盖) - **不继承**:messages、todos、memory_contents、tool_call_hashes - **输入**:单条 HumanMessage(任务描述) - **输出**:最后一条 AI 回复文本 → ToolMessage """ @behaviour CMDC.Tool require Logger @dialyzer [{:nowarn_function, build_child_opts: 3}, {:nowarn_function, run_sub_agent: 3}] @sub_agent_timeout 120_000 @general_purpose %CMDC.SubAgent{ name: "general-purpose", description: "通用子代理,适合需要独立上下文的复杂任务:" <> "代码搜索、多步骤实现、独立调研等。" } # ========================================================================== # Tool Behaviour 实现 # ========================================================================== @impl true def name, do: "task" @impl true def description do "Launch a sub-agent to work on a delegated task independently. " <> "The sub-agent has its own conversation loop and tools, " <> "and returns its final response when done. " <> "Use this for complex tasks that benefit from a separate context: " <> "code search, multi-step implementation, independent research, etc." end @impl true def description(%{} = context) do subagents = Map.get(context, :subagents, []) base = description() if subagents == [] do base else names = Enum.map_join(subagents, ", ", &extract_name/1) base <> " Available sub-agent types: #{names}." end end @impl true def meta(%{"description" => desc}), do: "Task: #{String.slice(desc, 0, 54)}" def meta(_), do: "Task" @impl true def parameters do %{ "type" => "object", "properties" => %{ "description" => %{ "type" => "string", "description" => "A detailed description of the task for the sub-agent to accomplish. " <> "Be specific about goals, constraints, and expected output format." }, "subagent_type" => %{ "type" => "string", "description" => "The type of sub-agent to use. Must match one of the declared " <> "sub-agent names. Defaults to \"general-purpose\" if not specified." } }, "required" => ["description"] } end @impl true def execute(%{"description" => task_description} = args, context) do subagent_type = Map.get(args, "subagent_type", "general-purpose") with {:ok, sub_spec} <- resolve_subagent(subagent_type, context), {:ok, parent_opts} <- build_parent_opts(context), {:ok, child_opts} <- build_child_opts(sub_spec, parent_opts, context) do run_sub_agent(task_description, child_opts, context) end end def execute(_args, _context) do {:error, "Missing required parameter: description"} end # ========================================================================== # 子代理解析 # ========================================================================== defp resolve_subagent(name, context) do subagents = extract_subagents(context) case Enum.find(subagents, &(extract_name(&1) == name)) do nil when name == "general-purpose" -> {:ok, @general_purpose} nil -> available = Enum.map_join(subagents, ", ", &extract_name/1) available = if available == "", do: "general-purpose", else: available <> ", general-purpose" {:error, "Unknown subagent_type: \"#{name}\". Available: #{available}"} spec -> {:ok, normalize_spec(spec)} end end defp extract_subagents(%CMDC.Context{subagents: subs}), do: subs defp extract_subagents(%{subagents: subs}) when is_list(subs), do: subs defp extract_subagents(_), do: [] defp extract_name(%CMDC.SubAgent{name: name}), do: name defp extract_name(%{name: name}), do: name defp extract_name(spec) when is_map(spec), do: Map.get(spec, "name") || Map.get(spec, :name, "") defp normalize_spec(%CMDC.SubAgent{} = spec), do: spec defp normalize_spec(spec) when is_map(spec) do %CMDC.SubAgent{ name: extract_name(spec), description: Map.get(spec, :description) || Map.get(spec, "description"), system_prompt: Map.get(spec, :system_prompt) || Map.get(spec, "system_prompt"), model: Map.get(spec, :model) || Map.get(spec, "model"), tools: Map.get(spec, :tools) || Map.get(spec, "tools"), plugins: Map.get(spec, :plugins) || Map.get(spec, "plugins"), skills_dirs: Map.get(spec, :skills_dirs) || Map.get(spec, "skills_dirs"), prompt_mode: extract_prompt_mode(spec) } end defp extract_prompt_mode(spec) do case Map.get(spec, :prompt_mode) || Map.get(spec, "prompt_mode") do nil -> :task mode when mode in [:full, :task, :minimal, :none] -> mode "full" -> :full "task" -> :task "minimal" -> :minimal "none" -> :none _ -> :task end end # ========================================================================== # Options 构建 # ========================================================================== defp build_parent_opts(context) do {:ok, %CMDC.Options{ model: context.model || "", tools: Map.get(context, :tools, []), system_prompt: nil, plugins: [], subagents: extract_subagents(context), skills_dirs: [], working_dir: context.working_dir || ".", user_data: extract_user_data(context), prompt_mode: extract_parent_prompt_mode(context) }} end defp build_child_opts(%CMDC.SubAgent{} = spec, %CMDC.Options{} = parent, context) do merged = CMDC.SubAgent.to_options(spec, parent) tools = (merged.tools || []) |> Enum.reject(&(&1 == __MODULE__)) config = extract_config(context) {:ok, %{ model: merged.model, tools: tools, working_dir: merged.working_dir, system_prompt: spec.system_prompt, plugins: merged.plugins, sandbox: Map.get(context, :sandbox), config: config, user_data: merged.user_data, prompt_mode: merged.prompt_mode }} end defp extract_parent_prompt_mode(%CMDC.Context{prompt_mode: mode}) when not is_nil(mode), do: mode defp extract_parent_prompt_mode(%{prompt_mode: mode}) when is_atom(mode) and not is_nil(mode), do: mode defp extract_parent_prompt_mode(_), do: :full defp extract_config(%CMDC.Context{config: config}) when not is_nil(config), do: config defp extract_config(%{config: config}) when is_map(config), do: config defp extract_config(_), do: %{} defp extract_user_data(%CMDC.Context{user_data: ud}) when is_map(ud), do: ud defp extract_user_data(%{user_data: ud}) when is_map(ud), do: ud defp extract_user_data(_), do: %{} # ========================================================================== # 子代理执行 # ========================================================================== defp run_sub_agent(task_description, child_opts, context) do session_id = generate_session_id() parent_session_id = context.session_id call_id = Map.get(context, :call_id, "") config = child_opts.config || %{} provider_opts = Map.get(config, :provider_opts, []) agent_opts = [ session_id: session_id, model: child_opts.model, working_dir: child_opts.working_dir, blueprint_system_prompt: child_opts.system_prompt || "", tools: child_opts.tools, plugins: child_opts.plugins || [], sandbox: child_opts.sandbox, config: config, provider_opts: provider_opts, user_data: Map.get(child_opts, :user_data, %{}) ] broadcast_parent( parent_session_id, {:subagent_start, parent_session_id, session_id, task_description} ) case CMDC.Agent.start_link(agent_opts) do {:ok, sub_pid} -> CMDC.EventBus.subscribe(session_id) CMDC.Agent.prompt(sub_pid, task_description) result = collect_and_forward( session_id, parent_session_id, call_id, "", [], @sub_agent_timeout ) CMDC.EventBus.unsubscribe(session_id) stop_sub_agent(sub_pid) case result do {:ok, response, tool_log} -> formatted = format_result(response, tool_log) broadcast_parent( parent_session_id, {:subagent_end, parent_session_id, session_id, {:ok, formatted}} ) {:ok, formatted} {:error, reason} -> broadcast_parent( parent_session_id, {:subagent_end, parent_session_id, session_id, {:error, reason}} ) {:error, "Sub-agent failed: #{inspect(reason)}"} end {:error, reason} -> broadcast_parent( parent_session_id, {:subagent_end, parent_session_id, session_id, {:error, reason}} ) {:error, "Failed to spawn sub-agent: #{inspect(reason)}"} end end # ========================================================================== # 事件收集与转发 # ========================================================================== defp collect_and_forward(sub_id, parent_id, call_id, text, tool_log, timeout) do receive do {:cmdc_event, ^sub_id, {:stream_chunk, ^sub_id, chunk}} -> forward(parent_id, sub_id, call_id, {:stream_chunk, sub_id, chunk}) collect_and_forward(sub_id, parent_id, call_id, text <> chunk, tool_log, timeout) {:cmdc_event, ^sub_id, {:tool_execution_start, tool_name, tc_id, tc_args}} -> forward(parent_id, sub_id, call_id, {:tool_execution_start, tool_name, tc_id, tc_args}) entry = %{tool: tool_name, call_id: tc_id, arguments: tc_args, result: nil} collect_and_forward(sub_id, parent_id, call_id, text, [entry | tool_log], timeout) {:cmdc_event, ^sub_id, {:tool_execution_end, tool_name, tc_id, result}} -> forward(parent_id, sub_id, call_id, {:tool_execution_end, tool_name, tc_id, result}) updated = update_tool_result(tool_log, tc_id, tool_name, result) collect_and_forward(sub_id, parent_id, call_id, text, updated, timeout) {:cmdc_event, ^sub_id, {:response_complete, msg}} -> new_text = extract_assistant_text(msg) collect_and_forward(sub_id, parent_id, call_id, new_text, tool_log, timeout) {:cmdc_event, ^sub_id, {:agent_end, _msgs, _token_usage}} -> {:ok, text, Enum.reverse(tool_log)} {:cmdc_event, ^sub_id, {:agent_abort, reason}} -> {:error, {:aborted, reason}} {:cmdc_event, ^sub_id, :agent_abort} -> {:error, :aborted} {:cmdc_event, ^sub_id, {:error, _sid, reason}} -> {:error, reason} {:cmdc_event, ^sub_id, _event} -> collect_and_forward(sub_id, parent_id, call_id, text, tool_log, timeout) after timeout -> {:error, :timeout} end end # ========================================================================== # 结果格式化 # ========================================================================== defp format_result("", []), do: "(Sub-agent completed with no output)" defp format_result(response, []), do: response defp format_result(response, tool_log) do tool_section = Enum.map_join(tool_log, "\n", fn entry -> result_str = case entry.result do {:ok, output} -> String.slice(to_string(output), 0, 500) {:error, reason} -> "ERROR: #{reason}" nil -> "(no result)" other -> String.slice(inspect(other), 0, 500) end "- #{entry.tool}: #{result_str}" end) """ ## Sub-agent response #{response} ## Tool calls made #{tool_section}\ """ end # ========================================================================== # 私有辅助 # ========================================================================== defp generate_session_id do "sub_" <> (:crypto.strong_rand_bytes(8) |> Base.encode16(case: :lower)) end defp stop_sub_agent(pid) do :gen_statem.stop(pid) rescue _ -> :ok end defp forward(parent_id, sub_id, call_id, event) do CMDC.EventBus.broadcast(parent_id, {:sub_agent_event, call_id, sub_id, event}) end defp broadcast_parent(parent_session_id, event) do CMDC.EventBus.broadcast(parent_session_id, event) end defp update_tool_result(tool_log, call_id, tool_name, result) do idx = Enum.find_index(tool_log, fn entry -> (entry.call_id == call_id or entry.tool == tool_name) and is_nil(entry.result) end) if idx, do: List.update_at(tool_log, idx, &%{&1 | result: result}), else: tool_log end defp extract_assistant_text(%CMDC.Message{role: :assistant, content: content}) when is_binary(content), do: content defp extract_assistant_text(%{content: content}) when is_binary(content), do: content defp extract_assistant_text(_), do: "" end