执行摘要
提取共享类型提取工具函数
减少代码重复,使 JSON Schema 类型提取逻辑可被其他工具解析器复用。原逻辑嵌入在 MinimaxM2ToolParser 类中,其他解析器无法访问。提取后 utils.py 成为共享起点。
值得阅读,展示了重构提取共享工具的实践经验。关注点:
1) 类方法转无状态函数;
2) 函数签名设计;
3) 排序确定性权衡。
审核机器人建议对返回列表排序以保证确定性顺序,作者认为当前调用者不关心顺序,拒绝了该建议。已关闭,维持无排序实现。
减少代码重复,使 JSON Schema 类型提取逻辑可被其他工具解析器复用。原逻辑嵌入在 MinimaxM2ToolParser 类中,其他解析器无法访问。提取后 utils.py 成为共享起点。
值得阅读,展示了重构提取共享工具的实践经验。关注点:
1) 类方法转无状态函数;
2) 函数签名设计;
3) 排序确定性权衡。
审核机器人建议对返回列表排序以保证确定性顺序,作者认为当前调用者不关心顺序,拒绝了该建议。已关闭,维持无排序实现。
MinimaxM2ToolParser._extract_types_from_schema 改为独立函数 extract_types_from_schema,移至 vllm/tool_parsers/utils.py。简化了 schema 为 None 或非 dict 时的判断(合并两个 if),递归逻辑不变。_parse_single_invoke 中,移除手动遍历 tools 列表的代码,改用已有的 find_tool_properties(tools, function_name) 获取参数配置。minimax_m2_tool_parser.py 中更新导入语句,新增 extract_types_from_schema 和 find_tool_properties;删除不再需要的私有方法和 from typing import Any。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 |
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
审核机器人建议对结果排序以保证确定性顺序,作者认为当前调用者不关心顺序,拒绝了该建议。
结论:不排序,维持现有行为。 · 已解决
纯重构,逻辑保持不变,回归风险低。但新函数变为公共接口,依赖工具解析器的模块可能调用,需保持稳定。测试覆盖主要 JSON Schema 结构,未覆盖 $ref 或 default 字段,未来扩展需注意。
对用户无行为影响。对开发者:其他工具解析器可直接使用 extract_types_from_schema,减少重复代码。无性能影响。
当前没有检测到明确关联的 Issue 链接,后续同步到相关引用后会出现在这里。
参与讨论