defmodule CMDC do @moduledoc """ CMDC — Elixir Agent Kernel 公共 API 入口。 提供从零创建 Agent 会话到收集回复的完整链路。 所有操作面向 session pid(`CMDC.SessionServer` 进程), 内部通过 `CMDC.EventBus` 实现异步解耦。 ## 快速开始 # 1. 创建 Agent 会话 {:ok, session} = CMDC.create_agent( model: "anthropic:claude-sonnet-4-5", tools: [CMDC.Tool.ReadFile, CMDC.Tool.Shell], system_prompt: "你是一个专业的代码助手。", working_dir: "/my/project" ) # 2. 发送 prompt CMDC.prompt(session, "帮我列出当前目录的文件") # 3. 等待回复(阻塞直到 Agent 回到 idle) {:ok, reply} = CMDC.collect_reply(session, timeout: 30_000) IO.puts(reply) # 4. 停止会话 CMDC.stop(session) ## 流式接收 CMDC.subscribe(session) CMDC.prompt(session, "写一首诗") receive do {:cmdc_event, _sid, {:stream_chunk, _sid, chunk}} -> IO.write(chunk) {:cmdc_event, _sid, {:agent_end, _msgs, _usage}} -> :done end ## 通过 Blueprint 启动 {:ok, session} = CMDC.create_agent( blueprint: MyApp.CodingAgent, session_id: "my-session" ) ## 通过 Options struct 启动 opts = CMDC.Options.new!(model: "openai:gpt-4o", tools: [CMDC.Tool.Shell]) {:ok, session} = CMDC.create_agent(opts) ## HITL 审批 CMDC.subscribe(session) CMDC.prompt(session, "删除所有日志文件") receive do {:cmdc_event, _sid, {:approval_required, _sid, tool, args, ref}} -> IO.inspect({tool, args}, label: "批准工具调用?") CMDC.approve(session, ref) # 或 CMDC.reject(session, ref) end """ alias CMDC.{Agent, EventBus, SessionServer} # ========================================================================== # 类型 # ========================================================================== @type session :: pid() @type create_opts :: keyword() | CMDC.Options.t() # ========================================================================== # create_agent/1 # ========================================================================== @doc """ 创建并启动一个 Agent 会话。 支持三种调用方式: **1. keyword list(推荐)** {:ok, session} = CMDC.create_agent( model: "anthropic:claude-sonnet-4-5", tools: [CMDC.Tool.Shell], working_dir: "/project" ) **2. `CMDC.Options.t()` struct** opts = CMDC.Options.new!(model: "anthropic:claude-sonnet-4-5") {:ok, session} = CMDC.create_agent(opts) **3. Blueprint 模块** {:ok, session} = CMDC.create_agent(blueprint: MyApp.CodingAgent) ## 必填选项 - `:model` — 模型标识符,如 `"anthropic:claude-sonnet-4-5"`(Blueprint 启动时可省略) ## 可选选项 - `:session_id` — 会话唯一标识符,默认自动生成 - `:blueprint` — Blueprint 模块或 `CMDC.Blueprint.t()` struct - `:working_dir` — 工作目录,默认当前目录 - `:system_prompt` — 系统提示词 - `:tools` — Tool 模块列表 - `:plugins` — Plugin 列表 - `:config` — `CMDC.Config.t()` struct - `:provider_opts` — 传给 req_llm 的选项,如 `[api_key: "sk-..."]` - `:max_turns` — 最大轮数,默认 100 返回 `{:ok, session_pid}` 或 `{:error, reason}`。 """ @spec create_agent(create_opts()) :: {:ok, session()} | {:error, term()} def create_agent(%CMDC.Options{} = opts) do opts |> options_to_session_opts() |> SessionServer.start_link() end def create_agent(opts) when is_list(opts) do opts |> ensure_session_id() |> SessionServer.start_link() end # ========================================================================== # prompt/2 # ========================================================================== @doc """ 向 Agent 发送用户 prompt。 - 若 Agent 处于 idle,立即开始处理,返回 `%{queued: false}` - 若 Agent 正忙,消息入队,处理完当前任务后自动处理,返回 `%{queued: true}` ## 示例 %{queued: false} = CMDC.prompt(session, "帮我写一个 Elixir GenServer") """ @spec prompt(session(), String.t()) :: %{queued: boolean()} def prompt(session, text) when is_binary(text) do session |> SessionServer.agent() |> Agent.prompt(text) end # ========================================================================== # collect_reply/2 # ========================================================================== @doc """ 等待并收集 Agent 的最终文本回复。 订阅当前会话的 EventBus,等待 `{:agent_end, messages, _usage}` 事件, 从最后一条 assistant 消息中提取文本内容返回。 ## 选项 - `:timeout` — 等待超时时间(毫秒),默认 60_000(60 秒) ## 返回 - `{:ok, text}` — 成功收到回复文本 - `{:error, :timeout}` — 超时 - `{:error, reason}` — Agent 出错或被中止 ## 示例 CMDC.prompt(session, "你好") {:ok, reply} = CMDC.collect_reply(session, timeout: 30_000) IO.puts(reply) """ @spec collect_reply(session(), keyword()) :: {:ok, String.t()} | {:error, :timeout} | {:error, term()} def collect_reply(session, opts \\ []) do timeout = Keyword.get(opts, :timeout, 60_000) session_id = session_id(session) EventBus.subscribe(session_id) result = await_reply(session_id, timeout) EventBus.unsubscribe(session_id) result end # ========================================================================== # stop/2 # ========================================================================== @doc """ 停止 Agent 会话,清理所有资源。 优雅停止 `SessionServer` 监督树(级联停止 SubAgent.Supervisor 和主 Agent)。 ## 选项 - `:reason` — 停止原因,默认 `:normal` - `:timeout` — 等待停止超时(毫秒),默认 5_000 ## 示例 :ok = CMDC.stop(session) """ @spec stop(session(), keyword()) :: :ok def stop(session, opts \\ []) do reason = Keyword.get(opts, :reason, :normal) timeout = Keyword.get(opts, :timeout, 5_000) Supervisor.stop(session, reason, timeout) end # ========================================================================== # subscribe / unsubscribe # ========================================================================== @doc """ 订阅当前进程接收指定 Agent 会话的所有事件。 订阅后,当前进程将收到 `{:cmdc_event, session_id, event}` 格式的消息。 常见事件: - `{:stream_chunk, session_id, chunk}` — 流式 token - `{:tool_start, session_id, tool_name, args}` — 工具开始 - `{:tool_end, session_id, tool_name, result}` — 工具完成 - `{:agent_end, messages, token_usage}` — 回复完成 - `{:approval_required, session_id, tool_name, args, ref}` — 等待审批 ## 示例 CMDC.subscribe(session) CMDC.prompt(session, "hello") receive do {:cmdc_event, _sid, {:agent_end, _msgs, _usage}} -> IO.puts("done") end """ @spec subscribe(session()) :: {:ok, pid()} | {:error, term()} def subscribe(session) do session |> session_id() |> EventBus.subscribe() end @doc "取消当前进程对指定 Agent 会话的事件订阅。" @spec unsubscribe(session()) :: :ok def unsubscribe(session) do session |> session_id() |> EventBus.unsubscribe() end # ========================================================================== # abort / approve / reject # ========================================================================== @doc "中止当前 Agent 运行中的任务,Agent 回到 idle 状态。" @spec abort(session()) :: :ok def abort(session) do session |> SessionServer.agent() |> Agent.abort() end @doc """ 批准一个待审批的工具调用(HITL 审批流)。 `approval_id` 来自 `{:approval_required, _sid, _tool, _args, approval_id}` 事件。 """ @spec approve(session(), String.t()) :: :ok def approve(session, approval_id) when is_binary(approval_id) do session |> SessionServer.agent() |> Agent.approve(approval_id) end @doc "拒绝一个待审批的工具调用(HITL 审批流)。" @spec reject(session(), String.t()) :: :ok def reject(session, approval_id) when is_binary(approval_id) do session |> SessionServer.agent() |> Agent.reject(approval_id) end # ========================================================================== # status / agent_pid # ========================================================================== @doc """ 获取 Agent 当前状态快照。 返回: %{ state: :idle | :running | :streaming | :executing_tools, session_id: String.t(), turns: non_neg_integer(), cost_usd: float(), uptime_ms: non_neg_integer(), timestamp_ms: integer() } """ @spec status(session()) :: map() def status(session) do session |> SessionServer.agent() |> Agent.status() end @doc "返回会话对应的主 Agent pid。" @spec agent_pid(session()) :: pid() | nil def agent_pid(session), do: SessionServer.agent(session) @doc "返回会话的 session_id 字符串。" @spec session_id(session()) :: String.t() def session_id(session) do session |> SessionServer.agent() |> Agent.status() |> Map.fetch!(:session_id) end # ========================================================================== # 私有辅助 # ========================================================================== defp ensure_session_id(opts) do if Keyword.has_key?(opts, :session_id) do opts else Keyword.put(opts, :session_id, generate_session_id()) end end defp generate_session_id do :crypto.strong_rand_bytes(8) |> Base.encode16(case: :lower) end defp options_to_session_opts(%CMDC.Options{} = opts) do opts |> Map.from_struct() |> Enum.reject(fn {_k, v} -> is_nil(v) end) |> Keyword.new() |> ensure_session_id() end defp await_reply(session_id, timeout) do receive do {:cmdc_event, ^session_id, {:agent_end, messages, _token_usage}} -> reply_text = extract_last_assistant(messages) {:ok, reply_text} {:cmdc_event, ^session_id, {:agent_abort, reason}} -> {:error, {:aborted, reason}} {:cmdc_event, ^session_id, {:error, _sid, reason}} -> {:error, reason} after timeout -> {:error, :timeout} end end defp extract_last_assistant(messages) do messages |> Enum.filter(&(&1.role == :assistant)) |> List.last() |> case do nil -> "" msg -> msg.content || "" end end end