"""Internal LLM types: stream events, tool calls, tool results.""" from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Literal StreamEventType = Literal[ "text_delta", "tool_call_start", "tool_call_delta", "tool_call_end", "usage", "done", "error", "reasoning_delta", ] @dataclass class StreamEvent: type: StreamEventType data: dict[str, Any] = field(default_factory=dict) @dataclass class ToolCallReq: id: str name: str arguments_json: str = "" index: int = 0 def arguments(self) -> dict[str, Any]: import json if not self.arguments_json.strip(): return {} try: v = json.loads(self.arguments_json) return v if isinstance(v, dict) else {} except json.JSONDecodeError: return {"__invalid_json__": self.arguments_json[:2000]} @dataclass class Artifact: type: str # xlsx | image | csv | file | quiz | sources | search file_id: str | None = None filename: str | None = None preview: Any = None url: str | None = None def to_dict(self) -> dict[str, Any]: return { "type": self.type, "file_id": self.file_id, "filename": self.filename, "preview": self.preview, "url": self.url, } @dataclass class ToolResult: content: str # for the model artifacts: list[Artifact] = field(default_factory=list) meta: dict[str, Any] = field(default_factory=dict) payload: dict[str, Any] = field(default_factory=dict) # for the UI card error: bool = False def summary(self, limit: int = 240) -> str: s = self.meta.get("summary") or self.content return s if len(s) <= limit else s[: limit - 1] + "…" @dataclass class Usage: input_tokens: int = 0 output_tokens: int = 0 cost_usd: float = 0.0 model: str = "" provider: str = "" generation_id: str = ""