Chat generation result object
| 164 | |
| 165 | |
| 166 | class ChatGenerationResult: |
| 167 | """Chat generation result object""" |
| 168 | def __init__(self, chunk: ChatChunk|None = None): |
| 169 | self.reasoning = "" |
| 170 | self.response = "" |
| 171 | self.thinking = False |
| 172 | self.thinking_tag = "" |
| 173 | self.unprocessed = "" |
| 174 | self.native_reasoning = False |
| 175 | self.thinking_pairs = [("<think>", "</think>"), ("<reasoning>", "</reasoning>")] |
| 176 | if chunk: |
| 177 | self.add_chunk(chunk) |
| 178 | |
| 179 | def add_chunk(self, chunk: ChatChunk) -> ChatChunk: |
| 180 | if chunk["reasoning_delta"]: |
| 181 | self.native_reasoning = True |
| 182 | |
| 183 | # if native reasoning detection works, there's no need to worry about thinking tags |
| 184 | if self.native_reasoning: |
| 185 | processed_chunk = ChatChunk(response_delta=chunk["response_delta"], reasoning_delta=chunk["reasoning_delta"]) |
| 186 | else: |
| 187 | # if the model outputs thinking tags, we ned to parse them manually as reasoning |
| 188 | processed_chunk = self._process_thinking_chunk(chunk) |
| 189 | |
| 190 | self.reasoning += processed_chunk.get("reasoning_delta", "") |
| 191 | self.response += processed_chunk.get("response_delta", "") |
| 192 | |
| 193 | return processed_chunk |
| 194 | |
| 195 | def _process_thinking_chunk(self, chunk: ChatChunk) -> ChatChunk: |
| 196 | response_delta = self.unprocessed + chunk["response_delta"] |
| 197 | self.unprocessed = "" |
| 198 | return self._process_thinking_tags(response_delta, chunk["reasoning_delta"]) |
| 199 | |
| 200 | def _process_thinking_tags(self, response: str, reasoning: str) -> ChatChunk: |
| 201 | if self.thinking: |
| 202 | close_pos = response.find(self.thinking_tag) |
| 203 | if close_pos != -1: |
| 204 | reasoning += response[:close_pos] |
| 205 | response = response[close_pos + len(self.thinking_tag):] |
| 206 | self.thinking = False |
| 207 | self.thinking_tag = "" |
| 208 | else: |
| 209 | if self._is_partial_closing_tag(response): |
| 210 | self.unprocessed = response |
| 211 | response = "" |
| 212 | else: |
| 213 | reasoning += response |
| 214 | response = "" |
| 215 | else: |
| 216 | for opening_tag, closing_tag in self.thinking_pairs: |
| 217 | if response.startswith(opening_tag): |
| 218 | response = response[len(opening_tag):] |
| 219 | self.thinking = True |
| 220 | self.thinking_tag = closing_tag |
| 221 | |
| 222 | close_pos = response.find(closing_tag) |
| 223 | if close_pos != -1: |
no outgoing calls
no test coverage detected