> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/kyutai-labs/unmute/llms.txt
> Use this file to discover all available pages before exploring further.
# LLM Utils
> Utilities for streaming LLM completions and preprocessing conversation history
## Overview
The `llm_utils` module provides utilities for interacting with Large Language Models via the OpenAI-compatible API, preprocessing conversation history, and streaming text responses.
## Constants
### INTERRUPTION\_CHAR
```python theme={null}
INTERRUPTION_CHAR = "—" # em-dash
```
Character appended to assistant messages when the bot is interrupted by the user.
### USER\_SILENCE\_MARKER
```python theme={null}
USER_SILENCE_MARKER = "..."
```
Marker inserted into user messages when they remain silent for an extended period.
## Classes
### VLLMStream
```python theme={null}
class VLLMStream:
def __init__(
self,
client: AsyncOpenAI,
temperature: float = 1.0,
)
```
Streaming LLM client for chat completions.
AsyncOpenAI client instance
Sampling temperature (0.0 to 2.0). Lower values are more deterministic.
#### Methods
##### chat\_completion
```python theme={null}
async def chat_completion(
self,
messages: list[dict[str, str]]
) -> AsyncIterator[str]
```
Generates streaming chat completion.
Conversation history in OpenAI format. Each dict should have "role" and "content" keys.
**Returns:** `AsyncIterator[str]` - Stream of text chunks
**Example:**
```python theme={null}
from unmute.llm.llm_utils import VLLMStream, get_openai_client
client = get_openai_client()
llm = VLLMStream(client, temperature=0.7)
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "What is Python?"}
]
async for chunk in llm.chat_completion(messages):
print(chunk, end="", flush=True)
```
## Functions
### get\_openai\_client
```python theme={null}
def get_openai_client(
server_url: str = LLM_SERVER,
api_key: str | None = KYUTAI_LLM_API_KEY
) -> AsyncOpenAI
```
Creates an AsyncOpenAI client instance.
Base URL of the LLM server
API key for authentication. Defaults to "EMPTY" for vLLM servers that don't require keys.
**Returns:** `AsyncOpenAI` client
**Example:**
```python theme={null}
client = get_openai_client(
server_url="http://localhost:8000",
api_key="your-api-key"
)
```
### autoselect\_model
```python theme={null}
@cache
def autoselect_model() -> str
```
Automatically selects an LLM model from the server.
**Returns:** `str` - Model identifier
**Raises:** `ValueError` if multiple models are available (requires explicit selection)
**Notes:**
* Uses `KYUTAI_LLM_MODEL` environment variable if set
* Otherwise queries the server and selects the model if only one is available
* Result is cached for performance
### preprocess\_messages\_for\_llm
```python theme={null}
def preprocess_messages_for_llm(
chat_history: list[dict[str, str]],
) -> list[dict[str, str]]
```
Preprocesses conversation history before sending to the LLM.
Raw conversation history with "role" and "content" keys
**Returns:** `list[dict[str, str]]` - Cleaned conversation history
**Processing steps:**
1. Removes messages containing only the `INTERRUPTION_CHAR`
2. Strips `INTERRUPTION_CHAR` suffix from interrupted messages
3. Merges consecutive messages from the same role
4. Adds dummy "Hello." user message if needed for model compatibility
5. Removes `USER_SILENCE_MARKER` prefix when user continues talking
**Example:**
```python theme={null}
from unmute.llm.llm_utils import preprocess_messages_for_llm
raw_history = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there—"}, # interrupted
{"role": "user", "content": "What's Python?"}
]
processed = preprocess_messages_for_llm(raw_history)
# Result: interruption marker removed, messages cleaned
```
### rechunk\_to\_words
```python theme={null}
async def rechunk_to_words(
iterator: AsyncIterator[str]
) -> AsyncIterator[str]
```
Rechunks a text stream into whole words for better TTS pronunciation.
Stream of text chunks (may break mid-word)
**Returns:** `AsyncIterator[str]` - Stream of complete words
**Behavior:**
* Spaces are included with the following word: `"foo bar baz"` → `"foo"`, `" bar"`, `" baz"`
* Multiple whitespace characters are merged into a single space
* Buffers partial words until whitespace is encountered
**Example:**
```python theme={null}
from unmute.llm.llm_utils import rechunk_to_words
async def stream_text():
chunks = ["Hel", "lo wo", "rld! How", " are you?"]
for chunk in chunks:
yield chunk
async for word in rechunk_to_words(stream_text()):
print(f"[{word}]")
# Output:
# [Hello]
# [ world!]
# [ How]
# [ are]
# [ you?]
```
## Protocol
### LLMStream
```python theme={null}
class LLMStream(Protocol):
async def chat_completion(
self,
messages: list[dict[str, str]]
) -> AsyncIterator[str]:
...
```
Protocol for LLM streaming clients. Any class implementing `chat_completion()` can be used as an LLM stream.
## Complete Example
```python theme={null}
import asyncio
from unmute.llm.llm_utils import (
get_openai_client,
VLLMStream,
preprocess_messages_for_llm,
rechunk_to_words,
)
async def chat_with_llm():
# Initialize client
client = get_openai_client()
llm = VLLMStream(client, temperature=0.7)
# Prepare conversation
raw_history = [
{"role": "system", "content": "You are a helpful AI assistant."},
{"role": "user", "content": "Tell me about Python programming."},
]
# Preprocess messages
messages = preprocess_messages_for_llm(raw_history)
# Stream response as complete words
print("Assistant: ", end="")
async for word in rechunk_to_words(llm.chat_completion(messages)):
print(word, end="", flush=True)
print() # newline
asyncio.run(chat_with_llm())
```
## Advanced Usage: Integration with TTS
```python theme={null}
import asyncio
from unmute.llm.llm_utils import get_openai_client, VLLMStream, rechunk_to_words
from unmute.tts.text_to_speech import TextToSpeech, TTSClientEosMessage
async def speak_llm_response():
# Initialize services
client = get_openai_client()
llm = VLLMStream(client, temperature=0.7)
tts = TextToSpeech()
await tts.start_up()
messages = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Explain machine learning briefly."}
]
# Stream LLM response directly to TTS
async for word in rechunk_to_words(llm.chat_completion(messages)):
await tts.send(word)
# Signal end of text
await tts.send(TTSClientEosMessage())
# Receive synthesized audio
async for message in tts:
pass # Process audio messages
await tts.shutdown()
asyncio.run(speak_llm_response())
```
## Environment Variables
* `KYUTAI_LLM_MODEL`: Model identifier to use (if not set, auto-selects)
* `KYUTAI_LLM_API_KEY`: API key for LLM server
* `LLM_SERVER`: Base URL of the LLM server
## Notes
* The `VLLMStream` class auto-selects the model if not explicitly configured
* Message preprocessing handles common conversation artifacts (interruptions, silence markers)
* Word rechunking is essential for natural TTS pronunciation
* All async functions should be run within an event loop
* The client supports any OpenAI-compatible API (vLLM, llama.cpp, etc.)
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.