Sagents.AgentUtils (Sagents v0.16.2)

Copy Markdown

Shared utilities for Agent and SubAgent HITL (Human-in-the-Loop) support.

Provides common functions for:

  • Checking for HITL interrupts
  • Building full decisions lists
  • Extracting tool calls from chains

These utilities provide consistent HITL behavior across Agent and SubAgent implementations.

Summary

Functions

Compute the state transition for a single HITL approve/reject decision in a host's pending-tool list.

Build a full decisions list for ALL tool calls in a message.

Check if a chain has pending tool calls that require human approval.

Changes that fully clear all interrupt-derived host state.

Extract tool calls from the last assistant message in a chain.

Map an interrupt-data payload to the assigns/state changes a host (LiveView, GenServer, etc.) should merge to display the interrupt.

Pair each HITL action request with the human decision that answers it.

Resolve the callback handler maps to run for an execution.

Changes a host should merge when the agent process goes away.

Functions

advance_hitl_decisions(state, tool_ref, decision_type)

@spec advance_hitl_decisions(map(), non_neg_integer() | String.t(), atom()) ::
  {:resume, [map()], map()} | {:more, map()} | {:error, :not_pending}

Compute the state transition for a single HITL approve/reject decision in a host's pending-tool list.

Reads :pending_tools and :hitl_decisions from state (treating missing keys as empty lists) and records decision_type for one pending tool. The tool is identified by either:

  • its tool_call_id (a string), or
  • its current position in :pending_tools (an integer). Positions shift as tools are decided, so a UI that shows several pending tools at once should address them by tool_call_id.

Each recorded decision carries the :tool_call_id of the tool it was made for, and Sagents.Middleware.HumanInTheLoop matches decisions to tools by that id on resume. Tools may therefore be decided in any order.

Returns:

  • {:resume, accumulated_decisions, changes}: all pending tools have been decided. The host should call Sagents.AgentServer.resume(agent_id, accumulated_decisions) and then merge changes (which clears :pending_tools, :interrupt_data, and :hitl_decisions).
  • {:more, changes}: tools still pending. Merge changes (which advances :pending_tools and :hitl_decisions).
  • {:error, :not_pending}: no pending tool matches. The index is out of range, or the tool_call_id was already decided (for example, a repeated click). Nothing is recorded.

Example

case AgentUtils.advance_hitl_decisions(socket.assigns, tool_call_id, :approve) do
  {:resume, decisions, changes} ->
    AgentServer.resume(agent_id, decisions)
    {:noreply, assign(socket, changes)}

  {:more, changes} ->
    {:noreply, assign(socket, changes)}

  {:error, :not_pending} ->
    {:noreply, socket}
end

build_full_decisions(all_tool_calls, hitl_tool_call_ids, human_decisions, action_requests)

Build a full decisions list for ALL tool calls in a message.

Mixes human decisions (for HITL tools) with auto-approvals (for non-HITL tools). This is needed because LLMChain.execute_tool_calls_with_decisions expects a decision for EVERY tool call, not just the ones that needed approval.

Parameters

  • all_tool_calls: All tool calls from assistant message (HITL + non-HITL)
  • hitl_tool_call_ids: List of tool_call_ids that needed human approval
  • human_decisions: List of decisions from human, paired with action_requests as described in pair_decisions/2
  • action_requests: List of action_requests (to map decisions to tool_call_ids)

Returns

  • List of decisions matching all_tool_calls order

Example

all_tool_calls = [tc1, tc2, tc3]  # 3 total tool calls
hitl_tool_call_ids = [tc1.call_id]  # Only tc1 needed approval
action_requests = [%{tool_call_id: tc1.call_id, ...}]
human_decisions = [%{type: :approve}]  # Human approved tc1

full_decisions = build_full_decisions(all_tool_calls, hitl_tool_call_ids, human_decisions, action_requests)
# => [%{type: :approve}, %{type: :approve}, %{type: :approve}]
# First is human decision, others are auto-approved

