Imp.Predict.Parallel (Imp v0.5.0)

Run a program across many inputs concurrently through a supervised BEAM task boundary.

map/3 is the homogeneous batch primitive. run/2 accepts a tree of {program, inputs} pairs when one workflow needs to execute different programs concurrently. Both run through the supervised Imp task boundary and preserve input order and nesting.

Successful calls return {:ok, %Imp.Prediction{}}. Program crashes, throws, invalid return shapes, and task exits are returned as per-input {:error, reason} tuples so one bad input does not bring down the whole batch.

iex> lm = Imp.LM.Static.new(
...>   handler: fn messages, _opts ->
...>     prompt = Enum.map_join(messages, "\n", &Map.fetch!(&1, :content))
...>
...>     cond do
...>       prompt =~ "alpha" -> %{answer: "A"}
...>       prompt =~ "beta" -> %{answer: "B"}
...>     end
...>   end
...> )
iex> program = Imp.Predict.new("question -> answer", lm: lm)
iex> results = Imp.Predict.Parallel.map(program, [%{question: "alpha"}, %{question: "beta"}])
iex> Enum.map(results, fn {:ok, prediction} -> Imp.Prediction.get(prediction, :answer) end)
["A", "B"]

Bad inputs remain local to their result slot:

iex> lm = Imp.LM.Static.new(
...>   handler: fn messages, _opts ->
...>     prompt = Enum.map_join(messages, "\n", &Map.fetch!(&1, :content))
...>     if prompt =~ "bad", do: raise("boom"), else: %{answer: "ok"}
...>   end
...> )
iex> program = Imp.Predict.new("question -> answer", lm: lm)
iex> results = Imp.Predict.Parallel.map(program, [%{question: "ok"}, %{question: "bad"}])
iex> match?([{:ok, %Imp.Prediction{}}, {:error, {:lm_failed, Imp.LM.Static, %RuntimeError{}}}], results)
true
iex> [{:ok, prediction}, _error] = results; Imp.Prediction.get(prediction, :answer)
"ok"

Summary

Functions

map(program, inputs, opts \\ [])

Maps program over inputs concurrently.

run(exec_pairs, opts \\ [])

Runs heterogeneous {program, inputs} pairs concurrently.

Functions

map(program, inputs, opts \\ [])

Maps program over inputs concurrently.

Options:

:num_threads - how many calls run at once, as in DSPy's dspy.Parallel. Defaults to System.schedulers_online/0.
:timeout - task timeout accepted by Task.async_stream/5. Defaults to 30_000.
:on_timeout - either :kill_task or :exit. Defaults to :kill_task.

run(exec_pairs, opts \\ [])

Runs heterogeneous {program, inputs} pairs concurrently.

Nested lists preserve their shape while every leaf shares one bounded task pool. This avoids nested task-pool deadlocks and makes diverse composed work as cheap to supervise as a homogeneous batch:

iex> first = Imp.predict("question -> answer", lm: Imp.LM.Static.new(handler: fn _, _ -> %{answer: "first"} end))
iex> second = Imp.predict("topic -> answer", lm: Imp.LM.Static.new(handler: fn _, _ -> %{answer: "second"} end))
iex> [first_result, [second_result]] = Imp.Predict.Parallel.run([
...>   {first, %{question: "q"}},
...>   [{second, %{topic: "t"}}]
...> ])
iex> {:ok, first_prediction} = first_result
iex> {:ok, second_prediction} = second_result
iex> {Imp.get(first_prediction, :answer), Imp.get(second_prediction, :answer)}
{"first", "second"}

The options are the same as map/3.