foley.agent

The SELECT stage — the search-agent that finds the right sound for a context (#7).

The headline capability: a narrative context → verified, license-clean sound candidates, and the sparse plan that places them. The loop (report 05 §5):

decompose_context → (per event) refine_query / search_sounds

→ gate_candidates (fail-closed license gate, FIRST) → verify_match (clap → listen → judge ladder) → decide (retrieve-vs-generate, the single branch) → place_in_timeline → a sparse SoundDesignTimeline

Design discipline (mirrors foley.index): this package is dol-only at import — the LLM rungs sit behind the Decomposer / Judge / Refiner protocols with deterministic fakes as defaults, and anthropic (the foley[agent] extra) is imported lazily inside the real impls only. So foley.find("a paragraph") works out of the box and the whole loop is exercisable in CI with no network / key / heavy dependency. Every find() opens one foley.obs run scope, so search + verify + generate aggregate into a single reproducible run-manifest.

Functions

find(context, *[, max_events, seconds, ...])	The headline: a narrative context → verified, license-clean sound candidates.
plan(candidates, *[, transcript])	Fold verified candidates into the SPARSE SoundDesignTimeline (the SELECT→WEAVE bridge).
decompose_context(context, *[, max_events, ...])	Decompose a passage into <= max_events sparse SoundEvents.
refine_query(query, *[, n, hint, refiner, _span])	Expand query into up to n paraphrases for multi-query retrieval.
search_sounds(queries, *[, k, library, filters])	Hybrid search for one query, or a multi-query RRF-merge (the SELECT retrieval tool).
verify_match(event, candidate, *[, level, ...])	Verify candidate against event up to rung level (AND-confirming ladder).
gate_candidates(candidates, intended_use)	Fail-closed license gate — run BEFORE verification/ranking (invariant #3).
decide(event, kept, verified, *, ...)	The single generate-vs-retrieve branch — a PURE function (report 05 §4).
generate_sound(query, *[, backend, library])	Generate a clip for query (the fallback tool) — a thin wrapper over foley.generate.
place_in_timeline(clip, *[, onset, gain, ...])	Emit the SPARSE per-item plan subset (clip_ref·onset·gain·layer·loop) — no more.

Classes

Decomposer(*args, **kwargs)	Narrative context → a sparse, salience-ranked, diegetic-tagged event list.
Judge(*args, **kwargs)	One rung of the verify ladder: does this candidate match this event? (report 10 §4.2).
Refiner(*args, **kwargs)	One event query → 2–4 paraphrases/expansions (query-expansion for retrieval).
Budget([max_refine_loops, max_generations, ...])	Bounded-cost accounting: per-event loop counts, and a cumulative dollar cap (#57).
Decision(action[, candidate, reason])	The tiny result of decide(); reason feeds the refine hint + the audit Step.
DecideAction(*values)	What decide() chose for one event (the single branch's outcomes).
KeywordDecomposer()	Deterministic cue-lexicon decomposer — the zero-dependency default and CI fake.
AnthropicDecomposer(*[, client, model, ...])	LLM-backed decomposer (foley[agent]): Claude → a structured event list.
KeywordRefiner()	Deterministic template-expansion refiner — the default and CI fake.
AnthropicRefiner(*[, client, model, max_tokens])	LLM-backed refiner (foley[agent]): Claude → a paraphrase list.
ClapJudge()	The zero-config clap rung: gate on the retrieval cosine (no ML, no network).
StringOverlapJudge(*[, threshold])	Deterministic stand-in for the listen/judge rungs (the hermetic CI fake).
AnthropicJudge(*[, client, model, max_tokens])	LLM arbiter for the judge rung (foley[agent]): Claude → a Verdict.
AudioLMJudge(*[, pipeline, model, ...])	The audio-LM listen rung (foley[fit]): Qwen2-Audio "does this contain {event}?".

class foley.agent.AnthropicDecomposer(*, client=None, model='claude-opus-4-8', max_tokens=2000)

Bases: object

LLM-backed decomposer (foley[agent]): Claude → a structured event list.

anthropic is imported lazily inside decompose() so import foley stays dol-only. Stashes the returned Message on self.last_response so decompose_context() can record the GenAI span (token usage / model / stop_reason). Model, thinking, and structured-output shape follow the claude-api conventions (claude-opus-4-8, adaptive thinking — never budget_tokens).

decompose(context, *, max_events=6, seconds=None)

Call Claude and round-trip each event through SoundEvent.from_dict().

Return type:

list[SoundEvent]

class foley.agent.AnthropicJudge(*, client=None, model='claude-opus-4-8', max_tokens=500)

Bases: object

LLM arbiter for the judge rung (foley[agent]): Claude → a Verdict.

anthropic imported lazily inside judge(). Stashes the Message on self.last_response for the GenAI span.

judge(event, candidate, *, level=VerifyLevel.judge)

Call Claude to arbitrate the match; returns a Verdict at level.

Return type:

Verdict

class foley.agent.AnthropicRefiner(*, client=None, model='claude-opus-4-8', max_tokens=500)

Bases: object

LLM-backed refiner (foley[agent]): Claude → a paraphrase list.

anthropic imported lazily inside refine(). Stashes the Message on self.last_response for the GenAI span.

refine(query, *, n=3, hint=None)

Call Claude for n paraphrases; the original query is always first.

Return type:

list[str]

class foley.agent.AudioLMJudge(*, pipeline=None, model='Qwen/Qwen2-Audio-7B-Instruct', max_tokens=300, tau=0.5)

Bases: object

The audio-LM listen rung (foley[fit]): Qwen2-Audio “does this contain {event}?”.

transformers/torch are imported lazily inside judge(); an injected pipeline (the test seam, mirroring the Stable-Audio fake-pipeline seam) drives the whole byte-decode → prompt → P(yes) → Verdict path with no torch. Stashes self.model + self.last_response so verify_match’s GenAI span stays informative (same getattr contract AnthropicJudge uses).

judge(event, candidate, *, level=VerifyLevel.listen)

Listen to the clip and return the AQAScore Verdict (P(yes) ≥ tau).

Return type:

Verdict

class foley.agent.Budget(max_refine_loops=1, max_generations=1, allow_generate=True, max_usd=1.0, approve_unknown_cost=None, _refines=0, _gens=0, _spent_usd=0.0, _unknown_calls=0, _lock=<factory>)

Bases: object

Bounded-cost accounting: per-event loop counts, and a cumulative dollar cap (#57).

Two kinds of bound:

• Per event — max_refine_loops / max_generations: the loop calls refine_ok() / gen_ok() to test, then spend_refine() / spend_gen() to charge; reset() zeroes them at each event.

• Per run, cumulative — max_usd (default $1): every paid call (generation, paid LLM rung) is checked with authorize() before it is made and recorded with charge() after. reset() never clears it, so the cap holds across the whole find / score run. A call whose cost is unknown (estimate None) is refused unless approve_unknown_cost (or $FOLEY_APPROVE_UNKNOWN_COST=1).

authorize(estimate_usd, *, what)

Alias of check() (kept for callers that only test affordability).

Return type:

None

charge(usd)

Record a call’s cost (None = unknown: counted, not summed).

Return type:

None

check(estimate_usd, *, what)

Raise if this run cannot afford estimate_usd (nothing is reserved).

Return type:

None

gen_ok()

Whether a generation fallback is allowed and within budget.

Return type:

bool

refine_ok()

Whether another refine→re-retrieve pass is within budget.

Return type:

bool

reserve(estimate_usd, *, what)

Check, then count estimate_usd as spent at once (atomic under a lock).

Reserving before the call is what keeps concurrent calls on one budget (MCP tools run in a threadpool) from all passing the check before any is charged. settle() later replaces the reservation with the actual cost, if known.

Return type:

None

reset()

Zero the spend counters so the caps apply per event, not per passage.

The find loop calls this at the top of each event so one hard event’s refine/generate spend never starves later events (the documented per-event semantics).

Return type:

None

settle(reserved_usd, actual_usd)

Replace a reservation with the actual cost (None actual: keep the reservation).

Return type:

None

spend_gen()

Charge one generation.

Return type:

None

spend_refine()

Charge one refine loop.

Return type:

None

property spent_usd: float

Dollars charged so far this run (calls of unknown cost count as 0 here).

class foley.agent.ClapJudge

Bases: object

The zero-config clap rung: gate on the retrieval cosine (no ML, no network).

match iff the candidate’s clap_score clears tau_clap; confidence is the score clamped to 0..1 so it is comparable to the higher rungs’ 0..1 scores.

judge(event, candidate, *, level=VerifyLevel.clap, tau_clap=0.35)

Return the clap-rung Verdict for candidate.

Return type:

Verdict

class foley.agent.DecideAction(*values)

Bases: str, Enum

What decide() chose for one event (the single branch’s outcomes).

class foley.agent.Decision(action, candidate=None, reason='')

Bases: object

The tiny result of decide(); reason feeds the refine hint + the audit Step.

class foley.agent.Decomposer(*args, **kwargs)

Bases: Protocol

Narrative context → a sparse, salience-ranked, diegetic-tagged event list.

Turns a story/commentary passage into <= max_events SoundEvents, enforcing the salience/density budget and the anachronism (era_place) guard. The default is the deterministic KeywordDecomposer; AnthropicDecomposer is the LLM-backed impl.

class foley.agent.Judge(*args, **kwargs)

Bases: Protocol

One rung of the verify ladder: does this candidate match this event? (report 10 §4.2).

level selects the rung — clap (cheap score gate), listen (audio-LM), judge (LLM arbitration + scene consistency). The returned Verdict carries level == the rung that produced it. Only the judge rung’s real impl calls the LLM.

class foley.agent.KeywordDecomposer

Bases: object

Deterministic cue-lexicon decomposer — the zero-dependency default and CI fake.

Scans the passage against _CUE_LEXICON, emits one SoundEvent per matched cue in first-appearance order (so salience descends with reading order), dedupes by canonical query, and truncates to max_events (the sparse density budget). No RNG, no network, no anthropic — same passage → identical list.

decompose(context, *, max_events=6, seconds=None)

Return <= max_events deterministic SoundEvents for context.

Parameters:

• context (str) – The narrative passage.

• max_events (int) – The sparse density cap (the salience budget).

• seconds (Optional[float]) – Accepted for signature parity (the per-second density window is a later refinement); ignored here.

Return type:

list[SoundEvent]

class foley.agent.KeywordRefiner

Bases: object

Deterministic template-expansion refiner — the default and CI fake.

Expands a query into n distinct paraphrases via fixed descriptor templates; a verify-failure hint nudges one extra variant. No RNG, no anthropic — same query → identical paraphrase list.

refine(query, *, n=3, hint=None)

Return up to n distinct paraphrases of query (the first is query).

Parameters:

• query (str) – The event query to expand.

• n (int) – How many paraphrases to return (2–4 is typical).

• hint (Optional[str]) – Optional verify-failure reason to steer re-retrieval.

Return type:

list[str]

class foley.agent.Refiner(*args, **kwargs)

Bases: Protocol

One event query → 2–4 paraphrases/expansions (query-expansion for retrieval).

hint carries the verify-failure reason so the re-retrieval loop can steer the next paraphrases. v1 feeds the list into a multi-query RRF search (embedding-fusion is a later drop-in behind this same seam). The default is the deterministic KeywordRefiner.

class foley.agent.StringOverlapJudge(*, threshold=0.3)

Bases: object

Deterministic stand-in for the listen/judge rungs (the hermetic CI fake).

Jaccard token overlap of the event query vs the candidate’s caption + tags. Lets the full ladder + decide() branch run with zero ML/network. The returned Verdict echoes back the requested level.

judge(event, candidate, *, level=VerifyLevel.listen)

Return the token-overlap Verdict at the requested level.

Return type:

Verdict

foley.agent.decide(event, kept, verified, *, tau_retrieve, budget, loop)

The single generate-vs-retrieve branch — a PURE function (report 05 §4).

Chooses among DecideAction from the already-gated (kept) and already-verified (verified) sets. It performs no I/O and never calls keep/search/generate — the foley.agent.tools loop acts on the returned Decision.

Policy (report 05 §4):

• a verified clip clearing tau_retrieve → USE (the best one);

• verified-but-low-confidence with refine budget → REFINE (feed the reason back);

• a non-diegetic cue, or a diegetic gap with no verified match, with generate budget → GENERATE;

• otherwise → DROP (silence), unless a lower-confidence verified clip exists and generation is off, in which case fall back to that best-effort pick.

Parameters:

• event (SoundEvent) – The event being resolved.

• kept (list[Candidate]) – The license-clean candidates (each license_ok is True).

• verified (list[Candidate]) – The subset of kept whose verdict matched.

• tau_retrieve (float) – The confidence threshold for auto-accepting a retrieved clip.

• budget (Budget) – The per-event cost budget.

• loop (int) – The current refine-loop index (for the audit reason).

Return type:

Decision

Returns:

A Decision.

foley.agent.decompose_context(context, *, max_events=6, seconds=None, decomposer=None, _span=None)

Decompose a passage into <= max_events sparse SoundEvents.

The pure SELECT tool (Python-API == agent == future-MCP surface): resolves the default decomposer when decomposer is None, calls it, and records the GenAI span on the real path (the fake’s last_response is None → no-op).

Parameters:

• context (str) – The narrative passage.

• max_events (int) – The sparse density cap.

• seconds (Optional[float]) – Optional passage duration (density-window hint; forwarded, else ignored).

• decomposer (Optional[Decomposer]) – An injected Decomposer (the DI seam); defaults to _default_decomposer().

• _span – Internal — the obs span handle find() opens for GenAI recording.

Return type:

list[SoundEvent]

foley.agent.find(context, *, max_events=6, seconds=None, intended_use=None, backend='auto', verify='listen', stream=False, k=10, tau_retrieve=0.5, tau_clap=0.35, max_refine_loops=1, budget=None, library=None, decomposer=None, judge=None, refiner=None, llm=None)

The headline: a narrative context → verified, license-clean sound candidates.

decompose → (per event) refine/search → verify_match ladder → decide (with the fail-closed license gate FIRST) → place (report 05 §5). Works out of the box — foley.find("She pushed open the heavy oak door; rain hammered outside.") — with deterministic defaults; every model / threshold / seam is an optional keyword.

Parameters:

• context (str) – The narrative passage.

• max_events (int) – The sparse density cap on decomposed events.

• seconds (Optional[float]) – Optional passage duration (density-window hint; forwarded).

• intended_use (Optional[IntendedUse]) – The caller’s rights intent (default: foley.licensing.DEFAULT_INTENDED_USE — commercial publishing; NC / SA material is refused unless you pass a different intent).

• backend (str) – Generation backend for the fallback ('auto' → foley.generate’s default).

• verify (Union[str, VerifyLevel]) – The max verify rung — 'clap' | 'listen' | 'judge'.

• stream (bool) – If True, return a generator yielding one Candidate per resolved event; else return the collected list.

• k (int) – Retrieval shortlist depth per query.

• tau_retrieve (float) – Confidence threshold to auto-accept a retrieved clip.

• tau_clap (float) – The clap-rung gate threshold.

• max_refine_loops (int) – Max refine→re-retrieve passes per event (also the default Budget).

• budget (Optional[Budget]) – An explicit Budget (overrides max_refine_loops). It is also the run’s spend cap: Budget(max_usd=...) (default $1; $3 when llm='anthropic' or $FOLEY_LLM=anthropic, whose default run is bounded at about $2.52 — foley.cost.DEFAULT_MAX_USD_BY_LLM) bounds every paid call the run makes — generations and paid LLM rungs — cumulatively, and a call of unknown cost needs Budget(approve_unknown_cost=True) (#57).

• library – Target SoundLibrary (default: the process-wide default).

• refiner (decomposer / judge /) – Injected DI seams (Decomposer / Judge / Refiner); each defaults to the llm provider’s implementation.

• llm (Optional[str]) – Which LLM the decompose / refine / judge rungs use when not injected — 'fake' | 'local' | 'anthropic'. None reads $FOLEY_LLM, then falls back to the free default (a configured local endpoint, else the deterministic fake). A key being present never opts in to paid calls.

Return type:

Union[list[Candidate], Iterator[Candidate]]

Returns:

list[Candidate] (stream=False) or an Iterator[Candidate] (stream=True) — one verified, license-clean candidate per resolved event.

foley.agent.gate_candidates(candidates, intended_use)

Fail-closed license gate — run BEFORE verification/ranking (invariant #3).

For each candidate: apply keep_sound() (any exception ⇒ False, fail-closed), record the result on candidate.license_ok, and keep ONLY where it is True — a None (never gated) or False is dropped. This is the single rights-rejection point; verify_match() asserts its survivors are license-clean.

Parameters:

• candidates (list[Candidate]) – The retrieved shortlist (license_ok typically None).

• intended_use (IntendedUse) – The caller’s declared rights intent.

Return type:

list[Candidate]

Returns:

The license-clean sublist (each with license_ok is True), order preserved.

foley.agent.generate_sound(query, *, backend='auto', library=None)

Generate a clip for query (the fallback tool) — a thin wrapper over foley.generate.

Returns a Candidate (origin=generated); raises the GenerationError hierarchy on refusal / no-result (caught fail-closed by _generate_and_reverify()).

Return type:

Candidate

foley.agent.place_in_timeline(clip, *, onset=None, gain=0.0, layer=Layer.sfx_fg, loop=False)

Emit the SPARSE per-item plan subset (clip_ref·onset·gain·layer·loop) — no more.

onset stays a SYMBOLIC anchor string (never a resolved offset); mix / processing / alignment / mastering are WEAVE’s (#8) job, not SELECT’s.

Return type:

TimelineItem

foley.agent.plan(candidates, *, transcript=None)

Fold verified candidates into the SPARSE SoundDesignTimeline (the SELECT→WEAVE bridge).

One TimelineItem per candidate (onset·gain·layer·loop only, from its SoundEvent), joined to the run-artifact via run_manifest_ref — the reserved #8 plan_ref slot is filled when called inside an active foley.obs run scope (None-safe otherwise).

Parameters:

• candidates (list[Candidate]) – The candidates returned by find().

• transcript (Optional[str]) – Optional narration transcript (WEAVE resolves the reference).

Return type:

SoundDesignTimeline

foley.agent.refine_query(query, *, n=3, hint=None, refiner=None, _span=None)

Expand query into up to n paraphrases for multi-query retrieval.

Parameters:

• query (str) – The event query to expand.

• n (int) – Number of paraphrases.

• hint (Optional[str]) – Optional verify-failure reason to steer re-retrieval.

• refiner (Optional[Refiner]) – An injected Refiner (the DI seam); defaults to _default_refiner().

• _span – Internal — the obs span handle for GenAI recording.

Return type:

list[str]

foley.agent.search_sounds(queries, *, k=10, library=None, filters=None)

Hybrid search for one query, or a multi-query RRF-merge (the SELECT retrieval tool).

Delegates a single query VERBATIM to SoundLibrary.search() (so the retrieval ranking the nDCG@10 gate measures is untouched); a multi-query list runs each query and RRF-merges (k=RRF_K) — the query-expansion recall lever, exercised only on a refine pass.

Parameters:

• queries (Union[str, list[str]]) – One query string, or a list of paraphrases to merge.

• k (int) – Number of results.

• library – Target library (default: the process-wide default).

• filters (Optional[dict]) – Extra SoundLibrary.search() kwargs (e.g. the license prefilter).

Return type:

list[Candidate]

foley.agent.verify_match(event, candidate, *, level=VerifyLevel.clap, judge=None, tau_clap=0.35, _span=None)

Verify candidate against event up to rung level (AND-confirming ladder).

Runs the clap gate always; if level is higher and the clap gate passed, escalates to the injected/​default judge for that rung and returns its verdict (Verdict.level == the producing rung).

Parameters:

• event (SoundEvent) – The wanted SoundEvent.

• candidate (Candidate) – A license-clean Candidate — this MUST run after the gate_candidates() gate (asserted).

• level (str | VerifyLevel) – The max rung to climb (clap | listen | judge).

• judge (Optional[Judge]) – An injected Judge for the higher rungs (the DI seam; defaults per _default_judge()).

• tau_clap (float) – The clap-gate threshold.

• _span – Internal — the obs span handle for GenAI recording on the LLM rung.

Raises:

AssertionError – If candidate.license_ok is not True (verify-before-gate is a bug — the license gate is the fail-closed first pass).

Return type:

Verdict

Modules

decompose	decompose_context — narrative passage → a sparse, salience-ranked event list.
llm	Which LLM the SELECT rungs use — one explicit, spend-safe resolver (the llm= seam).
local_llm	Offline / local-LLM SELECT rungs — OpenAI-compatible Decomposer / Judge / Refiner.
mcp	The MCP surface — foley's façade as agent-callable tools (py2mcp, #12, report 05/10).
policy	The SELECT policy: the fail-closed rights gate + the single generate-vs-retrieve branch.
preview	Audition UX — preview a sound, find similar ones, and refine by relevance feedback (#12).
protocols	The structural DI seams of the SELECT stage — Decomposer / Judge / Refiner.
refine	refine_query — one event query → 2–4 paraphrases for multi-query retrieval.
session	Session-scoped audition state — cached candidates, picks, and rejects (#12).
tools	The SELECT orchestration: find() / plan() + the small pure tool wrappers.
verify	verify_match — the retrieve→verify ladder (clap → listen → judge) + Judge impls.