| |
| from __future__ import annotations |
| import re |
| import json |
| import inspect |
| from ast import literal_eval |
| from logging import getLogger |
| from typing import List, Dict, Any |
| from .base_parser import BaseParser |
| from ..core import ToolCall |
|
|
| _logger = getLogger(f'sr_harness.{__name__}') |
|
|
|
|
| @BaseParser.register('text') |
| class TextParser(BaseParser): |
| """Parser for tagged text tool calls.""" |
|
|
| def format_tools(self) -> str: |
| """Format tool list into a description string for LLM. |
| |
| Returns: |
| Formatted tool description string. |
| """ |
| lines = ["## Available Tools:", ""] |
|
|
| for tool in self.tools: |
| name = tool.metadata.name |
| desc = tool.metadata.description |
|
|
| |
| try: |
| if (tool_cls := type(tool)): |
| |
| sig = inspect.signature(tool_cls.execute) |
| params = [] |
| for param_name, param in sig.parameters.items(): |
| if param_name in ('self', 'args', 'kwargs'): |
| continue |
| |
| param_parts = [] |
| param_parts.append(param_name) |
| |
| if param.annotation is not inspect.Parameter.empty: |
| type_hint = self._format_type(param.annotation) |
| param_parts.append(f": {type_hint}") |
| |
| if param.default is not inspect.Parameter.empty: |
| default_val = param.default |
| param_parts.append(f" = {default_val!r}") |
| params.append("".join(param_parts)) |
| signature = f"{name}({', '.join(params)})" |
|
|
| |
| docstring = inspect.getdoc(tool_cls.execute) or "No detailed documentation available." |
| else: |
| signature = f"{name}(...)" |
| docstring = "No detailed documentation available." |
| except Exception: |
| signature = f"{name}(...)" |
| docstring = "No detailed documentation available." |
|
|
| lines.append(f"### {name}") |
| lines.append(f"- **Description**: {desc}") |
| lines.append(f"- **Signature**: `{signature}`") |
| lines.append(f"- **DocString**: {docstring}") |
| lines.append("") |
|
|
| lines.append("## Output Format:") |
| lines.append("") |
| lines.append("Use the following format for tool calls:") |
| lines.append("") |
| lines.append("Action: tool_name(param1=value1, param2=value2)") |
| lines.append("") |
| lines.append("You can make multiple tool calls in sequence, one Action per line.") |
|
|
| return "\n".join(lines) |
|
|
| def parse_response(self, response: str) -> List[ToolCall]: |
| """Parse response. |
| |
| Args: |
| response: Provider response object. |
| |
| Returns: |
| List[ToolCall]: The operation result. |
| """ |
| tool_calls = [] |
| for line in response.strip().splitlines(): |
| line = line.strip() |
| if line.startswith('Action:'): |
| action_line = line.removeprefix('Action:').strip() |
| if (match := re.match(r'(\w+)\s*\((.*)\)', action_line)): |
| tool_name = match.group(1) |
| if (params_str := match.group(2).strip()): |
| params = self._parse_params(params_str) |
| else: |
| params = {} |
| tool_calls.append(ToolCall(name=tool_name, params=params, raw=line)) |
| else: |
| _logger.warning(f"Failed to parse action line: '{line}'") |
| return tool_calls |
|
|
| def _parse_params(self, params_str: str) -> Dict[str, Any]: |
| """解析参数字符串为字典。 |
| |
| Args: |
| params_str: 参数字符串，如 'x_vars=["phi0", "phi1"], y_var="y"' |
| |
| Returns: |
| 参数字典 |
| """ |
| try: |
| |
| if params_str.startswith('{') and params_str.endswith('}'): |
| return json.loads(params_str) |
| else: |
| |
| params_dict = {} |
| parts = self._split_params(params_str) |
| for part in parts: |
| if '=' in part: |
| key, value = part.split('=', 1) |
| key = key.strip().lower() |
| value = value.strip() |
| try: |
| params_dict[key] = literal_eval(value) |
| except (ValueError, SyntaxError): |
| params_dict[key] = value |
| return params_dict |
| except Exception as e: |
| _logger.warning(f"Failed to parse params '{params_str}': {e}") |
| return {'raw_params': params_str} |
|
|
| def _format_type(self, annotation) -> str: |
| """Format type annotation into a readable string. |
| |
| Args: |
| annotation: The type annotation to format. |
| |
| Returns: |
| Formatted type string. |
| """ |
| import typing |
| if annotation is inspect.Parameter.empty: |
| return "" |
|
|
| |
| if annotation is type(None): |
| return "None" |
| if annotation is ...: |
| return "..." |
|
|
| |
| origin = getattr(annotation, '__origin__', None) |
| if origin is not None: |
| |
| if origin is typing.Union: |
| args = getattr(annotation, '__args__', ()) |
| if type(None) in args: |
| non_none = [t for t in args if t is not type(None)] |
| if len(non_none) == 1: |
| return f"Optional[{self._format_type(non_none[0])}]" |
| |
| args = getattr(annotation, '__args__', ()) |
| args_str = ', '.join(self._format_type(arg) for arg in args) |
| name = getattr(origin, '__name__', str(origin)) |
| return f"{name}[{args_str}]" |
|
|
| |
| if hasattr(annotation, '__name__'): |
| return annotation.__name__ |
|
|
| return str(annotation) |
|
|
| def _split_params(self, params_str: str) -> List[str]: |
| """分割参数字符串，处理引号内的逗号。 |
| |
| Args: |
| params_str: 参数字符串，如 'x_vars=["phi0", "phi1"], y_var="y"' |
| |
| Returns: |
| 分割后的参数列表 |
| """ |
| parts = [] |
| current = [] |
| depth = 0 |
| in_string = False |
| string_char = None |
|
|
| for char in params_str: |
| if char in '"\'': |
| if not in_string: |
| in_string = True |
| string_char = char |
| elif char == string_char: |
| in_string = False |
| string_char = None |
| current.append(char) |
| elif char in '[{(': |
| depth += 1 |
| current.append(char) |
| elif char in ']})': |
| depth -= 1 |
| current.append(char) |
| elif char == ',' and depth == 0 and not in_string: |
| parts.append(''.join(current).strip()) |
| current = [] |
| else: |
| current.append(char) |
|
|
| if current: |
| parts.append(''.join(current).strip()) |
|
|
| return parts |
|
|
