Chat

chat_template: PromptTemplate | None

Get the custom chat template, if any.

client: Client | AsyncClient | None

Get the client used by this chat.

items: ConversationItems

Read-only access to conversation as typed Items.

last_response: Response | StreamingResponse | AsyncStreamingResponse | None

Get the last response from generation.

max_context_length: int | None

Get the model's maximum context length.

messages: list

Read-only view of conversation as standard OpenAI-format messages.

owns_client: bool

True if this Chat owns its Client (standalone mode).

prompt_id: str | None

The prompt document ID for this conversation.

When set, this links the conversation to a prompt/persona document. The prompt document can provide the system prompt content and tags that can be inherited via inherit_tags().

This is stored on the Chat object and used by inherit_tags(). For persistent lineage tracking in session records, use source_doc_id.

router: Router | None

Get the router used by this chat.

session_id: str | None

The session identifier for this conversation.

source_doc_id: str | None

The source document ID for lineage tracking.

Links this conversation to the prompt/persona document that spawned it. Used for tracking which document was used to create the conversation.

system: str | None

Get the system prompt.

def __call__(
    self,
    message: str | list[dict] | MessageItem | list[MessageItem],
    config: GenerationConfig | None = None,
    stream: bool = True,
    on_token: Callable[[str], None] | None = None,
    response_format: type | dict | Grammar | None = None,
    kwargs: Any
) → Response | StreamingResponse

Send a message and get a streaming response (callable syntax).

This is the primary way to chat. Call the Chat object directly with your message. By default, returns a StreamingResponse for real-time token display. Use stream=False for complete response.

For async usage, use send_async() instead.

>>> chat = Chat("Qwen/Qwen3-0.6B")
>>> for token in chat("Tell me a joke"):
...     print(token, end="", flush=True)

>>> response = chat("What is 2+2?", stream=False)
>>> print(response)
4

>>> from dataclasses import dataclass
>>> @dataclass
... class Answer:
...     value: int
>>> response = chat("What is 2+2?", response_format=Answer, stream=False)
>>> response.parsed.value
4

>>> chat = Chat("model", config=GenerationConfig(temperature=0.7))
>>> response = chat("Hi", temperature=0.1)  # Uses 0.1 for this call
>>> chat.config.temperature  # Still 0.7 (unchanged)
0.7

>>> response = chat("What is 2+2?", stream=False)
>>> response = response.append("Why?")  # Inherits stream=False

def append(
    self,
    role_or_item: str | MessageItem,
    content: str | None = None,
    hidden: bool = False
) → Self

Append a message to the conversation.

def append_hidden(
    self,
    role: str,
    content: str
) → Self

Append a hidden message to the conversation.

Hidden messages are included in LLM context but omitted from UI history.

def clear(self) → Self

Clear conversation history (keeps system prompt and settings).

def close(self) → None

Close the chat and release resources immediately.

If this Chat created its own internal Client (via model="..."), the Client and its Engine are closed, freeing memory.

If this Chat uses a shared Client (via client=...), only the lightweight chat state is freed. The Client stays alive.

Safe to call multiple times.

>>> for model in ["Qwen/0.5B", "Qwen/1.5B", "Qwen/4B"]:
...     chat = Chat(model)
...     print(chat("Hello"))
...     chat.close()  # Free memory before loading next model

>>> with Chat("Qwen/0.5B") as chat:
...     print(chat("Hello"))
... # Memory freed automatically here

def count_tokens(self, message: str | None = None) → int

Count tokens in current history or a hypothetical message.

def fork(self) → Chat

Fork this chat to explore alternative conversation paths.

Creates an independent copy of the chat with the same history, config, and client reference. Changes to the forked chat do not affect the original.

>>> chat = Chat("Qwen/Qwen3-0.6B")
>>> response = chat("I have chicken")
>>>
>>> # Fork to try different directions
>>> asian = response.chat.fork()
>>> italian = response.chat.fork()
>>>
>>> asian("Suggest an Asian recipe")
>>> italian("Suggest an Italian recipe")
>>>
>>> # Original unchanged
>>> print(len(chat.items))  # Same as before forking

