mirror of
https://github.com/langchain-ai/langchain.git
synced 2026-10-06 01:45:27 +03:00
feat(openai): extract gateway metadata from response headers when available (#39706)
Extracts gateway metadata from response headers when it's included. --------- Co-authored-by: Chester Curme <chester.curme@gmail.com>
This commit is contained in:
1 parent
9c21d84bcb
commit
a2ff1bb2ed
4 files changed
+255
-16
No files matched your search
@@ -108,7 +108,11 @@ from langchain_core.runnables.config import run_in_executor
|
||||
from langchain_core.tools import BaseTool
|
||||
from langchain_core.tools.base import _stringify
|
||||
from langchain_core.utils import get_pydantic_field_names
|
||||
from langchain_core.utils._gateway import _resolve_gateway_config
|
||||
from langchain_core.utils._gateway import (
|
||||
GATEWAY_METADATA_RESPONSE_KEY,
|
||||
_parse_gateway_metadata,
|
||||
_resolve_gateway_config,
|
||||
)
|
||||
from langchain_core.utils.function_calling import (
|
||||
convert_to_openai_function,
|
||||
convert_to_openai_tool,
|
||||
@@ -602,6 +606,21 @@ def _handle_openai_api_error(e: openai.APIError) -> None:
|
||||
raise
|
||||
|
||||
|
||||
def _add_gateway_metadata(generation_info: dict[str, Any], raw_response: Any) -> None:
|
||||
"""Add parsed LangSmith gateway metadata to `generation_info`, if present.
|
||||
|
||||
Args:
|
||||
generation_info: Generation info to mutate in place.
|
||||
raw_response: The raw provider response, or None.
|
||||
"""
|
||||
headers = getattr(raw_response, "headers", None)
|
||||
if headers is None:
|
||||
return
|
||||
gateway_metadata = _parse_gateway_metadata(headers)
|
||||
if gateway_metadata is not None:
|
||||
generation_info[GATEWAY_METADATA_RESPONSE_KEY] = gateway_metadata
|
||||
|
||||
|
||||
_RESPONSES_API_ONLY_PREFIXES = (
|
||||
"gpt-5-pro",
|
||||
"gpt-5.2-pro",
|
||||
@@ -1121,6 +1140,19 @@ class BaseChatOpenAI(BaseChatModel):
|
||||
|
||||
model_config = ConfigDict(populate_by_name=True)
|
||||
|
||||
@property
|
||||
def _uses_gateway(self) -> bool:
|
||||
"""Whether requests are routed through the LangSmith gateway.
|
||||
|
||||
Detected from the resolved API key: LangSmith keys (used to authenticate
|
||||
to the gateway) carry the `lsv2_` prefix. Callable keys cannot be
|
||||
inspected without invoking them, so they are treated as non-gateway.
|
||||
"""
|
||||
api_key = self.openai_api_key
|
||||
if isinstance(api_key, SecretStr):
|
||||
return api_key.get_secret_value().startswith("lsv2_")
|
||||
return False
|
||||
|
||||
@property
|
||||
def model(self) -> str:
|
||||
"""Same as model_name."""
|
||||
@@ -1497,16 +1529,19 @@ class BaseChatOpenAI(BaseChatModel):
|
||||
self._ensure_sync_client_available()
|
||||
kwargs["stream"] = True
|
||||
payload = self._get_request_payload(messages, stop=stop, **kwargs)
|
||||
headers: dict = {}
|
||||
base_generation_info: dict = {}
|
||||
try:
|
||||
if self.include_response_headers:
|
||||
if self.include_response_headers or self._uses_gateway:
|
||||
raw_context_manager = (
|
||||
self.root_client.with_raw_response.responses.create(**payload)
|
||||
)
|
||||
context_manager = raw_context_manager.parse()
|
||||
headers = {"headers": dict(raw_context_manager.headers)}
|
||||
if self.include_response_headers:
|
||||
headers = {"headers": dict(raw_context_manager.headers)}
|
||||
_add_gateway_metadata(base_generation_info, raw_context_manager)
|
||||
else:
|
||||
context_manager = self.root_client.responses.create(**payload)
|
||||
headers = {}
|
||||
original_schema_obj = kwargs.get("response_format")
|
||||
|
||||
with context_manager as response:
|
||||
@@ -1533,6 +1568,11 @@ class BaseChatOpenAI(BaseChatModel):
|
||||
output_version=self.output_version,
|
||||
)
|
||||
if generation_chunk:
|
||||
if is_first_chunk and base_generation_info:
|
||||
generation_chunk.generation_info = {
|
||||
**base_generation_info,
|
||||
**(generation_chunk.generation_info or {}),
|
||||
}
|
||||
if run_manager:
|
||||
run_manager.on_llm_new_token(
|
||||
generation_chunk.text, chunk=generation_chunk
|
||||
@@ -1555,20 +1595,23 @@ class BaseChatOpenAI(BaseChatModel):
|
||||
) -> AsyncIterator[ChatGenerationChunk]:
|
||||
kwargs["stream"] = True
|
||||
payload = self._get_request_payload(messages, stop=stop, **kwargs)
|
||||
headers: dict = {}
|
||||
base_generation_info: dict = {}
|
||||
try:
|
||||
if self.include_response_headers:
|
||||
if self.include_response_headers or self._uses_gateway:
|
||||
raw_context_manager = (
|
||||
await self.root_async_client.with_raw_response.responses.create(
|
||||
**payload
|
||||
)
|
||||
)
|
||||
context_manager = raw_context_manager.parse()
|
||||
headers = {"headers": dict(raw_context_manager.headers)}
|
||||
if self.include_response_headers:
|
||||
headers = {"headers": dict(raw_context_manager.headers)}
|
||||
_add_gateway_metadata(base_generation_info, raw_context_manager)
|
||||
else:
|
||||
context_manager = await self.root_async_client.responses.create(
|
||||
**payload
|
||||
)
|
||||
headers = {}
|
||||
original_schema_obj = kwargs.get("response_format")
|
||||
|
||||
async with context_manager as response:
|
||||
@@ -1599,6 +1642,11 @@ class BaseChatOpenAI(BaseChatModel):
|
||||
output_version=self.output_version,
|
||||
)
|
||||
if generation_chunk:
|
||||
if is_first_chunk and base_generation_info:
|
||||
generation_chunk.generation_info = {
|
||||
**base_generation_info,
|
||||
**(generation_chunk.generation_info or {}),
|
||||
}
|
||||
if run_manager:
|
||||
await run_manager.on_llm_new_token(
|
||||
generation_chunk.text, chunk=generation_chunk
|
||||
@@ -1662,10 +1710,12 @@ class BaseChatOpenAI(BaseChatModel):
|
||||
)
|
||||
context_manager = response_stream
|
||||
else:
|
||||
if self.include_response_headers:
|
||||
if self.include_response_headers or self._uses_gateway:
|
||||
raw_response = self.client.with_raw_response.create(**payload)
|
||||
response = raw_response.parse()
|
||||
base_generation_info = {"headers": dict(raw_response.headers)}
|
||||
if self.include_response_headers:
|
||||
base_generation_info = {"headers": dict(raw_response.headers)}
|
||||
_add_gateway_metadata(base_generation_info, raw_response)
|
||||
else:
|
||||
response = self.client.create(**payload)
|
||||
context_manager = response
|
||||
@@ -1737,12 +1787,26 @@ class BaseChatOpenAI(BaseChatModel):
|
||||
response = raw_response.parse()
|
||||
if self.include_response_headers:
|
||||
generation_info = {"headers": dict(raw_response.headers)}
|
||||
return _construct_lc_result_from_responses_api(
|
||||
generation_info = generation_info or {}
|
||||
_add_gateway_metadata(generation_info, raw_response)
|
||||
# Gateway metadata belongs on `generation_info`, not the message
|
||||
# `response_metadata` that `metadata` populates.
|
||||
gateway_metadata = generation_info.pop(
|
||||
GATEWAY_METADATA_RESPONSE_KEY, None
|
||||
)
|
||||
result = _construct_lc_result_from_responses_api(
|
||||
response,
|
||||
schema=original_schema_obj,
|
||||
metadata=generation_info,
|
||||
output_version=self.output_version,
|
||||
)
|
||||
if gateway_metadata is not None:
|
||||
for generation in result.generations:
|
||||
generation.generation_info = generation.generation_info or {}
|
||||
generation.generation_info[GATEWAY_METADATA_RESPONSE_KEY] = (
|
||||
gateway_metadata
|
||||
)
|
||||
return result
|
||||
else:
|
||||
raw_response = self.client.with_raw_response.create(**payload)
|
||||
response = raw_response.parse()
|
||||
@@ -1760,6 +1824,8 @@ class BaseChatOpenAI(BaseChatModel):
|
||||
and hasattr(raw_response, "headers")
|
||||
):
|
||||
generation_info = {"headers": dict(raw_response.headers)}
|
||||
generation_info = generation_info or {}
|
||||
_add_gateway_metadata(generation_info, raw_response)
|
||||
return self._create_chat_result(response, generation_info)
|
||||
|
||||
def _use_responses_api(self, payload: dict) -> bool:
|
||||
@@ -1922,12 +1988,14 @@ class BaseChatOpenAI(BaseChatModel):
|
||||
)
|
||||
context_manager = response_stream
|
||||
else:
|
||||
if self.include_response_headers:
|
||||
if self.include_response_headers or self._uses_gateway:
|
||||
raw_response = await self.async_client.with_raw_response.create(
|
||||
**payload
|
||||
)
|
||||
response = raw_response.parse()
|
||||
base_generation_info = {"headers": dict(raw_response.headers)}
|
||||
if self.include_response_headers:
|
||||
base_generation_info = {"headers": dict(raw_response.headers)}
|
||||
_add_gateway_metadata(base_generation_info, raw_response)
|
||||
else:
|
||||
response = await self.async_client.create(**payload)
|
||||
context_manager = response
|
||||
@@ -1980,7 +2048,7 @@ class BaseChatOpenAI(BaseChatModel):
|
||||
**kwargs: Any,
|
||||
) -> ChatResult:
|
||||
payload = self._get_request_payload(messages, stop=stop, **kwargs)
|
||||
generation_info = None
|
||||
generation_info = {}
|
||||
raw_response = None
|
||||
try:
|
||||
if "response_format" in payload:
|
||||
@@ -2006,12 +2074,25 @@ class BaseChatOpenAI(BaseChatModel):
|
||||
response = raw_response.parse()
|
||||
if self.include_response_headers:
|
||||
generation_info = {"headers": dict(raw_response.headers)}
|
||||
return _construct_lc_result_from_responses_api(
|
||||
_add_gateway_metadata(generation_info, raw_response)
|
||||
# Gateway metadata belongs on `generation_info`, not the message
|
||||
# `response_metadata` that `metadata` populates.
|
||||
gateway_metadata = generation_info.pop(
|
||||
GATEWAY_METADATA_RESPONSE_KEY, None
|
||||
)
|
||||
result = _construct_lc_result_from_responses_api(
|
||||
response,
|
||||
schema=original_schema_obj,
|
||||
metadata=generation_info,
|
||||
output_version=self.output_version,
|
||||
)
|
||||
if gateway_metadata is not None:
|
||||
for generation in result.generations:
|
||||
generation.generation_info = generation.generation_info or {}
|
||||
generation.generation_info[GATEWAY_METADATA_RESPONSE_KEY] = (
|
||||
gateway_metadata
|
||||
)
|
||||
return result
|
||||
else:
|
||||
raw_response = await self.async_client.with_raw_response.create(
|
||||
**payload
|
||||
@@ -2031,6 +2112,7 @@ class BaseChatOpenAI(BaseChatModel):
|
||||
and hasattr(raw_response, "headers")
|
||||
):
|
||||
generation_info = {"headers": dict(raw_response.headers)}
|
||||
_add_gateway_metadata(generation_info, raw_response)
|
||||
return await run_in_executor(
|
||||
None, self._create_chat_result, response, generation_info
|
||||
)
|
||||
|
||||
@@ -23,7 +23,7 @@ classifiers = [
|
||||
version = "1.5.1"
|
||||
requires-python = ">=3.10.0,<4.0.0"
|
||||
dependencies = [
|
||||
"langchain-core>=1.5.4,<2.0.0",
|
||||
"langchain-core>=1.5.6,<2.0.0",
|
||||
"certifi>=2024.6.2",
|
||||
"openai>=2.45.0,<4.0.0",
|
||||
"tiktoken>=0.7.0,<1.0.0",
|
||||
|
||||
@@ -10,6 +10,7 @@ from typing import Any, Literal, cast
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import httpx
|
||||
import httpx2
|
||||
import openai
|
||||
import pytest
|
||||
from langchain_core.exceptions import ContextOverflowError
|
||||
@@ -36,12 +37,14 @@ from langchain_core.runnables import RunnableLambda
|
||||
from langchain_core.runnables.base import RunnableBinding, RunnableSequence
|
||||
from langchain_core.tracers.base import BaseTracer
|
||||
from langchain_core.tracers.schemas import Run
|
||||
from langchain_core.utils._gateway import GATEWAY_METADATA_RESPONSE_KEY
|
||||
from langchain_core.utils.pydantic import PYDANTIC_VERSION
|
||||
from openai.types.responses import (
|
||||
ResponseApplyPatchToolCall,
|
||||
ResponseApplyPatchToolCallOutput,
|
||||
ResponseOutputMessage,
|
||||
ResponseReasoningItem,
|
||||
ResponseTextDeltaEvent,
|
||||
)
|
||||
from openai.types.responses.response import IncompleteDetails, Response
|
||||
from openai.types.responses.response_apply_patch_tool_call import OperationCreateFile
|
||||
@@ -722,6 +725,160 @@ async def test_openai_ainvoke(mock_async_client: AsyncMock) -> None:
|
||||
assert mock_async_client.with_raw_response.create.called
|
||||
|
||||
|
||||
class _GatewayMetadataTracer(BaseTracer):
|
||||
"""Captures gateway metadata promoted onto completed LLM runs."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.gateway_metadata: dict | None = None
|
||||
|
||||
def _persist_run(self, run: Run) -> None:
|
||||
"""No-op; runs are inspected as they complete."""
|
||||
|
||||
def _on_llm_end(self, run: Run) -> None:
|
||||
metadata = run.extra.get("metadata", {})
|
||||
if "ls_gateway_info" in metadata:
|
||||
self.gateway_metadata = metadata["ls_gateway_info"]
|
||||
|
||||
|
||||
_GATEWAY_METADATA_HEADERS = httpx2.Headers(
|
||||
{"x-langsmith-gateway-metadata": '{"provider": "openai"}'}
|
||||
)
|
||||
|
||||
_RESPONSES_API_COMPLETION = Response(
|
||||
id="resp_123",
|
||||
created_at=1234567890,
|
||||
model=OPENAI_TEST_MODEL,
|
||||
object="response",
|
||||
parallel_tool_calls=True,
|
||||
tools=[],
|
||||
tool_choice="auto",
|
||||
output=[
|
||||
ResponseOutputMessage(
|
||||
type="message",
|
||||
id="msg_123",
|
||||
content=[
|
||||
ResponseOutputText(type="output_text", text="Bar Baz", annotations=[])
|
||||
],
|
||||
role="assistant",
|
||||
status="completed",
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
_RESPONSES_API_STREAM = [
|
||||
ResponseTextDeltaEvent(
|
||||
content_index=0,
|
||||
delta="Bar Baz",
|
||||
item_id="msg_123",
|
||||
output_index=0,
|
||||
sequence_number=0,
|
||||
logprobs=[],
|
||||
type="response.output_text.delta",
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("use_responses_api", [False, True])
|
||||
def test_openai_invoke_surfaces_gateway_metadata(
|
||||
mock_completion: dict, *, use_responses_api: bool
|
||||
) -> None:
|
||||
"""Gateway metadata header is surfaced on `generation_info`, not the message."""
|
||||
llm = ChatOpenAI(use_responses_api=use_responses_api)
|
||||
mock_client = MagicMock()
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.headers = _GATEWAY_METADATA_HEADERS
|
||||
if use_responses_api:
|
||||
mock_resp.parse.return_value = _RESPONSES_API_COMPLETION
|
||||
mock_client.responses.with_raw_response.create.return_value = mock_resp
|
||||
client_attr = "root_client"
|
||||
else:
|
||||
mock_resp.parse.return_value = mock_completion
|
||||
mock_client.with_raw_response.create.return_value = mock_resp
|
||||
client_attr = "client"
|
||||
|
||||
tracer = _GatewayMetadataTracer()
|
||||
with patch.object(llm, client_attr, mock_client):
|
||||
res = llm.invoke("bar", config={"callbacks": [tracer]})
|
||||
|
||||
# Gateway metadata reaches the tracer via `generation_info`...
|
||||
assert tracer.gateway_metadata == {"provider": "openai"}
|
||||
# ...but is kept off the user-facing message `response_metadata`.
|
||||
assert GATEWAY_METADATA_RESPONSE_KEY not in res.response_metadata
|
||||
|
||||
|
||||
@pytest.mark.parametrize("use_responses_api", [False, True])
|
||||
def test_openai_stream_surfaces_gateway_metadata(
|
||||
mock_openai_completion: list, *, use_responses_api: bool
|
||||
) -> None:
|
||||
"""Gateway metadata reaches the tracer for a gateway-routed stream."""
|
||||
# A LangSmith API key signals gateway routing, so streaming fetches raw
|
||||
# headers.
|
||||
llm = ChatOpenAI(
|
||||
model=OPENAI_TEST_MODEL,
|
||||
api_key="lsv2_pt_example", # type: ignore[arg-type]
|
||||
use_responses_api=use_responses_api,
|
||||
)
|
||||
mock_client = MagicMock()
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.headers = _GATEWAY_METADATA_HEADERS
|
||||
if use_responses_api:
|
||||
mock_resp.parse.return_value = MockSyncContextManager(_RESPONSES_API_STREAM)
|
||||
mock_client.with_raw_response.responses.create.return_value = mock_resp
|
||||
mock_client.responses.create.return_value = MockSyncContextManager(
|
||||
_RESPONSES_API_STREAM
|
||||
)
|
||||
client_attr = "root_client"
|
||||
else:
|
||||
mock_resp.parse.return_value = MockSyncContextManager(mock_openai_completion)
|
||||
mock_client.with_raw_response.create.return_value = mock_resp
|
||||
client_attr = "client"
|
||||
|
||||
tracer = _GatewayMetadataTracer()
|
||||
with patch.object(llm, client_attr, mock_client):
|
||||
for chunk in llm.stream("what is your name?", config={"callbacks": [tracer]}):
|
||||
# Gateway metadata is kept off the user-facing chunk metadata.
|
||||
assert GATEWAY_METADATA_RESPONSE_KEY not in chunk.response_metadata
|
||||
|
||||
assert tracer.gateway_metadata == {"provider": "openai"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("use_responses_api", [False, True])
|
||||
async def test_openai_astream_surfaces_gateway_metadata(
|
||||
mock_openai_completion: list, *, use_responses_api: bool
|
||||
) -> None:
|
||||
"""Gateway metadata reaches the tracer for a gateway-routed async stream."""
|
||||
llm = ChatOpenAI(
|
||||
model=OPENAI_TEST_MODEL,
|
||||
api_key="lsv2_pt_example", # type: ignore[arg-type]
|
||||
use_responses_api=use_responses_api,
|
||||
)
|
||||
mock_client = AsyncMock()
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.headers = _GATEWAY_METADATA_HEADERS
|
||||
if use_responses_api:
|
||||
mock_resp.parse.return_value = MockAsyncContextManager(_RESPONSES_API_STREAM)
|
||||
mock_client.with_raw_response.responses.create.return_value = mock_resp
|
||||
mock_client.responses.create.return_value = MockAsyncContextManager(
|
||||
_RESPONSES_API_STREAM
|
||||
)
|
||||
client_attr = "root_async_client"
|
||||
else:
|
||||
mock_resp.parse.return_value = MockAsyncContextManager(mock_openai_completion)
|
||||
mock_client.with_raw_response.create.return_value = mock_resp
|
||||
client_attr = "async_client"
|
||||
|
||||
tracer = _GatewayMetadataTracer()
|
||||
with patch.object(llm, client_attr, mock_client):
|
||||
async for chunk in llm.astream(
|
||||
"what is your name?", config={"callbacks": [tracer]}
|
||||
):
|
||||
# Gateway metadata is kept off the user-facing chunk metadata.
|
||||
assert GATEWAY_METADATA_RESPONSE_KEY not in chunk.response_metadata
|
||||
|
||||
assert tracer.gateway_metadata == {"provider": "openai"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"model",
|
||||
[
|
||||
|
||||
Generated
+1
-1
@@ -707,7 +707,7 @@ typing = [
|
||||
|
||||
[[package]]
|
||||
name = "langchain-core"
|
||||
version = "1.5.4"
|
||||
version = "1.5.6"
|
||||
source = { editable = "../../core" }
|
||||
dependencies = [
|
||||
{ name = "httpx" },
|
||||
|
||||
Reference in new issue
Block a user