Skip to content

Search

search

Typed search spaces and the deterministic random strategy.

Parameter

Bases: Protocol

Sample one typed parameter value from a random stream.

sample 

sample(generator: Random) -> Any
Source code in src/tunarag/search.py 
17
def sample(self, generator: random.Random) -> Any: ...

accepts 

accepts(value: Any) -> bool
Source code in src/tunarag/search.py 
19
def accepts(self, value: Any) -> bool: ...

describe 

describe() -> Mapping[str, Any]
Source code in src/tunarag/search.py 
21
def describe(self) -> Mapping[str, Any]: ...

IntegerRange dataclass 

IntegerRange(minimum: int, maximum: int)

Inclusive integer range.

minimum instance-attribute 

minimum: int

maximum instance-attribute 

maximum: int

sample 

sample(generator: Random) -> int
Source code in src/tunarag/search.py 
40
41
def sample(self, generator: random.Random) -> int:
    return generator.randint(self.minimum, self.maximum)

accepts 

accepts(value: Any) -> bool
Source code in src/tunarag/search.py 
43
44
45
46
47
48
def accepts(self, value: Any) -> bool:
    return (
        isinstance(value, int)
        and not isinstance(value, bool)
        and self.minimum <= value <= self.maximum
    )

describe 

describe() -> Mapping[str, Any]
Source code in src/tunarag/search.py 
50
51
def describe(self) -> Mapping[str, Any]:
    return {"type": "integer", "minimum": self.minimum, "maximum": self.maximum}

FloatRange dataclass 

FloatRange(minimum: float, maximum: float)

Continuous float range.

minimum instance-attribute 

minimum: float

maximum instance-attribute 

maximum: float

sample 

sample(generator: Random) -> float
Source code in src/tunarag/search.py 
72
73
def sample(self, generator: random.Random) -> float:
    return generator.uniform(self.minimum, self.maximum)

accepts 

accepts(value: Any) -> bool
Source code in src/tunarag/search.py 
75
76
77
78
79
80
def accepts(self, value: Any) -> bool:
    return (
        isinstance(value, (int, float))
        and not isinstance(value, bool)
        and self.minimum <= value <= self.maximum
    )

describe 

describe() -> Mapping[str, Any]
Source code in src/tunarag/search.py 
82
83
def describe(self) -> Mapping[str, Any]:
    return {"type": "number", "minimum": self.minimum, "maximum": self.maximum}

Categorical dataclass 

Categorical(values: tuple[Any, ...])

A non-empty collection of candidate values.

values instance-attribute 

values: tuple[Any, ...]

sample 

sample(generator: Random) -> Any
Source code in src/tunarag/search.py 
96
97
def sample(self, generator: random.Random) -> Any:
    return generator.choice(self.values)

accepts 

accepts(value: Any) -> bool
Source code in src/tunarag/search.py 
 99
100
def accepts(self, value: Any) -> bool:
    return value in self.values

describe 

describe() -> Mapping[str, Any]
Source code in src/tunarag/search.py 
102
103
104
105
106
def describe(self) -> Mapping[str, Any]:
    return {
        "type": "categorical",
        "values": tuple(value for value in self.values if not isinstance(value, Secret)),
    }

SearchSpace dataclass 

SearchSpace(parameters: dict[str, Parameter])

Named typed parameters from which candidates can be sampled.

parameters instance-attribute 

parameters: dict[str, Parameter]

sample 

sample(generator: Random) -> Candidate
Source code in src/tunarag/search.py 
122
123
124
125
def sample(self, generator: random.Random) -> Candidate:
    return Candidate(
        {name: parameter.sample(generator) for name, parameter in self.parameters.items()}
    )

validate 

validate(parameters: Mapping[str, Any]) -> Candidate

Validate an exact candidate mapping against every parameter.

Source code in src/tunarag/search.py 
127
128
129
130
131
132
133
134
135
def validate(self, parameters: Mapping[str, Any]) -> Candidate:
    """Validate an exact candidate mapping against every parameter."""

    if set(parameters) != set(self.parameters):
        raise ValueError("candidate keys must exactly match the search space")
    for name, parameter in self.parameters.items():
        if not parameter.accepts(parameters[name]):
            raise ValueError(f"candidate value is outside the search space: {name}")
    return Candidate(dict(parameters))

describe 

describe() -> Mapping[str, Mapping[str, Any]]

Return a provider-safe structured search-space description.

Source code in src/tunarag/search.py 
137
138
139
140
def describe(self) -> Mapping[str, Mapping[str, Any]]:
    """Return a provider-safe structured search-space description."""

    return {name: parameter.describe() for name, parameter in self.parameters.items()}

RandomSearch 

RandomSearch(space: SearchSpace, *, seed: int | None = None)

Seeded random strategy implementing the common search contract.

Source code in src/tunarag/search.py 
146
147
148
149
150
def __init__(self, space: SearchSpace, *, seed: int | None = None) -> None:
    self._space = space
    self._seed = seed
    self._generator = random.Random(seed)
    self._replayed_candidates: list[Candidate] = []

suggest async 

suggest() -> Candidate
Source code in src/tunarag/search.py 
152
153
async def suggest(self) -> Candidate:
    return self._space.sample(self._generator)

observe async 

observe(candidate: Candidate, metrics: Any, usage: Any) -> None
Source code in src/tunarag/search.py 
155
156
async def observe(self, candidate: Candidate, metrics: Any, usage: Any) -> None:
    del candidate, metrics, usage

replay async 

replay(candidate: Candidate, metrics: Any, usage: Any) -> None

