Prhub

#43025 [Refactor] Extract extract_types_from_schema utility from Minimax M2 tool parser

原始 PR 作者 sfeng33 合并时间 2026-05-19 23:21 文件变更 3 提交数 1 评论 2 代码增减 +123 / -106

执行摘要

提取共享类型提取工具函数

减少代码重复,使 JSON Schema 类型提取逻辑可被其他工具解析器复用。原逻辑嵌入在 MinimaxM2ToolParser 类中,其他解析器无法访问。提取后 utils.py 成为共享起点。

值得阅读,展示了重构提取共享工具的实践经验。关注点:

1) 类方法转无状态函数;
2) 函数签名设计;
3) 排序确定性权衡。

讨论亮点

审核机器人建议对返回列表排序以保证确定性顺序,作者认为当前调用者不关心顺序,拒绝了该建议。已关闭,维持无排序实现。

实现拆解

  1. 提取类型提取函数:将 MinimaxM2ToolParser._extract_types_from_schema 改为独立函数 extract_types_from_schema,移至 vllm/tool_parsers/utils.py。简化了 schema 为 None 或非 dict 时的判断(合并两个 if),递归逻辑不变。
  2. 复用工具属性查找:在 _parse_single_invoke 中,移除手动遍历 tools 列表的代码,改用已有的 find_tool_properties(tools, function_name) 获取参数配置。
  3. 更新引用:在 minimax_m2_tool_parser.py 中更新导入语句,新增 extract_types_from_schemafind_tool_properties;删除不再需要的私有方法和 from typing import Any
  4. 添加单元测试:在 tests/tool_parsers/test_utils.py 中新增 TestExtractTypesFromSchema 类,覆盖直接类型、类型数组、anyOf/oneOf/allOf 递归、enum 值推断、空/非 dict 默认降级以及嵌套 anyOf。
文件 模块 状态 重要度
vllm/tool_parsers/minimax_m2_tool_parser.py 工具解析器 modified 7.36
vllm/tool_parsers/utils.py 工具解析器 modified 7.1
tests/tool_parsers/test_utils.py 测试 modified 6.39

关键符号

extract_types_from_schema _parse_single_invoke

关键源码片段

vllm/tool_parsers/minimax_m2_tool_parser.py core-logic

重构源文件:删除私有方法,更新导入,复用公共工具

# 重构后的 _parse_single_invoke 方法(简化版本)
def _parse_single_invoke(
    self, invoke_str: str, tools: list | None
) -> ToolCall | None:
    """Parse a single <invoke> block."""
    name_match = re.search(r"^([^>]+)", invoke_str)
    if not name_match:
        return None
    function_name = self._extract_name(name_match.group(1))
    # 使用公共函数 find_tool_properties 获取工具属性
    tool_properties = find_tool_properties(tools, function_name)
​
    # 提取参数
    param_dict = {}
    for match in self.parameter_complete_regex.findall(invoke_str):
        param_match = re.search(r"^([^>]+)>(.*)", match, re.DOTALL)
        if param_match:
            param_name = self._extract_name(param_match.group(1))
            param_value = param_match.group(2).strip()
            # 使用公共函数 extract_types_from_schema 获取参数类型列表
            param_types = extract_types_from_schema(
                tool_properties.get(param_name, {})
            )
            param_dict[param_name] = coerce_to_schema_type(param_value, param_types)
​
    return ToolCall(
        type="function",
        function=FunctionCall(
            name=function_name,
            arguments=json.dumps(param_dict, ensure_ascii=False),
        ),
    )
vllm/tool_parsers/utils.py core-logic

新增公共函数 extract_types_from_schema 的核心文件

# 提取所有可能的 JSON Schema 类型字符串
# 支持 type(字符串或列表)、enum 值推断、anyOf/oneOf/allOf 递归
# 当无法确定类型时返回 ["string"]
def extract_types_from_schema(schema: Any) -> list[str]:
    if schema is None or not isinstance(schema, dict):
        return ["string"]
​
    types: set[str] = set()
​
    # 处理直接的 type 字段(字符串或数组)
    if "type" in schema:
        type_value = schema["type"]
        if isinstance(type_value, str):
            types.add(type_value)
        elif isinstance(type_value, list):
            for t in type_value:
                if isinstance(t, str):
                    types.add(t)
​
    # 从 enum 列表的值推断类型
    if "enum" in schema and isinstance(schema["enum"], list) and schema["enum"]:
        for value in schema["enum"]:
            if value is None:
                types.add("null")
            elif isinstance(value, bool):
                types.add("boolean")
            elif isinstance(value, int):
                types.add("integer")
            elif isinstance(value, float):
                types.add("number")
            elif isinstance(value, str):
                types.add("string")
            elif isinstance(value, list):
                types.add("array")
            elif isinstance(value, dict):
                types.add("object")
​
    # 递归处理 anyOf / oneOf / allOf 组合
    for choice_field in ("anyOf", "oneOf", "allOf"):
        if choice_field in schema and isinstance(schema[choice_field], list):
            for choice in schema[choice_field]:
                types.update(extract_types_from_schema(choice))
​
    # 未收集到类型时默认字符串
    return list(types) if types else ["string"]
tests/tool_parsers/test_utils.py test-coverage

新增 TestExtractTypesFromSchema 测试类覆盖全面用例

# 测试类摘录,展示核心测试模式
class TestExtractTypesFromSchema:
    def test_direct_type_string(self):
        assert extract_types_from_schema({"type": "string"}) == ["string"]
​
    def test_direct_type_integer(self):
        assert extract_types_from_schema({"type": "integer"}) == ["integer"]
​
    def test_type_array(self):
        result = set(extract_types_from_schema({"type": ["string", "null"]}))
        assert result == {"string", "null"}
​
    def test_anyof(self):
        schema = {"anyOf": [{"type": "object"}, {"type": "null"}]}
        assert set(extract_types_from_schema(schema)) == {"object", "null"}
​
    def test_oneof(self):
        schema = {"oneOf": [{"type": "integer"}, {"type": "string"}]}
        assert set(extract_types_from_schema(schema)) == {"integer", "string"}
​
    def test_allof(self):
        schema = {"allOf": [{"type": "object"}]}
        assert extract_types_from_schema(schema) == ["object"]
​
    def test_enum_infers_types(self):
        schema = {"enum": [1, "a", None]}
        assert set(extract_types_from_schema(schema)) == {"integer", "string", "null"}
​
    # 更多测试包括 bool, float, list, dict, None schema, 非 dict, 空 dict, 嵌套 anyOf

评论区精华

extract_types_from_schema 返回列表排序确定性 设计

审核机器人建议对结果排序以保证确定性顺序,作者认为当前调用者不关心顺序,拒绝了该建议。

结论:不排序,维持现有行为。 · 已解决

风险与影响

纯重构,逻辑保持不变,回归风险低。但新函数变为公共接口,依赖工具解析器的模块可能调用,需保持稳定。测试覆盖主要 JSON Schema 结构,未覆盖 $ref 或 default 字段,未来扩展需注意。

对用户无行为影响。对开发者:其他工具解析器可直接使用 extract_types_from_schema,减少重复代码。无性能影响。

低风险重构 公共接口新增

关联 Issue

未识别关联 Issue

当前没有检测到明确关联的 Issue 链接,后续同步到相关引用后会出现在这里。

完整报告

参与讨论