check_for_hitl_interrupt(chain, interrupt_on)

Check if a chain has pending tool calls that require human approval.

Parameters

  • chain: LLMChain with potentially pending tool calls
  • interrupt_on: Map of tool_name => true/false/config

Returns

  • {:interrupt, interrupt_data} - Some tools need approval
  • :continue - No tools need approval or no tool calls present

Example

case AgentUtils.check_for_hitl_interrupt(chain, %{"write_file" => true}) do
  {:interrupt, interrupt_data} ->
    # Pause and request human decisions
    {:interrupt, state, interrupt_data}

  :continue ->
    # Execute tools automatically
    execute_tools(chain)
end

cleared_interrupt_changes()

@spec cleared_interrupt_changes() :: map()

Changes that fully clear all interrupt-derived host state.

Single source of truth for the complete set of pending_* / interrupt-related keys a host carries. Merge this into any status transition that lands on a non-interrupted status (:running, :idle, :cancelled, :error, :not_running) to enforce the invariant: any status other than :interrupted means there is no pending interrupt UI.

This prevents a dismissed/superseded interrupt's derived keys (e.g. a :pending_halt) from surviving invisibly behind a status == :interrupted render guard and re-appearing on the next transition back into :interrupted.

Example

def handle_status_idle,
  do: Map.merge(AgentUtils.cleared_interrupt_changes(),
        %{loading: false, agent_status: :idle, streaming_delta: nil})

get_tool_calls_from_last_message(chain)

Extract tool calls from the last assistant message in a chain.

Parameters

  • chain: LLMChain

Returns

  • List of tool calls or empty list

Example

tool_calls = AgentUtils.get_tool_calls_from_last_message(chain)
# => [%ToolCall{call_id: "1", name: "write_file", ...}, ...]

interrupt_session_changes(halt)

@spec interrupt_session_changes(map() | nil) :: map()

Map an interrupt-data payload to the assigns/state changes a host (LiveView, GenServer, etc.) should merge to display the interrupt.

Routes the four sagents-internal interrupt variants to their UI shapes:

  • :halt — surface as a terminal halt with :pending_halt
  • :ask_user_question — pull the single question to the front
  • :multiple_interrupts — "halt wins": if any sub-interrupt is :halt, surface as a halt; otherwise present as questions if all are questions, or as a HITL tool batch
  • :subagent_hitl — unwrap the inner action_requests
  • any other map — treat as a generic HITL tool batch

Returns an empty map for nil, so callers can apply the result unconditionally on top of base assigns.

Returned keys

When a halt is present: :pending_halt, :pending_question, :pending_tools

When questions are present: :pending_question, :remaining_questions, :question_responses, :pending_tools

When HITL tools are present: :pending_tools, :pending_question

Example

base = %{loading: false, agent_status: :interrupted, interrupt_data: data}
Map.merge(base, AgentUtils.interrupt_session_changes(data))

pair_decisions(action_requests, decisions)

@spec pair_decisions([map()], [term()]) ::
  {:ok, [{map(), term()}]} | {:error, String.t()}

Pair each HITL action request with the human decision that answers it.

Decisions are matched in one of two ways:

  • By :tool_call_id. When every decision carries a :tool_call_id, each one is matched to the action request with that id, so the order of decisions does not matter. Every action request must be answered exactly once. advance_hitl_decisions/3 produces decisions in this form.
  • By position. When no decision carries a :tool_call_id, decision N answers action request N.

A list mixing keyed and unkeyed decisions is rejected, because there is no safe way to tell which request an unkeyed decision was meant for.

Returns {:ok, [{action_request, decision}]} in action_requests order, or {:error, reason}.

Example

action_requests = [%{tool_call_id: "a", ...}, %{tool_call_id: "b", ...}]