def from_dict(
    cls,
    data: dict,
    model: str | None = None
) → Self

Deserialize from dictionary.

Restores a Chat/AsyncChat from a dict created by to_dict(). Use this to resume conversations from a database or file.

def inherit_tags(self) → None

Copy tags from the prompt document to this conversation.

Requires both prompt_id and session_id to be set, and requires TaluDB storage to be configured.

def insert(
    self,
    index: int,
    role: str,
    content: str,
    hidden: bool = False
) → Self

Insert a message at the specified index.

def pop(self) → Self

Remove and discard the last message.

def preview_prompt(
    self,
    add_generation_prompt: bool = True,
    config: GenerationConfig | None = None
) → str

Return the exact formatted prompt that would be sent to the model.

This is a read-only inspection tool for debugging template logic or verifying system prompts. It does NOT send anything to the engine or affect the conversation state.

def regenerate(
    self,
    message: str | None = None,
    config: GenerationConfig | None = None,
    stream: bool = False,
    response_format: type | dict | Grammar | None = None,
    kwargs: Any
) → Response | StreamingResponse

Regenerate the last conversation turn.

This method unwinds the conversation to the previous user message and triggers generation again. Use it to retry a response or edit the last user message.

The operation is atomic: it truncates to the point before the last user message and then sends (either the original or new text). This ensures fresh item IDs and timestamps for auditability.

>>> chat = Chat("Qwen/Qwen3-0.6B")
>>> chat("Tell me a joke")
>>> # Didn't like the joke? Retry:
>>> chat.regenerate()
>>> # Or edit and retry:
>>> chat.regenerate(message="Tell me a better joke")
>>> # With different parameters:
>>> chat.regenerate(temperature=1.2)

def remove(self, index: int) → Self

Remove message at the specified index.

def reset(self) → Self

Reset everything including system prompt.

def send(
    self,
    message: str | list[dict] | MessageItem | list[MessageItem],
    config: GenerationConfig | None = None,
    tools: list[Callable[..., Any]] | None = None,
    stream: bool = False,
    on_token: Callable[[str], None] | None = None,
    response_format: type | dict | Grammar | None = None,
    kwargs: Any
) → Response | StreamingResponse

Send a message and get a response (synchronous).

This is the explicit sync method. For streaming default, use chat(). For async, use send_async().

>>> response = chat.send("What is 2+2?")
>>> print(response)
4
>>> response = response.append("Why?")  # Continues with same mode

def set_item_parent(
    self,
    item_index: int,
    parent_item_id: int | None
) → None

Set parent_item_id for an item by index.

def set_item_validation_flags(
    self,
    item_index: int,
    json_valid: bool,
    schema_valid: bool,
    repaired: bool = False
) → None

Set structured validation flags for an item by index.

Use this after structured parsing/validation to mark JSON/schema validity.

def to_dict(self) → dict

Serialize to dictionary.

def to_json(self) → str

Get messages as JSON string (from Zig).

chat: Chat | None

The Chat that generated this response.

content: list[ContentPart]

Structured content parts for multimodal output symmetry.

Returns a list of content parts, enabling symmetric handling of input and output. For text-only responses, this returns [OutputText(text=...)]. Future multimodal models will return additional part types (OutputImage, etc.).

This property is the source of truth for response content. The .text property is a convenience that concatenates all text parts.

>>> response = chat("Hello!")
>>> for part in response.content:
...     if part.type == ContentType.OUTPUT_TEXT:
...         print(part.text)

finish_reason: str

Why generation stopped.

logprobs: list[TokenLogprob] | None

Token log probabilities (if requested).

model: str | None

Model identifier that generated this response.

parsed: Any

Parse and validate the response against the response_format schema.

If a response_format was provided during generation, this property parses the response text as JSON and validates/hydrates it into the specified type (dataclass or Pydantic model).

prompt: str | None

The fully rendered prompt sent to the model (audit trail).

Contains the exact string that was fed to the model engine after all templating, system prompt injection, and formatting was applied. Useful for debugging template issues and understanding exactly what the model saw.

