pi_llm.utils

utils

pi_llm.utils — Streaming infrastructure and utility functions.

AssistantMessageEventStream()

Bases: EventStream[AssistantMessageEvent, AssistantMessage]

Event stream specialized for assistant message streaming.

EventStream(is_complete, extract_result)

Bases: Generic[T, R]

Push-based async iterable stream with a final result.

The producer calls push(event) to emit events and end() to close the stream. The consumer uses async for event in stream to iterate over events and await stream.result() to get the final result.

Parameters:

Name	Type	Description	Default
is_complete	Callable[[T], bool]	Predicate that returns True for the terminal event.	required
extract_result	Callable[[T], R]	Extracts the final result from the terminal event.	required

push(event)

Push an event into the stream. Silently drops events after end().

end(result=None)

Signal end of stream. Sets result if not already set.

__aiter__() async

Async iterate over events until sentinel.

result() async

Await the final result. Resolves on completion event or end(result).

create_assistant_message_event_stream()

Factory function for AssistantMessageEventStream.

short_hash(value)

Generate a short deterministic hash (12-char hex).

parse_streaming_json(partial)

Parse potentially incomplete JSON from streaming tool call arguments.

Always returns a valid object — never raises.

is_context_overflow(message, context_window=None)

Check if an assistant message represents a context overflow error.

Handles two cases: 1. Error-based overflow: Most providers return stop_reason "error" with a specific error message pattern. 2. Silent overflow: Some providers (e.g. z.ai) accept overflow silently. For these, check if usage.input exceeds the context window.

sanitize_surrogates(text)

Replace lone Unicode surrogates with the replacement character (U+FFFD).

Valid emoji and paired surrogates are preserved. Lone surrogates (which break JSON serialization in many API providers) are replaced.

validate_tool_arguments(tool, tool_call)

Validate tool call arguments against JSON Schema.

Returns a deep copy of validated arguments. Raises ValueError with formatted message on validation failure.

validate_tool_call(tools, tool_call)

Find a tool by name and validate the call's arguments.

Parameters:

Name	Type	Description	Default
tools	list[Tool]	Available tools to search.	required
tool_call	ToolCall	The tool call to validate.	required

Returns:

Type	Description
Any	A deep copy of the validated arguments.

Raises:

Type	Description
ValueError	If the tool is not found or arguments are invalid.