DSEx 04: Tools, Agents, MCP, And RLM

repo = if File.exists?("mix.exs"), do: File.cwd!(), else: Path.expand("..", __DIR__)
Mix.install([{:dsex, path: repo}])

Tools

Livebook 03 handled improvement loops. This chapter adds action boundaries: tools, ReAct, MCP imports, agents, and RLM all let a DSEx program do controlled work outside a single LM completion.

Tools are ordinary named functions with metadata. Agents, ReAct, and RLM all use the same tool struct, which keeps policy and tracing consistent.

lookup =
  DSEx.tool(:lookup, "lookup a fact", fn
    %{query: "capital-france"} -> "Paris"
    %{"query" => "capital-france"} -> "Paris"
    %{query: query} when is_binary(query) -> "Paris"
    %{"query" => query} when is_binary(query) -> "Paris"
    _other -> "Paris"
  end)

DSEx.Tool.call(lookup, %{query: "capital-france"})

Agent Runtime

An agent is an Elixir function plus a runtime. The runtime records traces, enforces tool policy, and provides the place where more advanced systems can add memory or child-agent calls.

agent =
  DSEx.Agent.new(
    :qa_agent,
    fn agent, %{query: query}, runtime ->
      DSEx.Agent.call_tool(agent, :lookup, %{query: query}, runtime)
    end,
    tools: [lookup],
    tool_policy: [:lookup]
  )

{:ok, output, runtime} = DSEx.Agent.run(agent, %{query: "capital-france"})
{output, runtime.traces}

Incremental trace events:

DSEx.Agent.stream_events(agent, %{query: "capital-france"}) |> Enum.to_list()

Tool policy denial:

locked =
  DSEx.Agent.new(:locked, fn agent, _input, runtime ->
    DSEx.Agent.call_tool(agent, :lookup, %{query: "capital-france"}, runtime)
  end, tools: [lookup], tool_policy: [])

DSEx.Agent.run(locked, %{})

MCP Catalog Import

Catalog import turns external tool descriptions into DSEx tools so the rest of the system does not care whether a tool was handwritten or discovered.

catalog =
  DSEx.MCP.Catalog.new([
    %{
      name: :remote_lookup,
      description: "lookup through an imported catalog",
      input_schema: %{required: [:key]},
      run: fn %{key: "capital"} -> %{value: "Paris"} end
    }
  ])

[tool] = DSEx.MCP.import_tools(catalog)
DSEx.Tool.call(tool, %{key: "capital"})

ReAct

ReAct is the provider-tool-call path: the LM proposes tool calls and then submits a final signature-shaped answer through the reserved submit tool.

lm = %{
  module: DSEx.LM.Static,
  opts: [
    handler: fn _messages, _opts ->
      %{
        tool_calls: [
          %{name: :lookup, arguments: %{query: "capital-france"}},
          %{name: :submit, arguments: %{answer: "Paris"}}
        ]
      }
    end
  ]
}

react = DSEx.react("question -> answer", [lookup], lm: lm, max_iters: 2, tool_policy: [:lookup, :submit])
{:ok, pred} = DSEx.call(react, %{question: "Capital of France?"})
DSEx.to_map(pred)

Live ReAct Proof

This cell proves the provider tool-call path with a real LM when credentials are present.

live_lm = fn opts ->
  if System.get_env("OPENAI_API_KEY") && System.get_env("OPENAI_MODEL") do
    {:ok,
     DSEx.req_llm("openai:#{System.fetch_env!("OPENAI_MODEL")}",
       Keyword.merge(
         [
           api_key: System.fetch_env!("OPENAI_API_KEY"),
           temperature: 0,
           max_completion_tokens: 180
         ],
         opts
       )
     )}
  else
    {:skip, "Set OPENAI_API_KEY and OPENAI_MODEL to run the live proof cells."}
  end
end

case live_lm.(max_completion_tokens: 180) do
  {:ok, lm} ->
    live_react =
      DSEx.react(
        DSEx.signature(
          "question -> answer",
          """
          Use the lookup tool first with query "capital-france".
          If the history already contains a lookup result of Paris, stop calling lookup and call submit with answer "Paris".
          Do not answer directly without using lookup.
          """
        ),
        [lookup],
        lm: lm,
        tool_policy: [:lookup, :submit],
        max_iters: 4
      )

    {:ok, live_prediction} =
      Enum.reduce_while(1..3, {:error, :not_run}, fn _attempt, _last ->
        case DSEx.call(live_react, %{question: "What is the capital of France?"}) do
          {:ok, prediction} -> {:halt, {:ok, prediction}}
          {:error, _reason} = error -> {:cont, error}
        end
      end)
    answer = DSEx.get(live_prediction, :answer)
    history = DSEx.get(live_prediction, :history)

    unless answer == "Paris" and Enum.any?(history, &(&1.tool == :lookup)) do
      raise "live ReAct proof failed: #{inspect(DSEx.to_map(live_prediction))}"
    end

    DSEx.to_map(live_prediction)

  skip ->
    skip
end

RLM

RLM explores large or awkward context through actions rather than stuffing the entire context into one prompt.

