diff --git a/app/agent/llm/helper.py b/app/agent/llm/helper.py index 9b07abafc..aaea553f0 100644 --- a/app/agent/llm/helper.py +++ b/app/agent/llm/helper.py @@ -1321,7 +1321,15 @@ class LLMHelper: model_kwargs: dict[str, Any], ) -> tuple[dict[str, str] | None, dict[str, Any]]: """ - 为官方端点构造稳定会话及提示词缓存路由参数。 + 为官方端点及通用 OpenAI 兼容 Responses 构造稳定缓存路由参数。 + + :param provider: 提供商标识；openai 对应用户配置的通用兼容入口 + :param base_url: 实际 API 地址，用于识别供应商专属协议 + :param use_responses_api: 本次实际是否使用 Responses；None 表示未强制指定 + :param prompt_cache_key: 稳定且脱敏的会话键；空值不添加缓存参数 + :param default_headers: 原始请求头，不原地修改 + :param model_kwargs: 原始模型参数，不原地修改其 extra_body + :return: 合入会话路由参数的请求头与模型参数副本 """ cache_key = str(prompt_cache_key or "").strip() headers = dict(default_headers or {}) @@ -1344,7 +1352,10 @@ class LLMHelper: base_url, "api.x.ai", ) - if not is_openai and not is_xai: + # Responses 的标准缓存键也供 Sub2API 等网关维持会话路由； + # 通用 Chat Completions 仍保留原有兼容边界，避免向旧服务增加参数。 + is_compatible_responses = provider_name == "openai" and use_responses_api is True + if not is_openai and not is_xai and not is_compatible_responses: return headers or None, kwargs if is_xai and use_responses_api is not True: diff --git a/tests/test_agent_prompt_cache.py b/tests/test_agent_prompt_cache.py index d59324163..bc07f324a 100644 --- a/tests/test_agent_prompt_cache.py +++ b/tests/test_agent_prompt_cache.py @@ -170,6 +170,38 @@ def test_xai_chat_completions_uses_stable_conversation_header() -> None: assert kwargs == {} +@pytest.mark.parametrize("provider", ["openai", "chatgpt", "xai", "custom"]) +@pytest.mark.parametrize("use_responses_api", [True, False, None]) +@pytest.mark.parametrize("cache_key", [" cache-key ", None, " "]) +def test_compatible_responses_cache_options( + provider: str, use_responses_api: bool | None, cache_key: str | None, +) -> None: + """ + OpenAI 兼容 Responses 应透传非空缓存键，其余第三方协议保持原合同。 + + :param provider: 提供商标识，用于区分通用兼容入口与专属适配器 + :param use_responses_api: 本次模型实际使用的协议；None 表示未强制指定 + :param cache_key: 会话缓存键，空值不得创建额外路由参数 + """ + original_headers = {"User-Agent": "MoviePilot"} + original_kwargs = {"extra_body": {"existing": True}} + headers, kwargs = LLMHelper._build_openai_prompt_cache_options( + provider=provider, + base_url="https://sub2api.example/v1", + use_responses_api=use_responses_api, + prompt_cache_key=cache_key, + default_headers=original_headers, + model_kwargs=original_kwargs, + ) + + expected: dict[str, bool | str] = {"existing": True} + if provider == "openai" and use_responses_api is True and cache_key and cache_key.strip(): + expected["prompt_cache_key"] = cache_key.strip() + assert headers == original_headers + assert kwargs == {"extra_body": expected} + assert original_kwargs == {"extra_body": {"existing": True}} + + def test_prompt_cache_adapter_preserves_control_after_tool_binding() -> None: """ Provider 缓存参数应在 Agent 最终绑定工具时仍然存在。 diff --git a/tests/test_llm_responses_cache.py b/tests/test_llm_responses_cache.py new file mode 100644 index 000000000..fce7f94fd --- /dev/null +++ b/tests/test_llm_responses_cache.py @@ -0,0 +1,148 @@ +""" +通过真实 LangChain、OpenAI SDK 和离线传输验证连续工具调用的会话缓存路由。 +""" + +import asyncio +import json +from typing import Any +from unittest.mock import AsyncMock + +import httpx +import pytest +from langchain.agents import create_agent +from langchain_core.messages import HumanMessage + +from app.agent.llm import helper +from app.agent.llm.helper import LLMHelper +from app.agent.orchestrator import MoviePilotAgent + + +def _response(turn: int, model: str) -> dict[str, Any]: + """ + 生成两次工具调用及最终答复，响应中不伪造缓存命中率。 + + :param turn: 当前离线模型请求序号，从 0 开始 + :param model: 请求使用的模型名称 + :return: 可由 OpenAI SDK 解析的 Responses 响应 + """ + output: dict[str, Any] + if turn < 2: + output = { + "type": "function_call", "id": f"fc_{turn}", "call_id": f"call_{turn}", + "name": "check_status", "arguments": json.dumps({"target": f"step-{turn}"}), + "status": "completed", + } + else: + output = { + "type": "message", "id": "msg_done", "role": "assistant", "status": "completed", + "content": [{"type": "output_text", "text": "完成。", "annotations": []}], + } + return { + "id": f"resp_{turn}", "object": "response", "created_at": 0, + "status": "completed", "model": model, "output": [output], + } + + +def _stream_response(response: dict[str, Any]) -> str: + """ + 将同一工具响应转换为 SDK 实际消费的 SSE 事件。 + + :param response: 含完整输出条目的离线响应 + :return: 包含创建、输出和完成事件的 SSE 正文 + """ + item = response["output"][0] + events = [ + {"type": "response.created", "response": {**response, "output": []}}, + {"type": "response.output_item.added", "output_index": 0, "item": item}, + ] + if item["type"] == "message": + events.append({ + "type": "response.output_text.delta", "output_index": 0, "content_index": 0, + "item_id": item["id"], "delta": item["content"][0]["text"], + }) + events.append({"type": "response.completed", "response": response}) + return "".join(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n" for event in events) + + +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("model_name", ["gpt-6-luna", "gpt-6.1-sol"]) +def test_compatible_responses_preserves_cache_key_in_tool_loop( + monkeypatch: pytest.MonkeyPatch, streaming: bool, model_name: str, +) -> None: + """ + 真实 Agent 图的三次 SDK 请求应携带同一脱敏键、稳定工具和递增消息前缀。 + + :param monkeypatch: 将 HTTP 客户端和提供商发现限制到离线边界的夹具 + :param streaming: 分别覆盖流式和非流式 Responses 请求 + :param model_name: Issue 报告使用的模型，不依赖模型名称白名单 + """ + requests: list[httpx.Request] = [] + payloads: list[dict[str, Any]] = [] + targets: list[str] = [] + + def respond(request: httpx.Request) -> httpx.Response: + """ + 捕获真实 SDK 请求并提供确定性响应，禁止访问实际模型服务。 + + :param request: OpenAI SDK 完成序列化后的 HTTP 请求 + :return: 与请求流式设置对应的离线 HTTP 响应 + """ + payload = json.loads(request.content) + response = _response(len(requests), model_name) + requests.append(request) + payloads.append(payload) + if payload.get("stream"): + return httpx.Response(200, text=_stream_response(response), headers={"Content-Type": "text/event-stream"}) + return httpx.Response(200, json=response) + + def check_status(target: str) -> str: + """ + 记录图中真实执行的工具步骤。 + + :param target: 模型指定的离线步骤名称 + :return: 该步骤的确定性回执 + """ + targets.append(target) + return f"{target} 已完成" + + agent = MoviePilotAgent(session_id="private-session", user_id="private-user") + cache_key = agent._build_prompt_cache_key() + transport = httpx.MockTransport(respond) + with httpx.Client(transport=transport) as client: + async def invoke() -> Any: + """ + 创建真实模型并运行两次工具调用，客户端仅属于本次测试。 + + :return: Agent 图最终状态 + """ + async with httpx.AsyncClient(transport=transport) as async_client: + monkeypatch.setattr( + helper, "_build_httpx_client", + lambda _proxy, **kwargs: async_client if kwargs.get("async_client") else client, + ) + runtime = AsyncMock() + runtime.resolve_runtime.return_value = { + "runtime": "openai_compatible", "model_id": model_name, + "api_key": "test-key", "base_url": "https://sub2api.example/v1", + } + model = await LLMHelper.get_llm( + provider="openai", model=model_name, user_agent="", use_proxy=False, + api_protocol="responses", web_search_mode="disabled", streaming=streaming, + prompt_cache_key=cache_key, provider_runtime=runtime, + ) + graph = create_agent(model=model, tools=[check_status], system_prompt="执行用户任务并核对工具结果。") + return await graph.ainvoke({"messages": [HumanMessage(content="连续检查两个步骤。")]}) + + result = asyncio.run(invoke()) + + assert result["messages"][-1].text == "完成。" + assert targets == ["step-0", "step-1"] + assert len(requests) == 3 + assert all(request.url.path == "/v1/responses" for request in requests) + assert all(payload.get("prompt_cache_key") == cache_key for payload in payloads) + assert all(payload["tools"] == payloads[0]["tools"] for payload in payloads) + for previous, current in zip(payloads, payloads[1:]): + assert current["input"][:len(previous["input"])] == previous["input"] + assert len(current["input"]) > len(previous["input"]) + assert "private-session" not in cache_key + assert "private-user" not in cache_key