Jack1808/Claude_Code

1import re2import uuid3from enum import Enum4from typing import Any5 6from loguru import logger7 8# Some OpenAI-compatible backends/models occasionally leak internal sentinel tokens9# into `delta.content` (e.g. "<|tool_call_end|>"). These should never be shown to10# end users, and they can disrupt downstream parsing if left in place.11_CONTROL_TOKEN_RE = re.compile(r"<\|[^|>]{1,80}\|>")12_CONTROL_TOKEN_START = "<|"13_CONTROL_TOKEN_END = "|>"14 15 16class ParserState(Enum):17    TEXT = 118    MATCHING_FUNCTION = 219    PARSING_PARAMETERS = 320 21 22class HeuristicToolParser:23    """24    Stateful parser that detects raw text tool calls in the format:25    ● <function=Name><parameter=key>value</parameter>...26 27    This is used as a fallback for models that emit tool calls as text28    instead of using the structured API.29    """30 31    # Class-level compiled patterns (compiled once, not per instance)32    _FUNC_START_PATTERN = re.compile(r"●\s*<function=([^>]+)>")33    _PARAM_PATTERN = re.compile(34        r"<parameter=([^>]+)>(.*?)(?:</parameter>|$)", re.DOTALL35    )36 37    def __init__(self):38        self._state = ParserState.TEXT39        self._buffer = ""40        self._current_tool_id = None41        self._current_function_name = None42        self._current_parameters = {}43 44    def _strip_control_tokens(self, text: str) -> str:45        # Remove complete sentinel tokens. If a token is split across chunks it46        # will be removed once the buffer contains the full token.47        return _CONTROL_TOKEN_RE.sub("", text)48 49    def _split_incomplete_control_token_tail(self) -> str:50        """51        If the buffer ends with an incomplete "<|...|>" sentinel token, keep that52        fragment in the buffer and return the safe-to-emit prefix.53 54        This prevents leaking raw sentinel fragments to the user when streaming.55        """56        start = self._buffer.rfind(_CONTROL_TOKEN_START)57        if start == -1:58            return ""59        end = self._buffer.find(_CONTROL_TOKEN_END, start)60        if end != -1:61            return ""62 63        prefix = self._buffer[:start]64        self._buffer = self._buffer[start:]65        return prefix66 67    def feed(self, text: str) -> tuple[str, list[dict[str, Any]]]:68        """69        Feed text into the parser.70        Returns a tuple of (filtered_text, detected_tool_calls).71 72        filtered_text: Text that should be passed through as normal message content.73        detected_tools: List of Anthropic-format tool_use blocks.74        """75        self._buffer += text76        self._buffer = self._strip_control_tokens(self._buffer)77        detected_tools = []78        filtered_output_parts: list[str] = []79 80        while True:81            if self._state == ParserState.TEXT:82                # Look for the trigger character83                if "●" in self._buffer:84                    idx = self._buffer.find("●")85                    filtered_output_parts.append(self._buffer[:idx])86                    self._buffer = self._buffer[idx:]87                    self._state = ParserState.MATCHING_FUNCTION88                else:89                    # Avoid emitting an incomplete "<|...|>" sentinel fragment if the90                    # token got split across streaming chunks.91                    safe_prefix = self._split_incomplete_control_token_tail()92                    if safe_prefix:93                        filtered_output_parts.append(safe_prefix)94                        break95 96                    filtered_output_parts.append(self._buffer)97                    self._buffer = ""98                    break99 100            if self._state == ParserState.MATCHING_FUNCTION:101                # We need enough buffer to match the function tag102                # e.g. "● <function=Grep>"103                match = self._FUNC_START_PATTERN.search(self._buffer)104                if match:105                    self._current_function_name = match.group(1).strip()106                    self._current_tool_id = f"toolu_heuristic_{uuid.uuid4().hex[:8]}"107                    self._current_parameters = {}108 109                    # Consume the function start from buffer110                    self._buffer = self._buffer[match.end() :]111                    self._state = ParserState.PARSING_PARAMETERS112                    logger.debug(113                        "Heuristic bypass: Detected start of tool call '{}'",114                        self._current_function_name,115                    )116                else:117                    # If we have "●" but not the full tag yet, wait for more data118                    # Unless the buffer has grown too large without a match119                    if len(self._buffer) > 100:120                        # Probably not a tool call, treat as text121                        filtered_output_parts.append(self._buffer[0])122                        self._buffer = self._buffer[1:]123                        self._state = ParserState.TEXT124                    else:125                        break126 127            if self._state == ParserState.PARSING_PARAMETERS:128                # Look for parameters. We look for </parameter> to know a param is complete.129                # Or wait for another <parameter or the end of the text if it seems complete.130 131                # If we see a newline followed by anything other than <parameter or spaces,132                # we might be done with the tool call.133 134                finished_tool_call = False135 136                # Check if we have any complete parameters137                while True:138                    param_match = self._PARAM_PATTERN.search(self._buffer)139                    if param_match and "</parameter>" in param_match.group(0):140                        # Detect any content before the parameter match and preserve it141                        pre_match_text = self._buffer[: param_match.start()]142                        if pre_match_text:143                            filtered_output_parts.append(pre_match_text)144 145                        key = param_match.group(1).strip()146                        val = param_match.group(2).strip()147                        self._current_parameters[key] = val148                        self._buffer = self._buffer[param_match.end() :]149                    else:150                        break151 152                # Heuristic for completion:153                # 1. We have at least one param and we see a character that doesn't belong to the format154                # 2. Significant pause (not handled here, handled by caller via flush if needed)155                # 3. Another ● character (start of NEXT tool call)156 157                if "●" in self._buffer:158                    # Next tool call starting or something else, close current159                    # But first, capture any text before the ●160                    idx = self._buffer.find("●")161                    if idx > 0:162                        filtered_output_parts.append(self._buffer[:idx])163                        self._buffer = self._buffer[idx:]164                    finished_tool_call = True165                elif len(self._buffer) > 0 and not self._buffer.strip().startswith("<"):166                    # We have text that doesn't look like a tag, and we already parsed some or are in param state167                    # Let's see if we have trailing param starts168                    if "<parameter=" not in self._buffer:169                        # Treat the buffer as text (it's not a parameter)170                        # But wait, we are in PARSING_PARAMETERS.171                        # If we have " some text", we should emit it and finish tool call.172                        filtered_output_parts.append(self._buffer)173                        self._buffer = ""174                        finished_tool_call = True175 176                if finished_tool_call:177                    # Emit the tool call178                    detected_tools.append(179                        {180                            "type": "tool_use",181                            "id": self._current_tool_id,182                            "name": self._current_function_name,183                            "input": self._current_parameters,184                        }185                    )186                    logger.debug(187                        "Heuristic bypass: Emitting tool call '{}' with {} params",188                        self._current_function_name,189                        len(self._current_parameters),190                    )191                    self._state = ParserState.TEXT192                    # Continue loop to process remaining buffer (which is empty or starts with ●)193                else:194                    break195 196        return "".join(filtered_output_parts), detected_tools197 198    def flush(self) -> list[dict[str, Any]]:199        """200        Flush any remaining tool calls in the buffer.201        """202        self._buffer = self._strip_control_tokens(self._buffer)203        detected_tools = []204        if self._state == ParserState.PARSING_PARAMETERS:205            # Try to extract any partial parameters remaining in buffer206            # Even without </parameter>207            partial_matches = re.finditer(208                r"<parameter=([^>]+)>(.*)$", self._buffer, re.DOTALL209            )210            for m in partial_matches:211                key = m.group(1).strip()212                val = m.group(2).strip()213                self._current_parameters[key] = val214 215            detected_tools.append(216                {217                    "type": "tool_use",218                    "id": self._current_tool_id,219                    "name": self._current_function_name,220                    "input": self._current_parameters,221                }222            )223            self._state = ParserState.TEXT224            self._buffer = ""225 226        return detected_tools227