>>> response = chat("Hello!")
>>> print(response.prompt)
<|im_start|>system
You are a helpful assistant.
<|im_end|>
<|im_start|>user
Hello!
<|im_end|>
<|im_start|>assistant

text: str

The generated text content.

timings: Timings | None

Generation timing breakdown.

tokens: list[int]

List of generated token IDs.

tool_calls: list[ToolCall] | None

Tool calls requested by the model (if any).

usage: Usage | None

Token usage statistics.

def append(
    self,
    message: str,
    kwargs: Any
) → Response | StreamingResponse

Continue the conversation with a follow-up message (sync).

This is the primary way to have multi-turn conversations. The append uses the same Chat that generated this response, maintaining context.

Auto-Fork Behavior: If the conversation has moved past this response (i.e., more messages were added after this response was generated), append() automatically forks the conversation and truncates it back to this point before sending the new message. This enables intuitive branching where you can append to any previous response without worrying about conversation state.

The append automatically inherits streaming mode from the original response.

>>> r1 = chat("What is 2+2?")
>>> r2 = r1.append("Why?")      # Continues normally
>>> r3 = r2.append("Thanks!")   # Continues normally

>>> r1 = chat("Idea 1")
>>> r2 = r1.append("Critique it")   # chat has [Idea 1, Critique]
>>> r3 = r1.append("Expand on it")  # Auto-forks! r3.chat is new
>>> # Original chat unchanged, r3.chat has [Idea 1, Expand]

def endswith(self, suffix: str) → bool

Check if text ends with suffix.

def lower(self) → str

Return text in lowercase.

def replace(
    self,
    old: str,
    new: str,
    count: int = -1
) → str

Replace occurrences in text.

def split(
    self,
    sep: str | None = None,
    maxsplit: int = -1
) → list[str]

Split text.

def startswith(self, prefix: str) → bool

Check if text starts with prefix.

def strip(self, chars: str | None = None) → str

Return text with leading/trailing chars removed.

def submit_tool_result(
    self,
    tool_call_id: str,
    result: Any
) → Response

Submit a tool result and continue generation.

When the model requests tool calls (via response.tool_calls), execute them and submit the results back using this method. The model will then continue generation with the tool results in context.

>>> response = chat("What's the weather?", tools=[get_weather])
>>> while response.tool_calls:
...     for call in response.tool_calls:
...         result = call.execute()
...         response = response.submit_tool_result(call.id, result)
>>> print(response)

def to_dict(self) → dict[str, Any]

Convert response to a JSON-serializable dictionary.

This solves the "serialization trap" where Response acts like a string but isn't directly JSON serializable. Use this for API responses, logging, or any context requiring JSON.

>>> @app.post("/chat")
>>> async def chat_endpoint(message: str):
...     response = await chat(message)
...     return response.to_dict()  # JSON serializable

>>> import json
>>> response = chat("Hello!")
>>> json.dumps(response.to_dict())  # Works!

>>> result = {
...     "success": True,
...     "data": response.to_dict(),
... }

def upper(self) → str

Return text in uppercase.

chat: Chat | None

The Chat that generated this response.

content: list[ContentPart]

Structured content parts for multimodal output symmetry.

Returns a list of content parts, enabling symmetric handling of input and output. For text-only responses, this returns [OutputText(text=...)]. Future multimodal models will return additional part types (OutputImage, etc.).

This property is the source of truth for response content. The .text property is a convenience that concatenates all text parts.

>>> response = chat("Hello!")
>>> for part in response.content:
...     if part.type == ContentType.OUTPUT_TEXT:
...         print(part.text)

finish_reason: str

Why generation stopped.

logprobs: list[TokenLogprob] | None

Token log probabilities (if requested).

model: str | None

Model identifier that generated this response.

prompt: str | None

The fully rendered prompt (available after iteration completes).

For streaming responses, the prompt is captured after iteration finishes since messages are added during streaming. Access this property after consuming the stream.

text: str

The generated text content.

For StreamingResponse, accessing this property will auto-drain the stream if it hasn't been consumed yet. This ensures that `.text` always returns the complete generated text, regardless of whether the caller explicitly iterated over the response.

