utils

utils

Shared utilities — LLM calls, RAG retrieval, slugification, error handling.

set_verbose

set_verbose(enabled: bool) -> None

Enable or disable verbose logging for LLM requests/responses.

When enabled, prints request body summaries and response metadata to stdout via _print_llm_request() and _print_llm_response().

Parameters:

Name	Type	Description	Default
enabled	bool	True to enable verbose output, False to disable.	required

add_monitor_hook

add_monitor_hook(hook) -> None

Register a monitoring callback for LLM calls.

The hook is called after every successful LLM API response with (request_body, response_data) arguments. Useful for observability, logging, or analytics integration. Exceptions raised by hooks are silently caught to avoid disrupting the main flow.

Parameters:

Name	Type	Description	Default
hook		Callable accepting (request_body: dict, response_data: dict).	required

count_tokens

count_tokens(text: str) -> int

Count tokens in a text string using tiktoken.

Falls back to a rough estimate (chars / 4) if tiktoken is unavailable or encoding fails.

Parameters:

Name	Type	Description	Default
text	str	The text string to count tokens for.	required

Returns:

Type	Description
int	Estimated token count as an integer.

count_messages_tokens

count_messages_tokens(messages: list[dict]) -> int

Count total tokens across a list of chat messages.

Sums token counts for message content strings and for tool call function arguments and names.

Parameters:

Name	Type	Description	Default
messages	list[dict]	List of message dicts with "content" and optional "tool_calls" keys (OpenAI-compatible format).	required

Returns:

Type	Description
int	Total estimated token count for all messages combined.

slugify

slugify(text: str, max_len: int = 40) -> str

Create a URL-safe slug from arbitrary text.

Lowercases the text, replaces non-alphanumeric characters with underscores, collapses consecutive underscores, and strips leading/trailing underscores.

Parameters:

Name	Type	Description	Default
text	str	Input text to slugify.	required
max_len	int	Maximum length of the output slug. Defaults to 40.	40

Returns:

Type	Description
str	URL-safe slug string containing only [a-z0-9_-].

sanitize_text

sanitize_text(text: str) -> str

Sanitize text by encoding to UTF-8 and decoding back.

This is a safety measure for text that may contain characters incompatible with Unicode. Uses "replace" error handling to substitute problematic bytes.

Parameters:

Name	Type	Description	Default
text	str	Input string that may contain non-UTF-8 data.	required

Returns:

Type	Description
str	Clean UTF-8 string with any invalid bytes replaced.

call_llm

call_llm(
    messages: list[dict],
    *,
    temperature: float | None = None,
    max_tokens: int | None = None,
    response_format: dict | None = None,
    timeout: int | None = None,
    tools: list | None = None,
    api_key: str | None = None,
) -> dict

Synchronous LLM API call to DeepSeek.

Sends a chat completion request and returns the full JSON response. Supports optional tool definitions, response format constraints (e.g., json_object), and configurable temperature/max_tokens.

Parameters:

Name	Type	Description	Default
messages	list[dict]	List of message dicts in OpenAI-compatible format.	required
temperature	float | None	Sampling temperature. Defaults to LLM_TEMPERATURE.	None
max_tokens	int | None	Maximum tokens in response. Defaults to LLM_MAX_TOKENS.	None
response_format	dict | None	Optional format constraint dict (e.g., {"type": "json_object"}).	None
timeout	int | None	Request timeout in seconds. Defaults to LLM_TIMEOUT.	None
tools	list | None	Optional list of tool/function definitions.	None
api_key	str | None	Optional per-user API key. Falls back to global DEEPSEEK_API_KEY.	None

Returns:

Type	Description
dict	Full JSON response dict from the LLM API.

Raises:

Type	Description
RuntimeError	On timeout, connection error, HTTP error, or empty/invalid response from the API.

call_llm_text

call_llm_text(messages: list[dict], **kwargs) -> str

Convenience wrapper for call_llm() that extracts only the text content.

Parameters:

Name	Type	Description	Default
messages	list[dict]	List of message dicts in OpenAI-compatible format.	required
**kwargs		Passed through to call_llm() (temperature, max_tokens, etc.).	{}

Returns:

Type	Description
str	The text content from the first choice's message, or empty string
str	if no content was returned.

Raises:

Type	Description
RuntimeError	Propagated from call_llm() on API errors.

rag_retrieve

rag_retrieve(
    embedder, kb, query: str, top_k: int = TOP_K
) -> list

Retrieve top-k relevant chunks from the knowledge base.

Delegates to KnowledgeBase.query() which handles embedding, cosine similarity, and the Python 3.15 JIT pre-allocation workaround.

Parameters:

Name	Type	Description	Default
embedder		Fitted Embedder instance (unused — kept for API compat).	required
kb		KnowledgeBase instance with loaded vectors and chunks.	required
query	str	The search query string.	required
top_k	int	Number of results to return. Defaults to TOP_K config value.	TOP_K

Returns:

Type	Description
list	List of ChunkResult objects sorted by descending similarity score.

rag_retrieve_md

rag_retrieve_md(
    embedder, kb, query: str, top_k: int = TOP_K
) -> str

Retrieve top-k chunks and format them as markdown.

Combines rag_retrieve() with markdown formatting for use in prompts or reports. Each chunk is rendered as a markdown section with source and score. Content is truncated to 500 characters.

Parameters:

Name	Type	Description	Default
embedder		Fitted Embedder instance.	required
kb		KnowledgeBase instance with loaded vectors and chunks.	required
query	str	The search query string.	required
top_k	int	Number of results to return. Defaults to TOP_K.	TOP_K

Returns:

Type	Description
str	Markdown-formatted string with retrieved chunk sections.

safe_json_loads

safe_json_loads(text: str, context: str = '') -> dict

Parse JSON with a descriptive error message on failure.

Wraps json.loads() and converts ValueError/TypeError into RuntimeError with context information for debugging tool call argument parsing.

Parameters:

Name	Type	Description	Default
text	str	JSON string to parse.	required
context	str	Optional label for error messages (e.g., tool name).	''

Returns:

Type	Description
dict	Parsed dictionary from the JSON string.

Raises:

Type	Description
RuntimeError	If the text is not valid JSON, with context and truncated argument text in the error message.

call_llm_stream async

call_llm_stream(
    messages: list[dict],
    *,
    temperature: float | None = None,
    max_tokens: int | None = None,
    timeout: int | None = None,
    tools: list | None = None,
    api_key: str | None = None,
) -> AsyncGenerator

Stream tokens from the DeepSeek API via httpx.AsyncClient.stream().

Sends a chat completion request with streaming enabled. Parses SSE chunks from the response, yielding TokenEvent for each content delta and ToolCallEvent when finish_reason='tool_calls' is detected.

Parameters:

Name	Type	Description	Default
messages	list[dict]	List of message dicts (role + content) forming the conversation context.	required
temperature	float | None	Sampling temperature (defaults to LLM_TEMPERATURE).	None
max_tokens	int | None	Max completion tokens (defaults to LLM_MAX_TOKENS).	None
timeout	int | None	Request timeout in seconds (defaults to LLM_TIMEOUT).	None
tools	list | None	Optional list of tool schema dicts for function calling.	None
api_key	str | None	Optional per-user API key override.	None

Yields:

Type	Description
AsyncGenerator	TokenEvent for each content delta, ToolCallEvent for tool call
AsyncGenerator	finish reasons.

Raises:

Type	Description
RuntimeError	On timeout, connection error, or HTTP error from the LLM API.