mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-28 19:06:36 +08:00
298 lines
10 KiB
Python
298 lines
10 KiB
Python
"""
|
||
OpenAI 兼容请求中函数工具参数 schema 的规整契约测试。
|
||
|
||
Pydantic 为 `Optional[T]` 字段生成不带顶层 `type` 的 `anyOf`；把 OpenAI tools
|
||
转成 Gemini `functionDeclaration` 的网关不处理 `anyOf`，整次 Agent 请求会被 400
|
||
拒绝。这里用真实的 langchain-openai 请求构造与真实的 Agent 工具入参模型，固定
|
||
请求上线前的 schema 规整行为。
|
||
"""
|
||
|
||
import asyncio
|
||
import copy
|
||
import sys
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from langchain_core.messages import HumanMessage
|
||
from langchain_core.tools import StructuredTool
|
||
from langchain_openai import ChatOpenAI
|
||
|
||
from app.agent.llm import helper as llm_module
|
||
from app.agent.middleware.skills import SkillToolInput
|
||
from app.agent.middleware.subagents import _SubAgentControlInput
|
||
from app.agent.tools.impl.api import MoviePilotApiInput
|
||
from app.agent.tools.impl.view_image import ViewImageInput
|
||
|
||
_PATCH_MARKER = "_moviepilot_test_tool_schema_patched"
|
||
_TOOL_INPUTS = {
|
||
"read_skill": SkillToolInput,
|
||
"subagent": _SubAgentControlInput,
|
||
"moviepilot_api": MoviePilotApiInput,
|
||
"view_image": ViewImageInput,
|
||
}
|
||
|
||
|
||
def _build_model(model_name: str, **kwargs: Any) -> ChatOpenAI:
|
||
"""
|
||
构造带独立补丁的 ChatOpenAI 子类实例，避免修补真实类或跨用例残留。
|
||
"""
|
||
|
||
class _PatchedChatOpenAI(ChatOpenAI):
|
||
"""
|
||
承载本用例工具 schema 补丁的模型类。
|
||
"""
|
||
|
||
llm_module._patch_tool_schema_request_support(
|
||
_PatchedChatOpenAI,
|
||
patch_marker=_PATCH_MARKER,
|
||
)
|
||
return _PatchedChatOpenAI(
|
||
model=model_name,
|
||
api_key="sk-test",
|
||
base_url="https://gateway.example/v1",
|
||
**kwargs,
|
||
)
|
||
|
||
|
||
def _build_tools() -> list[StructuredTool]:
|
||
"""
|
||
用真实 Agent 工具入参模型构造 LangChain 工具。
|
||
"""
|
||
|
||
async def _noop(**_kwargs: Any) -> str:
|
||
"""
|
||
提供构造工具所需的离线执行入口。
|
||
"""
|
||
return ""
|
||
|
||
return [
|
||
StructuredTool.from_function(
|
||
coroutine=_noop,
|
||
name=name,
|
||
description=f"{name} tool",
|
||
args_schema=args_schema,
|
||
)
|
||
for name, args_schema in _TOOL_INPUTS.items()
|
||
]
|
||
|
||
|
||
def _request_tools(model: ChatOpenAI, **bind_kwargs: Any) -> tuple[list[dict], list[dict]]:
|
||
"""
|
||
返回绑定后的原始工具定义与请求 payload 中的最终工具定义。
|
||
"""
|
||
bound = model.bind_tools(_build_tools(), **bind_kwargs)
|
||
bound_tools = bound.kwargs["tools"]
|
||
snapshot = copy.deepcopy(bound_tools)
|
||
payload = model._get_request_payload([HumanMessage("hi")], **bound.kwargs)
|
||
assert bound_tools == snapshot, "规整不得修改已绑定的工具定义"
|
||
return bound_tools, payload["tools"]
|
||
|
||
|
||
def _parameters(tools: list[dict], name: str) -> dict:
|
||
"""
|
||
按名称取 Chat Completions 或 Responses 格式工具的参数 schema。
|
||
"""
|
||
for tool in tools:
|
||
function = tool.get("function", tool)
|
||
if function.get("name") == name:
|
||
return function["parameters"]
|
||
raise AssertionError(f"缺少工具 {name}")
|
||
|
||
|
||
def _iter_schema_nodes(schema: dict, path: str = ""):
|
||
"""
|
||
遍历 properties/items/additionalProperties 位置上的全部 schema 节点。
|
||
"""
|
||
for name, child in (schema.get("properties") or {}).items():
|
||
yield f"{path}.{name}", child
|
||
yield from _iter_schema_nodes(child, f"{path}.{name}")
|
||
for keyword in ("items", "additionalProperties"):
|
||
child = schema.get(keyword)
|
||
if isinstance(child, dict):
|
||
yield f"{path}[{keyword}]", child
|
||
yield from _iter_schema_nodes(child, f"{path}[{keyword}]")
|
||
|
||
|
||
@pytest.mark.parametrize("use_responses_api", [False, True], ids=["chat", "responses"])
|
||
@pytest.mark.parametrize(
|
||
"bind_kwargs",
|
||
[{}, {"strict": False}, {"strict": True}],
|
||
ids=["strict-omitted", "strict-false", "strict-true"],
|
||
)
|
||
def test_non_gemini_requests_preserve_nullable_tool_parameters(
|
||
use_responses_api: bool,
|
||
bind_kwargs: dict[str, Any],
|
||
) -> None:
|
||
"""
|
||
两种协议的非 Gemini 工具保留可空语义，不依赖上游是否自动启用 strict。
|
||
"""
|
||
model = _build_model("gpt-4o", use_responses_api=use_responses_api)
|
||
bound_tools, tools = _request_tools(model, **bind_kwargs)
|
||
|
||
expected_tools = (
|
||
[{"type": "function", **tool["function"]} for tool in bound_tools]
|
||
if use_responses_api else bound_tools
|
||
)
|
||
assert tools == expected_tools
|
||
skill = _parameters(tools, "read_skill")
|
||
assert skill["properties"]["file"]["anyOf"] == [{"type": "string"}, {"type": "null"}]
|
||
assert skill["required"] == (["name", "file"] if bind_kwargs.get("strict") else ["name"])
|
||
|
||
|
||
def test_optional_fields_collapse_to_single_type_for_gemini_models() -> None:
|
||
"""
|
||
Gemini 的 Optional[T] 字段折叠为 T，保留字段描述、默认值与分支约束。
|
||
"""
|
||
_, tools = _request_tools(_build_model("gemini-2.5-flash"))
|
||
|
||
skill = _parameters(tools, "read_skill")
|
||
assert skill["properties"]["file"] == {
|
||
"type": "string",
|
||
"default": None,
|
||
"description": SkillToolInput.model_fields["file"].description,
|
||
}
|
||
assert skill["required"] == ["name"]
|
||
|
||
subagent = _parameters(tools, "subagent")["properties"]
|
||
assert subagent["description"]["type"] == "string"
|
||
assert subagent["timeout_ms"]["type"] == "integer"
|
||
assert subagent["tasks"]["type"] == "array"
|
||
assert subagent["tasks"]["items"]["properties"]["description"]["type"] == "string"
|
||
|
||
image_data = _parameters(tools, "view_image")["properties"]["image_data"]
|
||
assert image_data["type"] == "string"
|
||
assert "anyOf" not in image_data
|
||
|
||
|
||
def test_true_unions_keep_their_semantics_for_non_gemini_models() -> None:
|
||
"""
|
||
非 Gemini 模型的多类型联合及 null 分支保持原样。
|
||
"""
|
||
bound_tools, tools = _request_tools(_build_model("gpt-4o"))
|
||
|
||
body = _parameters(tools, "moviepilot_api")["properties"]["body"]
|
||
assert body == _parameters(bound_tools, "moviepilot_api")["properties"]["body"]
|
||
assert {variant.get("type") for variant in body["anyOf"]} == {
|
||
"object",
|
||
"array",
|
||
"string",
|
||
"null",
|
||
}
|
||
|
||
|
||
@pytest.mark.parametrize("model_name", ["gemini-2.5-flash", "google/Gemini-2.5-Pro"])
|
||
@pytest.mark.parametrize("use_responses_api", [False, True], ids=["chat", "responses"])
|
||
def test_every_schema_node_is_typed_for_gemini_models(
|
||
model_name: str,
|
||
use_responses_api: bool,
|
||
) -> None:
|
||
"""
|
||
Gemini 模型的每个参数节点都必须带 type，且不再依赖 anyOf/oneOf。
|
||
"""
|
||
_, tools = _request_tools(_build_model(model_name, use_responses_api=use_responses_api))
|
||
|
||
for name in _TOOL_INPUTS:
|
||
for path, node in _iter_schema_nodes(_parameters(tools, name)):
|
||
assert isinstance(node.get("type"), str), f"{name}{path} 缺少 type: {node}"
|
||
assert "anyOf" not in node and "oneOf" not in node, f"{name}{path}"
|
||
|
||
body = _parameters(tools, "moviepilot_api")["properties"]["body"]
|
||
assert body["type"] == "object"
|
||
assert "additionalProperties" not in body
|
||
assert body["description"] == MoviePilotApiInput.model_fields["body"].description
|
||
|
||
|
||
def test_responses_api_function_tools_are_normalized() -> None:
|
||
"""
|
||
Responses API 的扁平函数工具格式同样在上线前规整。
|
||
"""
|
||
model = _build_model("gemini-2.5-flash", use_responses_api=True)
|
||
|
||
_, tools = _request_tools(model)
|
||
|
||
assert tools[0]["type"] == "function"
|
||
assert "function" not in tools[0]
|
||
assert _parameters(tools, "read_skill")["properties"]["file"]["type"] == "string"
|
||
assert _parameters(tools, "moviepilot_api")["properties"]["body"]["type"] == "object"
|
||
|
||
|
||
@pytest.mark.parametrize("use_responses_api", [False, True], ids=["chat", "responses"])
|
||
def test_strict_tools_are_left_unchanged(use_responses_api: bool) -> None:
|
||
"""
|
||
strict 工具依赖 anyOf 表达可空必填字段，规整不得改变其语义。
|
||
"""
|
||
model = _build_model("gemini-2.5-flash", use_responses_api=use_responses_api)
|
||
bound_tools, tools = _request_tools(model, strict=True)
|
||
|
||
for name in _TOOL_INPUTS:
|
||
assert _parameters(tools, name) == _parameters(bound_tools, name)
|
||
assert all(tool.get("function", tool)["strict"] is True for tool in tools)
|
||
|
||
|
||
def test_normalization_is_idempotent_when_patched_twice() -> None:
|
||
"""
|
||
重复修补同一模型类不会重复包装请求构造。
|
||
"""
|
||
|
||
class _PatchedChatOpenAI(ChatOpenAI):
|
||
"""
|
||
验证修补标记的模型类。
|
||
"""
|
||
|
||
llm_module._patch_tool_schema_request_support(_PatchedChatOpenAI, patch_marker=_PATCH_MARKER)
|
||
patched = _PatchedChatOpenAI._get_request_payload
|
||
llm_module._patch_tool_schema_request_support(_PatchedChatOpenAI, patch_marker=_PATCH_MARKER)
|
||
|
||
assert _PatchedChatOpenAI._get_request_payload is patched
|
||
|
||
|
||
def test_openai_runtime_applies_tool_schema_patch(monkeypatch) -> None:
|
||
"""
|
||
OpenAI 兼容运行时构造的模型类必须挂上工具 schema 规整。
|
||
"""
|
||
|
||
class _FakeChatOpenAI:
|
||
"""
|
||
隔离模型初始化，只保留请求补丁需要的最小接口。
|
||
"""
|
||
|
||
def __init__(self, **kwargs: Any) -> None:
|
||
"""
|
||
记录模型身份与工具调用能力。
|
||
"""
|
||
self.model_name = kwargs["model"]
|
||
self.profile = {"tool_calling": True}
|
||
|
||
def _get_request_payload(self, input_, *, stop=None, **kwargs):
|
||
"""
|
||
返回工具定义，让真实补丁处理出站参数。
|
||
"""
|
||
return {"tools": kwargs.get("tools", [])}
|
||
|
||
monkeypatch.setitem(sys.modules, "langchain_openai", SimpleNamespace(ChatOpenAI=_FakeChatOpenAI))
|
||
monkeypatch.setattr(llm_module, "_patch_openai_responses_instructions_support", lambda: None)
|
||
|
||
model = asyncio.run(
|
||
llm_module.LLMHelper.get_llm(
|
||
provider="openai",
|
||
model="gemini-2.5-flash",
|
||
api_key="sk-test",
|
||
base_url="https://gateway.example/v1",
|
||
)
|
||
)
|
||
tool = {
|
||
"type": "function",
|
||
"function": {
|
||
"name": "read_skill",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {"file": {"anyOf": [{"type": "string"}, {"type": "null"}]}},
|
||
},
|
||
},
|
||
}
|
||
|
||
payload = model._get_request_payload([], tools=[tool])
|
||
|
||
assert payload["tools"][0]["function"]["parameters"]["properties"]["file"] == {"type": "string"}
|