diff --git a/app/agent/llm/helper.py b/app/agent/llm/helper.py index 149e6d37f..8ae6a66a5 100644 --- a/app/agent/llm/helper.py +++ b/app/agent/llm/helper.py @@ -565,6 +565,142 @@ def _patch_openai_responses_empty_output_support(): logger.debug("已修补 langchain-openai responses API 空 output 兼容性") +_SCHEMA_MAP_KEYWORDS = frozenset({"properties", "$defs", "definitions"}) +_SCHEMA_LIST_KEYWORDS = frozenset({"anyOf", "oneOf", "allOf"}) +_SCHEMA_NODE_KEYWORDS = frozenset({"items", "additionalProperties"}) +_SCHEMA_UNION_KEYWORDS = ("anyOf", "oneOf") + + +def _schema_variant_type(variant: Any) -> str | None: + """返回联合分支声明的单一 JSON 类型，无法确定时返回 None。""" + schema_type = variant.get("type") if isinstance(variant, dict) else None + return schema_type if isinstance(schema_type, str) else None + + +def _collapse_schema_union(schema: dict[str, Any], *, gemini_compatible: bool) -> dict[str, Any]: + """把不带顶层 type 的 anyOf/oneOf 折叠为单一分支，字段级关键字优先保留。""" + if "type" in schema: + return schema + for keyword in _SCHEMA_UNION_KEYWORDS: + variants = schema.get(keyword) + if not isinstance(variants, list): + continue + candidates = [variant for variant in variants if _schema_variant_type(variant) != "null"] + candidate_types = {_schema_variant_type(variant) for variant in candidates} + single_type = len(candidate_types) == 1 and None not in candidate_types + if not candidates or not (single_type or gemini_compatible): + return schema + chosen = next( + (variant for variant in candidates if _schema_variant_type(variant)), + candidates[0], + ) + field_keywords = {key: value for key, value in schema.items() if key != keyword} + return {**(chosen if isinstance(chosen, dict) else {}), **field_keywords} + return schema + + +def _infer_gemini_schema_type(schema: dict[str, Any]) -> dict[str, Any]: + """为 Gemini 补齐无法从联合推断类型的节点，默认 string 与原生转换一致。""" + if "type" in schema or any(key in schema for key in ("$ref", *_SCHEMA_LIST_KEYWORDS)): + return schema + if "properties" in schema: + return {**schema, "type": "object"} + if "items" in schema: + return {**schema, "type": "array"} + return {**schema, "type": "string"} + + +def _normalize_tool_schema(schema: Any, *, gemini_compatible: bool) -> Any: + """ + 返回规整后的工具参数 JSON Schema 副本，不修改入参。 + + 所有模型：anyOf/oneOf 的非 null 分支只有一种类型时折叠为该类型，保留字段级 + 描述与默认值；参数仍由工具的 Pydantic 模型校验，省略可选字段与传 null 等价。 + Gemini 模型：其函数声明要求每个节点都带 type 且不支持 additionalProperties， + 多类型联合取首个非 null 分支、丢弃 schema 形式的 additionalProperties、无法 + 推断类型的节点按 string 处理，与 langchain-google-genai 的原生转换保持一致。 + + :param schema: 工具参数 schema 节点，非字典节点原样返回 + :param gemini_compatible: 是否输出 Gemini 函数声明可接受的 schema + :return: 规整后的新 schema + """ + if not isinstance(schema, dict): + return schema + + def _normalize(child: Any) -> Any: + return _normalize_tool_schema(child, gemini_compatible=gemini_compatible) + + normalized: dict[str, Any] = {} + for key, value in schema.items(): + if key in _SCHEMA_MAP_KEYWORDS and isinstance(value, dict): + normalized[key] = {name: _normalize(child) for name, child in value.items()} + elif key in _SCHEMA_LIST_KEYWORDS and isinstance(value, list): + normalized[key] = [_normalize(child) for child in value] + elif key in _SCHEMA_NODE_KEYWORDS and isinstance(value, dict): + if gemini_compatible and key == "additionalProperties": + continue + normalized[key] = _normalize(value) + else: + normalized[key] = value + normalized = _collapse_schema_union(normalized, gemini_compatible=gemini_compatible) + return _infer_gemini_schema_type(normalized) if gemini_compatible else normalized + + +def _normalize_function_tool(tool: Any, *, gemini_compatible: bool) -> Any: + """规整 Chat Completions 或 Responses 格式的函数工具；strict 工具保持原样。""" + if not isinstance(tool, dict) or tool.get("type") != "function": + return tool + function = tool.get("function") + definition = function if isinstance(function, dict) else tool + parameters = definition.get("parameters") + if definition.get("strict") is True or not isinstance(parameters, dict): + return tool + normalized = { + **definition, + "parameters": _normalize_tool_schema(parameters, gemini_compatible=gemini_compatible), + } + return {**tool, "function": normalized} if definition is function else normalized + + +def _patch_tool_schema_request_support(model_cls: Any, *, patch_marker: str) -> None: + """ + 为 OpenAI 兼容模型在请求上线前规整函数工具的参数 schema。 + + Pydantic 为 Optional[T] 字段生成不带顶层 type 的 anyOf；把 OpenAI tools 转成 + Gemini functionDeclaration 的网关不处理 anyOf，整次请求会因字段缺少 type 被 + 400 拒绝。在请求构造出口统一规整，可覆盖主 Agent、子代理与内部模型调用的 + 全部工具，且不修改已绑定的工具定义。 + + :param model_cls: 需要修补的 ChatOpenAI 兼容模型类 + :param patch_marker: 记录已修补状态的类属性名 + """ + if getattr(model_cls, patch_marker, False): + return + original_get_request_payload = getattr(model_cls, "_get_request_payload", None) + if not callable(original_get_request_payload): + logger.debug(f"{model_cls.__name__} 缺少 _get_request_payload，跳过工具 schema 规整") + return + + @wraps(original_get_request_payload) + def _patched_get_request_payload(self: Any, input_: Any, *, stop: Any = None, **kwargs: Any) -> Any: + payload = original_get_request_payload(self, input_, stop=stop, **kwargs) + tools = payload.get("tools") if isinstance(payload, dict) else None + if not isinstance(tools, list): + return payload + model_name = getattr(self, "model_name", None) or getattr(self, "model", None) + gemini_compatible = "gemini" in str(model_name or "").lower() + return { + **payload, + "tools": [ + _normalize_function_tool(tool, gemini_compatible=gemini_compatible) + for tool in tools + ], + } + + setattr(model_cls, "_get_request_payload", _patched_get_request_payload) + setattr(model_cls, patch_marker, True) + + class LLMHelper: """LLM模型相关辅助功能""" @@ -1554,6 +1690,10 @@ class LLMHelper: from langchain_openai import ChatOpenAI _patch_openai_responses_instructions_support() + _patch_tool_schema_request_support( + ChatOpenAI, + patch_marker="_moviepilot_tool_schema_patched", + ) # ChatGPT Codex 端点强制要求 stream: True if runtime.get("use_responses_api") and "chatgpt.com/backend-api/codex" in str(runtime.get("base_url") or ""): diff --git a/tests/test_llm_tool_schema.py b/tests/test_llm_tool_schema.py new file mode 100644 index 000000000..5ae09e7c9 --- /dev/null +++ b/tests/test_llm_tool_schema.py @@ -0,0 +1,221 @@ +"""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}]") + + +def test_optional_fields_collapse_to_single_type_for_any_model() -> None: + """Optional[T] 字段折叠为 T，保留字段描述、默认值与分支约束。""" + _, tools = _request_tools(_build_model("gpt-4o")) + + 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 模型只做无损折叠，多类型联合保持原样。""" + 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"]) +def test_every_schema_node_is_typed_for_gemini_models(model_name: str) -> None: + """Gemini 模型的每个参数节点都必须带 type，且不再依赖 anyOf/oneOf。""" + _, tools = _request_tools(_build_model(model_name)) + + 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" + + +def test_strict_tools_are_left_unchanged() -> None: + """strict 工具依赖 anyOf 表达可空必填字段，规整不得改变其语义。""" + bound_tools, tools = _request_tools(_build_model("gemini-2.5-flash"), strict=True) + + assert tools == bound_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"}