File size: 3,074 Bytes
d06fa57 | 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 | # Copyright (c) 2026-present, Yumeow. Licensed under the MIT License.
from __future__ import annotations
import json
from logging import getLogger
from typing import List, Dict, Any
from .base_parser import BaseParser
from ..core import ToolCall, ToolCallResult
_logger = getLogger(f'sr_harness.{__name__}')
@BaseParser.register('openai')
class OpenAIParser(BaseParser):
"""Parser for native OpenAI-compatible tool calls."""
native_tool_calls = True
def format_tools(self) -> List[Dict[str, Any]]:
"""Return OpenAI-compatible native function descriptions."""
return [
{
"type": "function",
"function": {
"name": tool.metadata.name,
"description": tool.metadata.description,
"parameters": tool.metadata.parameters,
},
}
for tool in self.tools
]
def parse_response(self, response: Any) -> List[ToolCall]:
"""Normalize provider-native function calls already extracted by BaseAPI."""
normalized = []
for tool_call in response or []:
if isinstance(tool_call, dict):
raw = tool_call
elif hasattr(tool_call, "to_dict"):
raw = tool_call.to_dict()
elif hasattr(tool_call, "model_dump"):
raw = tool_call.model_dump()
else:
raise ValueError(f"Unrecognized tool call format: {tool_call}")
function = raw.get("function") or {}
name = function.get("name") or raw.get("name") or None
params = (
function.get("arguments")
or raw.get("arguments")
or raw.get("arguments_json")
or raw.get("args")
or {}
)
try:
if isinstance(params, str):
params = json.loads(params) if params.strip() else {}
except json.JSONDecodeError as error:
_logger.warning(f"Skip malformed native tool call {raw!r}: {error}")
continue
normalized.append(ToolCall(
name=name,
params=params,
id=raw.get("id") or raw.get("call_id"),
raw=raw,
))
return normalized
def format_tool_result_messages(
self,
tool_calls: List[ToolCall],
results: List[ToolCallResult],
) -> List[Dict[str, Any]]:
"""Format tool result messages.
Args:
tool_calls: Tool calls returned by the model.
results: Result records to process.
Returns:
List[Dict[str, Any]]: The operation result.
"""
messages = []
for tool_call, result in zip(tool_calls, results):
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"name": tool_call.name,
"content": result.result_str,
})
return messages
|