ExAgent.Patterns.MapReduce (ExAgent v0.4.1)

Copy Markdown View Source

Split one big input into pieces, process them in parallel, combine the results.

The analogy: reading a long report with colleagues. You each take a chapter, summarise it, then someone stitches the summaries into one. Nobody reads the whole thing.

Use it when the input does not fit - or fits but answers worse whole than in parts. A 300-page contract, a week of logs, forty customer interviews.

alias ExAgent.Patterns.MapReduce

{:ok, summary} =
  MapReduce.run(chapters, provider,
    map: &"Summarise this chapter in three bullets:\n\n#{&1}",
    reduce: fn summaries ->
      {:ok, "Overall:\n" <> Enum.join(summaries, "\n")}
    end
  )

Pass reduce: {provider, prompt_builder} to have a model do the combining, which is the usual case:

MapReduce.run(chapters, worker,
  map: &"Summarise:\n\n#{&1}",
  reduce: {editor, fn parts -> "Merge these into one summary:\n\n" <> Enum.join(parts, "\n---\n") end}
)

Sections are independent, and that is the catch

Each piece is processed with no knowledge of the others, so anything that spans a boundary is invisible: a clause on page 40 that contradicts page 7, a log line that only matters next to one an hour earlier. Overlap your sections, or make the reduce step look for conflicts explicitly.

Partial failure is reported, not hidden

One section failing does not fail the run. The reduce step receives only the sections that succeeded, and :failures on the result tells you what was missing - a summary built from 38 of 40 interviews is usually still worth having, but never worth mistaking for all 40.

When not to use it

If the pieces need to know about each other, this is the wrong shape - chain them with ExAgent.Patterns.Chain so each sees the last. If you are sending the same input to several models rather than different inputs to one, you want ExAgent.Patterns.Consensus.

Summary

Types

Where to send a prompt: a provider struct for a stateless call, or a running agent when the step should remember the conversation.

Functions

Maps sections through target in parallel, then reduces the outputs.

Types

map_reduce_opts()

@type map_reduce_opts() :: [
  map: (section() -> String.t()),
  reduce: reduce(),
  timeout: pos_integer(),
  max_concurrency: pos_integer()
]

reduce()

@type reduce() ::
  ([String.t()] -> {:ok, term()} | {:error, term()})
  | {target(), ([String.t()] -> String.t())}

result()

@type result() :: %{
  output: term(),
  sections: non_neg_integer(),
  failures: [{non_neg_integer(), term()}]
}

section()

@type section() :: term()

target()

@type target() :: struct() | GenServer.server()

Where to send a prompt: a provider struct for a stateless call, or a running agent when the step should remember the conversation.

Functions

run(sections, target, opts)

@spec run([section()], target(), map_reduce_opts()) ::
  {:ok, result()} | {:error, term()}

Maps sections through target in parallel, then reduces the outputs.

Options

  • :map (required) - (section -> prompt)
  • :reduce (required) - either (outputs -> {:ok, value} | {:error, reason}), or {target, (outputs -> prompt)} to have a model combine them

  • :timeout - per section, in milliseconds (default: 5 minutes)
  • :max_concurrency - parallel sections (default: System.schedulers_online/0, which is a poor proxy for how many requests a provider will tolerate - lower it if you are being rate limited)

Returns {:ok, result}, or {:error, {:all_sections_failed, failures}} when nothing survived to reduce - the failures come with it, since "everything failed" on its own does not say why.