Advance and verify deterministic generator state during resume.

Source code in src/tunarag/search.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
async def replay(self, candidate: Candidate, metrics: Any, usage: Any) -> None:
    """Advance and verify deterministic generator state during resume."""

    del metrics, usage
    if candidate in self._replayed_candidates:
        raise ValueError("persisted candidate does not match seeded random search state")
    for _ in range(10_000):
        expected = self._space.sample(self._generator)
        if expected == candidate:
            self._replayed_candidates.append(candidate)
            return
        if expected not in self._replayed_candidates:
            break
    raise ValueError("persisted candidate does not match seeded random search state")

fingerprint 

fingerprint() -> Any

Return stable search-space and seed semantics.

Source code in src/tunarag/search.py 
173
174
175
176
def fingerprint(self) -> Any:
    """Return stable search-space and seed semantics."""

    return {"space": self._space, "seed": self._seed}

SearchObservation dataclass 

SearchObservation(candidate: Candidate, metrics: tuple[MetricValue, ...])

A prior reserved candidate and any committed metrics.

candidate instance-attribute 

candidate: Candidate

metrics instance-attribute 

metrics: tuple[MetricValue, ...]

LLMSuggestionRequest dataclass 

LLMSuggestionRequest(search_space: Mapping[str, Mapping[str, Any]], observations: tuple[SearchObservation, ...], suggestion_number: int, attempt_number: int)

Provider-neutral structured input for one suggestion attempt.

search_space instance-attribute 

search_space: Mapping[str, Mapping[str, Any]]

observations instance-attribute 

observations: tuple[SearchObservation, ...]

suggestion_number instance-attribute 

suggestion_number: int

attempt_number instance-attribute 

attempt_number: int

LLMSearchProvider

Bases: Protocol

Produce one structured candidate proposal without provider coupling.

suggest async 

suggest(request: LLMSuggestionRequest) -> Mapping[str, Any]
Source code in src/tunarag/search.py 
200
async def suggest(self, request: LLMSuggestionRequest) -> Mapping[str, Any]: ...

LLMSearch 

LLMSearch(space: SearchSpace, provider: LLMSearchProvider, *, max_suggestion_attempts: int = 3, fallback_seed: int = 0, fallback: bool = True)

Validate bounded provider suggestions with deterministic random fallback.

Source code in src/tunarag/search.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def __init__(
    self,
    space: SearchSpace,
    provider: LLMSearchProvider,
    *,
    max_suggestion_attempts: int = 3,
    fallback_seed: int = 0,
    fallback: bool = True,
) -> None:
    if (
        not isinstance(max_suggestion_attempts, int)
        or isinstance(max_suggestion_attempts, bool)
        or max_suggestion_attempts < 1
    ):
        raise ValueError("maximum suggestion attempts must be positive")
    self._space = space
    self._provider = provider
    self._max_suggestion_attempts = max_suggestion_attempts
    self._fallback_seed = fallback_seed
    self._fallback = fallback
    self._suggestion_number = 0
    self._observations: list[SearchObservation] = []

suggest async 

suggest() -> Candidate
Source code in src/tunarag/search.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
async def suggest(self) -> Candidate:
    causes: list[str] = []
    for attempt_number in range(self._max_suggestion_attempts):
        request = LLMSuggestionRequest(
            self._space.describe(),
            tuple(_provider_observation(item) for item in self._observations),
            self._suggestion_number,
            attempt_number,
        )
        try:
            proposal = await self._provider.suggest(request)
            candidate = self._space.validate(proposal)
        except Exception as error:
            causes.append(type(error).__name__)
            continue
        self._suggestion_number += 1
        return candidate
    if self._fallback:
        generator = random.Random(f"{self._fallback_seed}:{self._suggestion_number}")
        candidate = self._space.sample(generator)
        self._suggestion_number += 1
        return candidate
    raise SearchSpaceError(
        "LLM search exhausted invalid suggestion attempts",
        stage="strategy.suggest",
        details={"causes": causes, "attempts": self._max_suggestion_attempts},
    )

observe async 

observe(candidate: Candidate, metrics: Sequence[MetricValue], usage: Sequence[UsageRecord]) -> None
Source code in src/tunarag/search.py 
257
258
259
260
261
262
263
264
async def observe(
    self,
    candidate: Candidate,
    metrics: Sequence[MetricValue],
    usage: Sequence[UsageRecord],
) -> None:
    del usage
    self._observations.append(SearchObservation(candidate, tuple(metrics)))

replay async 

replay(candidate: Candidate, metrics: Sequence[MetricValue], usage: Sequence[UsageRecord]) -> None
Source code in src/tunarag/search.py 
266
267
268
269
270
271
272
273
274
async def replay(
    self,
    candidate: Candidate,
    metrics: Sequence[MetricValue],
    usage: Sequence[UsageRecord],
) -> None:
    del usage
    self._suggestion_number += 1
    self._observations.append(SearchObservation(candidate, tuple(metrics)))

fingerprint 

fingerprint() -> Any
Source code in src/tunarag/search.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def fingerprint(self) -> Any:
    provider_type = type(self._provider)
    provider: dict[str, Any] = {
        "type": f"{provider_type.__module__}.{provider_type.__qualname__}"
    }
    if isinstance(self._provider, Fingerprintable):
        provider["fingerprint"] = self._provider.fingerprint()
    return {
        "space": self._space,
        "provider": provider,
        "max_suggestion_attempts": self._max_suggestion_attempts,
        "fallback_seed": self._fallback_seed,
        "fallback": self._fallback,
    }