timings: Timings | None

Generation timing breakdown.

tokens: list[int]

List of generated token IDs.

tool_calls: list[ToolCall] | None

Tool calls requested by the model (if any).

usage: Usage | None

Token usage statistics.

def append(
    self,
    message: str,
    kwargs: Any
) → StreamingResponse

Continue the conversation with a follow-up message.

Returns a StreamingResponse (inherits streaming mode from this response). See Response.append() for full documentation including auto-fork behavior.

def endswith(self, suffix: str) → bool

Check if text ends with suffix.

def lower(self) → str

Return text in lowercase.

def replace(
    self,
    old: str,
    new: str,
    count: int = -1
) → str

Replace occurrences in text.

def split(
    self,
    sep: str | None = None,
    maxsplit: int = -1
) → list[str]

Split text.

def startswith(self, prefix: str) → bool

Check if text starts with prefix.

def strip(self, chars: str | None = None) → str

Return text with leading/trailing chars removed.

def to_dict(self) → dict[str, Any]

Convert response to a JSON-serializable dictionary.

This solves the "serialization trap" where Response acts like a string but isn't directly JSON serializable. Use this for API responses, logging, or any context requiring JSON.

>>> @app.post("/chat")
>>> async def chat_endpoint(message: str):
...     response = await chat(message)
...     return response.to_dict()  # JSON serializable

>>> import json
>>> response = chat("Hello!")
>>> json.dumps(response.to_dict())  # Works!

>>> result = {
...     "success": True,
...     "data": response.to_dict(),
... }

def upper(self) → str

Return text in uppercase.

chat_template: PromptTemplate | None

Get the custom chat template, if any.

client: Client | AsyncClient | None

Get the client used by this chat.

items: ConversationItems

Read-only access to conversation as typed Items.

last_response: Response | StreamingResponse | AsyncStreamingResponse | None

Get the last response from generation.

max_context_length: int | None

Get the model's maximum context length.

messages: list

Read-only view of conversation as standard OpenAI-format messages.

owns_client: bool

True if this Chat owns its Client (standalone mode).

prompt_id: str | None

The prompt document ID for this conversation.

When set, this links the conversation to a prompt/persona document. The prompt document can provide the system prompt content and tags that can be inherited via inherit_tags().

This is stored on the Chat object and used by inherit_tags(). For persistent lineage tracking in session records, use source_doc_id.

router: Router | None

Get the router used by this chat.

session_id: str | None

The session identifier for this conversation.

source_doc_id: str | None

The source document ID for lineage tracking.

Links this conversation to the prompt/persona document that spawned it. Used for tracking which document was used to create the conversation.

system: str | None

Get the system prompt.

def __call__(
    self,
    message: str | list[dict] | MessageItem | list[MessageItem],
    config: GenerationConfig | None = None,
    stream: bool = True,
    on_token: Callable[[str], None] | None = None,
    response_format: type | dict | Grammar | None = None,
    kwargs: Any
) → AsyncResponse | AsyncStreamingResponse

Send a message and get an async streaming response (callable syntax).

This is the primary async way to chat. Call the AsyncChat object directly with your message. By default, returns an AsyncStreamingResponse.

>>> response = await chat("Tell me a joke")
>>> async for token in response:
...     print(token, end="", flush=True)

def append(
    self,
    role_or_item: str | MessageItem,
    content: str | None = None,
    hidden: bool = False
) → Self

Append a message to the conversation.

def append_hidden(
    self,
    role: str,
    content: str
) → Self

Append a hidden message to the conversation.

Hidden messages are included in LLM context but omitted from UI history.

def clear(self) → Self

Clear conversation history (keeps system prompt and settings).

def close(self) → None

Close the chat and release resources immediately.

If this AsyncChat created its own internal AsyncClient (via model="..."), the Client and its Engine are closed, freeing memory.

If this AsyncChat uses a shared AsyncClient (via client=...), only the lightweight chat state is freed. The Client stays alive.

Safe to call multiple times.

