starbreeze_local_rag

Validated offline retrieval for the user-supplied STARBREEZE corpus.

The bundled corpus is data, not an instruction source. This module imports no code from the supplied archive: it validates a small allowlist of immutable assets, recreates the documented lexical/LSA query transform, and returns only bounded evidence records with their source citations.

exception starbreeze_local_rag.StarbreezeIndexValidationError

Bases: ValueError

Raised when a bundled retrieval asset fails its immutable contract.

starbreeze_local_rag.channel_preference_key(platform, channel_id)

Return the independent per-channel STARBREEZE preference key.

Return type:

str

Parameters:

  • platform (str)

  • channel_id (str)

async starbreeze_local_rag.is_channel_enabled(redis_client, *, platform, channel_id)

Read the channel preference; an absent key means default-off.

Return type:

bool

Parameters:

  • redis_client (Any)

  • platform (str)

  • channel_id (str)

async starbreeze_local_rag.set_channel_enabled(redis_client, *, platform, channel_id, enabled)

Persist only channel opt-ins so the default remains disabled.

Return type:

None

Parameters:

  • redis_client (Any)

  • platform (str)

  • channel_id (str)

  • enabled (bool)

starbreeze_local_rag.system_injection()

Return the trusted contract for standalone STARBREEZE evidence.

Return type:

str

class starbreeze_local_rag.StarbreezeIndex(records, vectors, idf, components, vocabulary, token_counts, lengths, average_length, postings, aliases, symbols)

Bases: object

Parameters:
records: tuple[dict[str, Any], ...]
vectors: ndarray
idf: ndarray
components: ndarray
vocabulary: dict[str, int]
token_counts: tuple[Counter[str], ...]
lengths: tuple[int, ...]
average_length: float
postings: dict[str, tuple[tuple[int, int], ...]]
aliases: tuple[tuple[str, ...], ...]
symbols: tuple[tuple[str, ...], ...]
starbreeze_local_rag.load_starbreeze_index(root=PosixPath('/home/ubuntu/stargazer-v3/stargazer-v3/rag_assets/starbreeze_rag'))

Load and cache the immutable STARBREEZE LSA index after validation.

Return type:

StarbreezeIndex

Parameters:

root (Path)

starbreeze_local_rag.search_starbreeze(query, *, top_k=5, root=PosixPath('/home/ubuntu/stargazer-v3/stargazer-v3/rag_assets/starbreeze_rag'))

Return bounded hybrid matches from the immutable local corpus.

Return type:

list[dict[str, Any]]

Parameters:
starbreeze_local_rag.format_starbreeze_context(results, *, max_chars=7000)

Serialize whole evidence records inside an explicit untrusted boundary.

Return type:

str

Parameters:
starbreeze_local_rag.starbreeze_result_was_mentioned(result, recent_assistant_context)

Return whether an assistant turn explicitly named this result’s topic.

Return type:

bool

Parameters:
async starbreeze_local_rag.suppress_recent_starbreeze_results(results, *, redis_client, platform, channel_id, recent_assistant_context, cooldown_seconds=86400)

Drop recently used topics and remember new mentions per channel.

Return type:

list[dict[str, Any]]

Parameters:
async starbreeze_local_rag.aretrieve_starbreeze_context(query, *, top_k=5, max_chars=7000, root=PosixPath('/home/ubuntu/stargazer-v3/stargazer-v3/rag_assets/starbreeze_rag'), redis_client=None, platform='', channel_id='', recent_assistant_context='')

Run validated hybrid retrieval off the event loop and format evidence.

Return type:

str

Parameters:

  • query (str)

  • top_k (int)

  • max_chars (int)

  • root (Path)

  • redis_client (Any | None)

  • platform (str)

  • channel_id (str)

  • recent_assistant_context (str)