66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
"""LLM Response - Unified message classes for LLM communication
|
|
|
|
This module provides unified data classes for message passing throughout the LLM pipeline.
|
|
"""
|
|
from typing import Dict, Any, List, Optional
|
|
from dataclasses import dataclass, field
|
|
|
|
|
|
@dataclass
|
|
class ParsedDelta:
|
|
"""Streaming response delta
|
|
|
|
Represents a single unit of streaming response data.
|
|
Used for streaming responses where content is accumulated incrementally.
|
|
|
|
Attributes:
|
|
thinking: Accumulated thinking/reasoning content
|
|
text: Accumulated text content
|
|
tool_calls: List of tool call requests
|
|
is_complete: Whether this is the final delta
|
|
usage: Token usage statistics
|
|
"""
|
|
thinking: str = ""
|
|
text: str = ""
|
|
tool_calls: List[Dict] = field(default_factory=list)
|
|
is_complete: bool = False
|
|
usage: Dict[str, int] = field(default_factory=dict)
|
|
|
|
def has_thinking(self) -> bool:
|
|
"""Check if there's thinking content"""
|
|
return bool(self.thinking)
|
|
|
|
def has_text(self) -> bool:
|
|
"""Check if there's text content"""
|
|
return bool(self.text)
|
|
|
|
def has_tool_calls(self) -> bool:
|
|
"""Check if there are tool calls"""
|
|
return bool(self.tool_calls)
|
|
|
|
def has_content(self) -> bool:
|
|
"""Check if there's any content"""
|
|
return self.has_thinking() or self.has_text() or self.has_tool_calls()
|
|
|
|
|
|
@dataclass
|
|
class LLMResponse:
|
|
"""Complete LLM response
|
|
|
|
Represents a complete non-streaming response.
|
|
|
|
Attributes:
|
|
content: Final text content
|
|
thinking: Final thinking content (if any)
|
|
tool_calls: List of tool calls (if any)
|
|
usage: Token usage statistics
|
|
"""
|
|
content: str = ""
|
|
thinking: str = ""
|
|
tool_calls: List[Dict] = field(default_factory=list)
|
|
usage: Dict[str, int] = field(default=dict)
|
|
|
|
def has_tool_calls(self) -> bool:
|
|
"""Check if there are tool calls"""
|
|
return bool(self.tool_calls)
|