>>> for model in ["Qwen/0.5B", "Qwen/1.5B", "Qwen/4B"]:
...     chat = AsyncChat(model)
...     print(await chat("Hello"))
...     await chat.close()  # Free memory before loading next model

>>> async with AsyncChat("Qwen/0.5B") as chat:
...     print(await chat("Hello"))
... # Memory freed automatically here

def count_tokens(self, message: str | None = None) → int

Count tokens in current history or a hypothetical message.

def fork(self) → AsyncChat

Fork this chat to explore alternative conversation paths.

def from_dict(
    cls,
    data: dict,
    model: str | None = None
) → Self

Deserialize from dictionary.

Restores a Chat/AsyncChat from a dict created by to_dict(). Use this to resume conversations from a database or file.

def inherit_tags(self) → None

Copy tags from the prompt document to this conversation.

Requires both prompt_id and session_id to be set, and requires TaluDB storage to be configured.

def insert(
    self,
    index: int,
    role: str,
    content: str,
    hidden: bool = False
) → Self

Insert a message at the specified index.

def pop(self) → Self

Remove and discard the last message.

def preview_prompt(
    self,
    add_generation_prompt: bool = True,
    config: GenerationConfig | None = None
) → str

Return the exact formatted prompt that would be sent to the model.

This is a read-only inspection tool for debugging template logic or verifying system prompts. It does NOT send anything to the engine or affect the conversation state.

def regenerate(
    self,
    message: str | None = None,
    config: GenerationConfig | None = None,
    stream: bool = False,
    response_format: type | dict | Grammar | None = None,
    kwargs: Any
) → AsyncResponse | AsyncStreamingResponse

Regenerate the last conversation turn.

This method unwinds the conversation to the previous user message and triggers generation again. Use it to retry a response or edit the last user message.

The operation is atomic: it truncates to the point before the last user message and then sends (either the original or new text). This ensures fresh item IDs and timestamps for auditability.

>>> chat = AsyncChat("Qwen/Qwen3-0.6B")
>>> await chat("Tell me a joke")
>>> # Didn't like the joke? Retry:
>>> await chat.regenerate()
>>> # Or edit and retry:
>>> await chat.regenerate(message="Tell me a better joke")

def remove(self, index: int) → Self

Remove message at the specified index.

def reset(self) → Self

Reset everything including system prompt.

def send(
    self,
    message: str | list[dict] | MessageItem | list[MessageItem],
    config: GenerationConfig | None = None,
    tools: list[Callable[..., Any]] | None = None,
    stream: bool = False,
    on_token: Callable[[str], None] | None = None,
    response_format: type | dict | Grammar | None = None,
    kwargs: Any
) → AsyncResponse | AsyncStreamingResponse

Send a message and get a response (async, non-streaming by default).

>>> response = await chat.send("What is 2+2?")
>>> print(response)
4
>>> response = await response.append("Why?")

def set_item_parent(
    self,
    item_index: int,
    parent_item_id: int | None
) → None

Set parent_item_id for an item by index.

def set_item_validation_flags(
    self,
    item_index: int,
    json_valid: bool,
    schema_valid: bool,
    repaired: bool = False
) → None

Set structured validation flags for an item by index.

Use this after structured parsing/validation to mark JSON/schema validity.

def to_dict(self) → dict

Serialize to dictionary.

def to_json(self) → str

Get messages as JSON string (from Zig).

chat: Chat | None

The Chat that generated this response.

content: list[ContentPart]

Structured content parts for multimodal output symmetry.

Returns a list of content parts, enabling symmetric handling of input and output. For text-only responses, this returns [OutputText(text=...)]. Future multimodal models will return additional part types (OutputImage, etc.).

This property is the source of truth for response content. The .text property is a convenience that concatenates all text parts.

>>> response = chat("Hello!")
>>> for part in response.content:
...     if part.type == ContentType.OUTPUT_TEXT:
...         print(part.text)

finish_reason: str

Why generation stopped.

logprobs: list[TokenLogprob] | None

Token log probabilities (if requested).

model: str | None

Model identifier that generated this response.

parsed: Any

Parse and validate the response against the response_format schema.