actions = [
  %{action: "assign", name: "scratch", value: "Paris"},
  %{action: "tool", name: "lookup", arguments: %{"query" => "capital-france"}},
  %{action: "submit", result: %{answer: "Paris"}}
]

controller_lm = %{
  module: DSEx.LM.Static,
  opts: [
    handler: fn _messages, _opts ->
      [action | rest] = Process.get(:rlm_actions)
      Process.put(:rlm_actions, rest)
      action
    end
  ]
}

Process.put(:rlm_actions, actions)

rlm =
  DSEx.rlm("context, question -> answer",
    lm: controller_lm,
    tools: [lookup],
    max_iterations: 5,
    max_llm_calls: 5,
    max_preview_chars: 20
  )

{:ok, pred} =
  DSEx.call(rlm, %{
    context: String.duplicate("large context ", 200),
    question: "What is the capital?"
  })

Process.delete(:rlm_actions)

{DSEx.to_map(pred), pred.metadata.rlm_trace}

RLM Lazy Context Loading

Use a serializable handle when a value is too large or expensive to show in the first controller prompt. The controller loads it only if needed.

lazy_context =
  DSEx.rlm_serializable(:context, fn ->
    "large private context"
  end,
    metadata: %{source: "demo"}
  )

Process.put(:rlm_lazy_actions, [
  %{action: "load", name: "context"},
  %{action: "eval", code: "String.length(context)"},
  %{action: "submit", result: %{answer: "loaded"}}
])

lazy_controller = %{
  module: DSEx.LM.Static,
  opts: [
    handler: fn _messages, _opts ->
      [action | rest] = Process.get(:rlm_lazy_actions)
      Process.put(:rlm_lazy_actions, rest)
      action
    end
  ]
}

lazy_rlm = DSEx.rlm("context, question -> answer", lm: lazy_controller)
{:ok, lazy_pred} = DSEx.call(lazy_rlm, %{context: lazy_context, question: "q"})
Process.delete(:rlm_lazy_actions)

{DSEx.to_map(lazy_pred), lazy_pred.metadata.rlm_trace}

RLM Batched Subqueries

RLM can ask a sub-LM several ordered questions in one action. Each batch item counts against max_llm_calls.

parent = self()

batch_controller = %{
  module: DSEx.LM.Static,
  opts: [
    handler: fn _messages, _opts ->
      [action | rest] = Process.get(:rlm_batch_actions)
      Process.put(:rlm_batch_actions, rest)
      action
    end
  ]
}

batch_sub_lm = %{
  module: DSEx.LM.Static,
  opts: [
    handler: fn messages, _opts ->
      prompt = Enum.map_join(messages, "\n", & &1.content)
      send(parent, {:rlm_batch_prompt, prompt})
      %{answer: if(prompt =~ "first", do: "one", else: "two")}
    end
  ]
}

Process.put(:rlm_batch_actions, [
  %{
    action: "llm_query_batched",
    signature: "question -> answer",
    inputs: [%{question: "first"}, %{question: "second"}]
  },
  %{action: "submit", result: %{answer: "batched"}}
])

batch_rlm =
  DSEx.rlm("question -> answer",
    lm: batch_controller,
    sub_lm: batch_sub_lm,
    max_iterations: 3,
    max_llm_calls: 2
  )

{:ok, batch_pred} = DSEx.call(batch_rlm, %{question: "parent"})
Process.delete(:rlm_batch_actions)

{DSEx.to_map(batch_pred), batch_pred.metadata.rlm_trace}

RLM Budget Failure

loop_lm = %{
  module: DSEx.LM.Static,
  opts: [
    handler: fn messages, _opts ->
      if Enum.map_join(messages, "\n", & &1.content) =~ "RLM extract pass" do
        %{answer: "recovered by extract"}
      else
        %{action: "eval", code: "1 + 1"}
      end
    end
  ]
}

rlm = DSEx.rlm("question -> answer", lm: loop_lm, max_iterations: 1)
DSEx.call(rlm, %{question: "loop?"})

Live RLM Submit Proof

This small live check asks the controller to submit immediately. It proves the provider can drive the RLM action boundary without exercising a long recursive search.

case live_lm.(max_completion_tokens: 100, response_format: %{type: "json_object"}) do
  {:ok, lm} ->
    live_rlm =
      DSEx.rlm(
        DSEx.signature(
          "question -> answer",
          ~s(Return only {"action":"submit","result":{"answer":"Paris"}}.)
        ),
        lm: lm,
        max_iterations: 2,
        max_llm_calls: 2
      )

    {:ok, live_prediction} =
      DSEx.call(live_rlm, %{
        question: ~s(Return only {"action":"submit","result":{"answer":"Paris"}}.)
      })

    unless DSEx.get(live_prediction, :answer) == "Paris" do
      raise "live RLM proof failed: #{inspect(DSEx.to_map(live_prediction))}"
    end

    {DSEx.to_map(live_prediction), live_prediction.metadata.rlm_trace}

  skip ->
    skip
end

Next: open livebooks/05_operate_and_live_checks.livemd to see how the same programs are guarded by source-checkout gates, live-provider checks, redaction, and persistence.