AgentUtils.pair_decisions(action_requests, [
  %{type: :reject, tool_call_id: "b"},
  %{type: :approve, tool_call_id: "a"}
])
# => {:ok, [
#      {%{tool_call_id: "a", ...}, %{type: :approve, tool_call_id: "a"}},
#      {%{tool_call_id: "b", ...}, %{type: :reject, tool_call_id: "b"}}
#    ]}

resolve_callbacks(middleware, opts)

@spec resolve_callbacks(
  [Sagents.MiddlewareEntry.t()],
  keyword()
) :: [map()]

Resolve the callback handler maps to run for an execution.

Collecting middleware callbacks is intrinsic to running an agent: this always collects middleware's callbacks (via Sagents.Middleware.collect_callbacks/1) and merges any caller-supplied :callbacks from opts on top rather than substituting them. Supplying :callbacks therefore never disables an agent's middleware callbacks.

Supplied callbacks come first to preserve the historical ordering of the server path (PubSub broadcasting, then middleware handlers). Empty %{} maps are dropped so a %{} default never adds a no-op callback to the chain.

Both execution engines share this contract: Sagents.Agent passes its own agent.middleware, while Sagents.SubAgent passes the inherited parent_middleware carried in its chain's custom_context. The server layers pass only their PubSub callbacks via :callbacks; collection happens here, so middleware is never collected twice.

Parameters

  • middleware: List of Sagents.MiddlewareEntry structs to collect from
  • opts: Keyword list; reads the optional :callbacks key (a map or list of maps)

Example

callbacks = AgentUtils.resolve_callbacks(agent.middleware, opts)

shutdown_session_changes(state, shutdown_data)

@spec shutdown_session_changes(map(), map()) :: map()

Changes a host should merge when the agent process goes away.

Takes the host's state map and the {:agent_shutdown, shutdown_data} payload (see Sagents.AgentServer for its shape).

Liveness is not status

agent_status describes what the conversation is waiting on. agent_alive? describes whether a producer process is backing this session right now. They are independent facts, and conflating them is what loses an open question when the agent naps: the author still owes an answer, so the conversation is still :interrupted, even though nothing is running.

Hosts should therefore carry both keys:

  • merge agent_alive?: true in every handle_status_* handler (an inbound agent event is proof of life)
  • merge agent_alive?: false here, and in the :DOWN handler
  • gate "wake the agent" affordances on agent_alive?, not on agent_status == :not_running

Behaviour

  • The session has a pending interrupt and shutdown_data.interrupt_restorable is true: keep every pending_* key, keep :interrupt_data, and keep agent_status: :interrupted. Returns only %{agent_alive?: false, loading: false, streaming_delta: nil}. The prompt stays on screen; answering it wakes an agent that boots straight back into :interrupted (see Sagents.Session.resume/4).
  • Otherwise: today's behaviour, cleared_interrupt_changes/0 plus agent_status: :not_running.
  • :interrupt_restorable absent (an agent predating this field): treated as not restorable. A version-skewed pair degrades to the old clear-everything behaviour rather than to a prompt no agent can accept.

Idempotent: a shutdown delivers this event more than once, and every delivery carries the same fields, so re-merging the result is a no-op.

What this does not do

It does not touch the Sagents.Subscriber subs map. Unsubscribing here would delete the entry, and Sagents.Subscriber.handle_presence_diff/3 only revives entries in the :pending state, so the subscription could never come back without a remount. Leave it alone: handle_publisher_down/3 flips it to :pending on the :DOWN that is already on its way, and the next presence arrival re-subscribes. That is the designed recovery path and exactly what is wanted after a wake.

It also does not clear :agent_id. That is a pure function of the conversation id, it is what the host needs in order to wake the agent again, and nil-ing it breaks any handler that compares an inbound event's agent id against it.

Example

def handle_agent_shutdown(state, shutdown_data) do
  AgentUtils.shutdown_session_changes(state, shutdown_data)
end