File size: 4,367 Bytes
25effb7 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 |
import re
import time
from config import logger
class ParserState:
__slots__ = ['answer', 'thought', 'in_think', 'in_answer', 'start_time', 'last_pos', 'total_think_time']
def __init__(self):
self.answer = ""
self.thought = ""
self.in_think = False
self.in_answer = False
self.start_time = 0
self.last_pos = 0
self.total_think_time = 0.0
def format_time(seconds_float):
total_seconds = int(round(seconds_float))
hours = total_seconds // 3600
remaining_seconds = total_seconds % 3600
minutes = remaining_seconds // 60
seconds = remaining_seconds % 60
if hours > 0:
return f"{hours}h {minutes}m {seconds}s"
elif minutes > 0:
return f"{minutes}m {seconds}s"
else:
return f"{seconds}s"
def parse_response(text, state):
buffer = text[state.last_pos:]
state.last_pos = len(text)
while buffer:
if not state.in_think and not state.in_answer:
think_start = buffer.find('<think>')
reasoning_start = buffer.find('<reasoning>')
answer_start = buffer.find('<answer>')
starts = []
if think_start != -1:
starts.append((think_start, '<think>', 7, 'think'))
if reasoning_start != -1:
starts.append((reasoning_start, '<reasoning>', 11, 'think'))
if answer_start != -1:
starts.append((answer_start, '<answer>', 8, 'answer'))
if not starts:
state.answer += buffer
break
start_pos, start_tag, tag_length, mode = min(starts, key=lambda x: x[0])
state.answer += buffer[:start_pos]
if mode == 'think':
state.in_think = True
state.start_time = time.perf_counter()
else:
state.in_answer = True
buffer = buffer[start_pos + tag_length:]
elif state.in_think:
think_end = buffer.find('</think>')
reasoning_end = buffer.find('</reasoning>')
ends = []
if think_end != -1:
ends.append((think_end, '</think>', 8))
if reasoning_end != -1:
ends.append((reasoning_end, '</reasoning>', 12))
if ends:
end_pos, end_tag, tag_length = min(ends, key=lambda x: x[0])
state.thought += buffer[:end_pos]
duration = time.perf_counter() - state.start_time
state.total_think_time += duration
state.in_think = False
buffer = buffer[end_pos + tag_length:]
if end_tag == '</reasoning>':
state.answer += buffer
break
else:
state.thought += buffer
break
elif state.in_answer:
answer_end = buffer.find('</answer>')
if answer_end != -1:
state.answer += buffer[:answer_end]
state.in_answer = False
buffer = buffer[answer_end + 9:]
else:
state.answer += buffer
break
elapsed = time.perf_counter() - state.start_time if state.in_think else 0
return state, elapsed
def format_response(state, elapsed):
answer_part = state.answer
collapsible = []
collapsed = "<details open>"
if state.thought or state.in_think:
if state.in_think:
total_elapsed = state.total_think_time + elapsed
formatted_time = format_time(total_elapsed)
status = f"💭 Thinking for {formatted_time}"
else:
formatted_time = format_time(state.total_think_time)
status = f"✅ Thought for {formatted_time}"
collapsed = "<details>"
collapsible.append(
f"{collapsed}<summary>{status}</summary>\n\n<div class='thinking-container'>\n{state.thought}\n</div>\n</details>"
)
return collapsible, answer_part
def remove_tags(text):
if text is None:
return None
return re.sub(r'<[^>]+>', ' ', text).strip() |