If a response_format was provided during generation, this property parses the response text as JSON and validates/hydrates it into the specified type (dataclass or Pydantic model).

prompt: str | None

The fully rendered prompt sent to the model (audit trail).

Contains the exact string that was fed to the model engine after all templating, system prompt injection, and formatting was applied. Useful for debugging template issues and understanding exactly what the model saw.

>>> response = chat("Hello!")
>>> print(response.prompt)
<|im_start|>system
You are a helpful assistant.
<|im_end|>
<|im_start|>user
Hello!
<|im_end|>
<|im_start|>assistant

text: str

The generated text content.

timings: Timings | None

Generation timing breakdown.

tokens: list[int]

List of generated token IDs.

tool_calls: list[ToolCall] | None

Tool calls requested by the model (if any).

usage: Usage | None

Token usage statistics.

def append(
    self,
    message: str,
    kwargs: Any
) → AsyncResponse | AsyncStreamingResponse

Continue the conversation with a follow-up message (async).

This is the async way to have multi-turn conversations. The append uses the same AsyncChat that generated this response. Must be awaited.

Auto-Fork Behavior: If the conversation has moved past this response, append() automatically forks the conversation and truncates it back to this point before sending the new message. See Response.append() for details.

The append automatically inherits streaming mode from the original response.

>>> response = await chat.send("What is 2+2?")
>>> response = await response.append("Why?")
>>> response = await response.append("Are you sure?")

def endswith(self, suffix: str) → bool

Check if text ends with suffix.

def lower(self) → str

Return text in lowercase.

def replace(
    self,
    old: str,
    new: str,
    count: int = -1
) → str

Replace occurrences in text.

def split(
    self,
    sep: str | None = None,
    maxsplit: int = -1
) → list[str]

Split text.

def startswith(self, prefix: str) → bool

Check if text starts with prefix.

def strip(self, chars: str | None = None) → str

Return text with leading/trailing chars removed.

def submit_tool_result(
    self,
    tool_call_id: str,
    result: Any
) → AsyncResponse

Submit a tool result and continue generation (async).

def to_dict(self) → dict[str, Any]

Convert response to a JSON-serializable dictionary.

This solves the "serialization trap" where Response acts like a string but isn't directly JSON serializable. Use this for API responses, logging, or any context requiring JSON.

>>> @app.post("/chat")
>>> async def chat_endpoint(message: str):
...     response = await chat(message)
...     return response.to_dict()  # JSON serializable

>>> import json
>>> response = chat("Hello!")
>>> json.dumps(response.to_dict())  # Works!

>>> result = {
...     "success": True,
...     "data": response.to_dict(),
... }

def upper(self) → str

Return text in uppercase.

chat: Chat | None

The Chat that generated this response.

content: list[ContentPart]

Structured content parts for multimodal output symmetry.

Returns a list of content parts, enabling symmetric handling of input and output. For text-only responses, this returns [OutputText(text=...)]. Future multimodal models will return additional part types (OutputImage, etc.).

This property is the source of truth for response content. The .text property is a convenience that concatenates all text parts.

>>> response = chat("Hello!")
>>> for part in response.content:
...     if part.type == ContentType.OUTPUT_TEXT:
...         print(part.text)

finish_reason: str

Why generation stopped.

logprobs: list[TokenLogprob] | None

Token log probabilities (if requested).

model: str | None

Model identifier that generated this response.

prompt: str | None

The fully rendered prompt (available after iteration completes).

For async streaming responses, the prompt is captured after iteration finishes since messages are added during streaming. Access this property after consuming the stream.

text: str

The generated text content.

For AsyncStreamingResponse, this returns the text accumulated so far. To get the complete text, ensure you have consumed the stream first by iterating with async for token in response.

async for _ in response:
    pass
full_text = response.text

timings: Timings | None

Generation timing breakdown.

tokens: list[int]

List of generated token IDs.

tool_calls: list[ToolCall] | None

Tool calls requested by the model (if any).

usage: Usage | None

Token usage statistics.

def append(
    self,
    message: str,
    kwargs: Any
) → AsyncStreamingResponse

