utils

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

add_monitor_hook(hook)

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

call_llm(messages, *, temperature=None, max_tokens=None, response_format=None, timeout=None, tools=None, api_key=None, thinking=None, reasoning_effort=None)

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), configurable temperature/max_tokens, and thinking-mode overrides.

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. Ignored by the API when thinking mode is enabled.	None
max_tokens	int | None	Maximum tokens in response. Defaults to LLM_MAX_TOKENS. In thinking mode this budget covers reasoning + content.	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
thinking	bool | None	Override LLM_THINKING_ENABLED for this call.	None
reasoning_effort	str | None	Override LLM_REASONING_EFFORT ("low"/"high"/"max").	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_stream(messages, *, temperature=None, max_tokens=None, timeout=None, tools=None, api_key=None, thinking=None, reasoning_effort=None, http_client=None) async

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, StatusEvent when reasoning (chain-of-thought) deltas arrive, ToolCallEvent when finish_reason='tool_calls' is detected, and TruncatedEvent when finish_reason='length' cuts the reply short.

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). In thinking mode this budget covers reasoning + content.	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
thinking	bool | None	Override LLM_THINKING_ENABLED for this call.	None
reasoning_effort	str | None	Override LLM_REASONING_EFFORT ("low"/"high"/"max").	None
http_client	'httpx.AsyncClient | None'	Optional shared httpx.AsyncClient for connection reuse. When provided, avoids TLS handshake per call.	None

Yields:

Type	Description
AsyncGenerator	TokenEvent for each content delta, StatusEvent for reasoning
AsyncGenerator	liveness, ToolCallEvent for tool call finish reasons, and
AsyncGenerator	TruncatedEvent when finish_reason='length' (max_tokens cut
AsyncGenerator	the reply short).

Raises:

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

call_llm_text(messages, **kwargs)

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.

count_messages_tokens(messages)

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.

count_tokens(text)

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.

rag_retrieve(embedder, kb, query, top_k=TOP_K)

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(embedder, kb, query, top_k=TOP_K)

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(text, context='')

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.

sanitize_text(text)

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.

set_verbose(enabled)

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

slugify(text, max_len=40)

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_-].