65 lines
1.6 KiB
Python
65 lines
1.6 KiB
Python
"""LLM Response - Unified message classes for LLM communication"""
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, Optional
|
|
|
|
|
|
class StepType:
|
|
"""Step type constants"""
|
|
THINKING = "thinking"
|
|
TEXT = "text"
|
|
TOOL_CALL = "tool_call"
|
|
TOOL_RESULT = "tool_result"
|
|
|
|
|
|
@dataclass
|
|
class Step:
|
|
"""Single step - used for storage and transport"""
|
|
id: str
|
|
index: int
|
|
type: str
|
|
content: str = ""
|
|
name: str = ""
|
|
arguments: str = ""
|
|
id_ref: str = ""
|
|
success: bool = True
|
|
|
|
def to_dict(self) -> Dict:
|
|
return {
|
|
"id": self.id,
|
|
"index": self.index,
|
|
"type": self.type,
|
|
"content": self.content,
|
|
"name": self.name,
|
|
"arguments": self.arguments,
|
|
"id_ref": self.id_ref,
|
|
"success": self.success
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class ParsedDelta:
|
|
"""LLM streaming response delta"""
|
|
content: str = ""
|
|
thinking: str = "" # Includes DeepSeek reasoning_content
|
|
text: str = ""
|
|
tool_call: Optional[Dict] = None
|
|
usage: Dict[str, int] = field(default_factory=dict)
|
|
is_complete: bool = False
|
|
error_msg: Optional[str] = None
|
|
|
|
def has_thinking(self) -> bool:
|
|
return bool(self.thinking)
|
|
|
|
def has_text(self) -> bool:
|
|
return bool(self.text)
|
|
|
|
def has_tool_call(self) -> bool:
|
|
return self.tool_call is not None
|
|
|
|
def has_content(self) -> bool:
|
|
return bool(self.content) or self.has_thinking() or self.has_text() or self.has_tool_call()
|
|
|
|
def has_error(self) -> bool:
|
|
return self.error_msg is not None
|