DSEx 01: Real LM Front Door

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

Program, Don’t Prompt

This notebook is the DSEx front door. It mirrors the core DSP idea: declare a task, run it as a program, then grow the same task into tools, evaluation, optimization, and persistence.

Manual path: real provider shape first, deterministic development next, evaluation and optimization after that, then tools and operations.

It expects OPENAI_API_KEY and OPENAI_MODEL in the environment. Without those variables the live cells return {:skip, reason} so the notebook remains safe to execute in local gates.

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 cells."}
  end
end

Extract Structured Data

Start with a signature. It names the inputs and outputs that the rest of the system can test, evaluate, optimize, and persist.

case live_lm.(max_completion_tokens: 120) do
  {:ok, lm} ->
    extract =
      DSEx.predict(
        DSEx.signature(
          "email -> event_name: string, date: string",
          "Extract the event name and date from the email. Return JSON only."
        ),
        lm: lm,
        adapter: DSEx.Adapter.JSON,
        config: [json_retries: 1]
      )

    {:ok, prediction} =
      DSEx.call(extract, %{
        email: "Team Offsite moved to Thursday, June 5. Bring questions for planning."
      })

    event_name = prediction |> DSEx.get(:event_name, "") |> to_string() |> String.downcase()
    date = prediction |> DSEx.get(:date, "") |> to_string() |> String.downcase()

    unless String.contains?(event_name, "offsite") and
             (String.contains?(date, "june") or String.contains?(date, "thursday") or
                String.contains?(date, "06-05") or String.contains?(date, "6/5")) do
      raise "live extraction proof failed: #{inspect(DSEx.to_map(prediction))}"
    end

    DSEx.to_map(prediction)

  skip ->
    skip
end

Change The Module, Keep The Task

The task contract stays stable while the execution strategy changes. A module is a program shape around the same signature idea.

case live_lm.(max_completion_tokens: 160) do
  {:ok, lm} ->
    classify =
      DSEx.chain_of_thought(
        DSEx.signature(
          "ticket -> reasoning: string, urgency: enum[low,high], team: string",
          "Route the support ticket. Return JSON only."
        ),
        lm: lm,
        adapter: DSEx.Adapter.JSON,
        config: [json_retries: 1]
      )

    {:ok, prediction} =
      DSEx.call(classify, %{
        ticket: "Production checkout is failing for all EU customers."
      })

    unless DSEx.get(prediction, :urgency) == "high" and
             prediction |> DSEx.get(:reasoning, "") |> to_string() |> byte_size() > 0 do
      raise "live ChainOfThought proof failed: #{inspect(DSEx.to_map(prediction))}"
    end

    DSEx.to_map(prediction)

  skip ->
    skip
end

Add Conversation History

History is prior task data keyed by the signature fields. DSEx renders it as prior turns for the model, while the program still receives named inputs and returns named outputs.

case live_lm.(max_completion_tokens: 120) do
  {:ok, lm} ->
    qa =
      DSEx.predict(
        DSEx.signature(
          "question, history -> answer: short_span",
          "Answer the latest question using the conversation history when relevant."
        ),
        lm: lm
      )

    history =
      DSEx.history([
        %{question: "What is the capital of France?", answer: "Paris"},
        %{question: "What country is Paris in?", answer: "France"}
      ])

    {:ok, prediction} =
      DSEx.call(qa, %{question: "What city did we identify first?", history: history})

    answer = prediction |> DSEx.get(:answer, "") |> to_string() |> String.downcase()

    unless String.contains?(answer, "paris") do
      raise "live history proof failed: #{inspect(DSEx.to_map(prediction))}"
    end

    DSEx.to_map(prediction)

  skip ->
    skip
end

Add Tools With ReAct

Only add tools when the task needs action outside the LM. ReAct keeps tool use inside an explicit policy and final submit boundary.

case live_lm.(max_completion_tokens: 180) do
  {:ok, lm} ->
    lookup =
      DSEx.tool(:lookup, "Lookup a fact by query.", fn
        %{query: "capital-france"} -> "Paris"
        %{"query" => "capital-france"} -> "Paris"
        other -> {:error, {:unexpected_query, other}}
      end,
        schema: %{
          "type" => "object",
          "properties" => %{"query" => %{"type" => "string", "enum" => ["capital-france"]}},
          "required" => ["query"]
        }
      )

    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, prediction} =
      Enum.reduce_while(1..3, {:error, :not_run}, fn _attempt, _last ->
        case DSEx.call(react, %{question: "What is the capital of France?"}) do
          {:ok, prediction} -> {:halt, {:ok, prediction}}
          {:error, _reason} = error -> {:cont, error}
        end
      end)

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

    DSEx.to_map(prediction)

  skip ->
    skip
end

Evaluate And Compile

Use deterministic examples to prove your metric and workflow before spending provider calls on larger runs. Livebook 03 expands this into the normal evaluation and optimization workflow.

static_lm = %{
  module: DSEx.LM.Static,
  opts: [handler: fn _messages, _opts -> %{answer: "Paris"} end]
}

program = DSEx.predict("question -> answer", lm: static_lm)

trainset = [
  DSEx.example(question: "Eiffel Tower city?", answer: "Paris")
  |> DSEx.with_inputs(:question)
]

devset = [
  DSEx.example(question: "Capital of France?", answer: "Paris")
  |> DSEx.with_inputs(:question)
]

metric = DSEx.exact_match(:answer)

compiled =
  program
  |> DSEx.optimize(
    DSEx.Optimizer.RandomSearch.new(metric, candidates: 2, demos_per_candidate: 1),
    trainset,
    devset
  )

{DSEx.evaluate(program, devset, metric).score, DSEx.evaluate(compiled, devset, metric).score}

Save Without Secrets

Persistence stores the program shape and safe configuration, not live credentials. Livebook 05 returns to this from the operations side.

case live_lm.(temperature: 0) do
  {:ok, lm} ->
    program = DSEx.predict("question -> answer", lm: lm)
    state = DSEx.dump(program)

    proof = %{
      credential_leaked?: inspect(state) =~ System.fetch_env!("OPENAI_API_KEY"),
      loadable?: match?(%{__struct__: DSEx.Predict.Predict}, DSEx.load(state))
    }

    unless proof == %{credential_leaked?: false, loadable?: true} do
      raise "live save/load proof failed: #{inspect(proof)}"
    end

    proof

  skip ->
    skip
end

Next: open livebooks/02_programming_not_prompting.livemd to run the same programming model without provider calls and inspect the generated messages.