fix(agent): preserve cache keys for compatible Responses endpoints (#7236)

Forward the stable anonymized session key through the generic OpenAI-compatible Responses path so gateways can retain cache routing across tool calls.

Refs #7233
This commit is contained in:
jxxghp
2026-10-29 17:17:48 +08:00
committed by GitHub
parent b55d9bd167
commit 8fe8ea477e
3 changed files with 193 additions and 2 deletions

View File

@@ -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:

View File

@@ -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 最终绑定工具时仍然存在。

View File

@@ -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