ahvn.utils.exts.examples_utils module

ahvn.utils.exts.examples_utils.normalize_examples(examples, search_args=None)

Normalize examples input to a generator of CacheEntry.

Supported input types:

1. None: Returns an empty generator

2. Iterable[Union[Dict, CacheEntry, ExperienceUKFT]]:

   • Dict: Convert using CacheEntry.from_dict()

   • CacheEntry: Return directly

   • ExperienceUKFT: Convert using to_cache_entry() method

3. BaseKLStore: Use ScanKLEngine to iterate all entries, convert ExperienceUKFT to CacheEntry

4. BaseKLEngine: Use search() method to get results, extract ‘kl’ field, convert to CacheEntry if it’s ExperienceUKFT

5. KLBase: Use search() method to get results, extract ‘kl’ field, convert to CacheEntry if it’s ExperienceUKFT

Parameters:

• examples (Union[Iterable[Union[Dict[str, Any], CacheEntry, ExperienceUKFT]], BaseKLStore, BaseKLEngine, KLBase, None]) – Input examples, can be various types including: None, Iterable, KLStore, KLEngine, and KLBase

• search_args (Optional[Dict[str, Any]]) – Arguments passed to KLEngine.search() or KLBase.search() (only used when examples is a KLEngine or KLBase)

Returns:

A generator of CacheEntry, all entries will become Few-shot examples

Return type:

Generator[CacheEntry, None, None]

Examples

>>> # List of dictionaries
>>> examples = [{"inputs": {"x": 1}, "output": 2}]
>>> list(normalize_examples(examples))
[CacheEntry(...)]

>>> # List of ExperienceUKFT
>>> exp = ExperienceUKFT.from_cache_entry(CacheEntry.from_args(x=1, output=2))
>>> list(normalize_examples([exp]))
[CacheEntry(...)]

>>> # KLStore
>>> from ahvn.klstore import CacheKLStore
>>> store = CacheKLStore()
>>> # ... add some ExperienceUKFT entries
>>> list(normalize_examples(store))
[CacheEntry(...), ...]

>>> # KLEngine
>>> from ahvn.klengine import ScanKLEngine
>>> engine = ScanKLEngine(storage=store)
>>> list(normalize_examples(engine, search_args={"topk": 5, "type": "experience"}))
[CacheEntry(...), ...]

>>> # KLBase
>>> from ahvn.klbase import KLBase
>>> klbase = KLBase(engines={"engine": engine})
>>> klbase.set_default_engine("engine")
>>> list(normalize_examples(klbase, search_args={"topk": 5, "type": "experience"}))
[CacheEntry(...), ...]