Continue the conversation with a follow-up message (async streaming).

Returns AsyncStreamingResponse (inherits streaming mode). Must be awaited.

See Response.append() for full documentation including auto-fork behavior.

>>> response = await chat("Hello")  # stream=True by default
>>> async for token in response:
...     print(token, end="")
>>> response2 = await response.append("Continue")
>>> async for token in response2:
...     print(token, end="")

def endswith(self, suffix: str) → bool

Check if text ends with suffix.

def lower(self) → str

Return text in lowercase.

def replace(
    self,
    old: str,
    new: str,
    count: int = -1
) → str

Replace occurrences in text.

def split(
    self,
    sep: str | None = None,
    maxsplit: int = -1
) → list[str]

Split text.

def startswith(self, prefix: str) → bool

Check if text starts with prefix.

def strip(self, chars: str | None = None) → str

Return text with leading/trailing chars removed.

def to_dict(self) → dict[str, Any]

Convert response to a JSON-serializable dictionary.

This solves the "serialization trap" where Response acts like a string but isn't directly JSON serializable. Use this for API responses, logging, or any context requiring JSON.

>>> @app.post("/chat")
>>> async def chat_endpoint(message: str):
...     response = await chat(message)
...     return response.to_dict()  # JSON serializable

>>> import json
>>> response = chat("Hello!")
>>> json.dumps(response.to_dict())  # Works!

>>> result = {
...     "success": True,
...     "data": response.to_dict(),
... }

def upper(self) → str

Return text in uppercase.

first: ConversationItem | None

Get the first item, or None if empty.

last: ConversationItem | None

Get the last item, or None if empty.

system: str | None

Get the system message content, or None if no system message.

def count(self, value)

S.count(value) -> integer -- return number of occurrences of value

def filter_by_role(self, role: MessageRole) → list[MessageItem]

Get all message items with a specific role.

def filter_by_type(self, item_type: type[ConversationItem]) → list[ConversationItem]

Get all items of a specific type.

def index(
    self,
    value,
    start = 0,
    stop = None
)

S.index(value, [start, [stop]]) -> integer -- return first index of value. Raises ValueError if the value is not present.

Supporting start and stop arguments is optional, but recommended.

def on_first_token(
    self,
    chat: Chat | AsyncChat,
    time_ms: float
) → None

Handle first token event (streaming only).

This is the Time-To-First-Token (TTFT) measurement point, critical for perceived latency in interactive applications.

>>> def on_first_token(self, chat, time_ms):
...     metrics.histogram("llm.ttft", time_ms)

def on_generation_end(
    self,
    chat: Chat | AsyncChat,
    response: Response | None,
    error: Exception | None = None
) → None

Handle generation end event (success or error).

>>> def on_generation_end(self, chat, response, error=None):
...     if error:
...         metrics.counter("llm.errors", 1)
...     else:
...         metrics.counter("llm.tokens", response.usage.total_tokens)

def on_generation_start(
    self,
    chat: Chat | AsyncChat,
    input_text: str,
    config: Any = None
) → None

Handle generation start event.

>>> def on_generation_start(self, chat, input_text, config=None):
...     self.start_time = time.perf_counter()
...     self.input_tokens = len(input_text.split())  # Rough estimate

hooks: list[Hook]

Return the list of registered hooks.

def add(self, hook: Hook) → None

Add a hook to the manager.

def dispatch_end(
    self,
    chat: Chat | AsyncChat,
    response: Response | None,
    error: Exception | None = None
) → None

Dispatch on_generation_end to all hooks.

def dispatch_first_token(
    self,
    chat: Chat | AsyncChat,
    time_ms: float
) → None

Dispatch on_first_token to all hooks.

def dispatch_start(
    self,
    chat: Chat | AsyncChat,
    input_text: str,
    config: Any = None
) → None

Dispatch on_generation_start to all hooks.

def remove(self, hook: Hook) → None

Remove a hook from the manager.

def from_ns(
    cls,
    prefill_ns: int,
    generation_ns: int,
    token_count: int
) → Timings

Create Timings from nanosecond values.