mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-10-25 16:31:29 +01:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d493558e16 |
No files matched your search
@@ -1 +1 @@
|
||||
__version__ = "0.4.38"
|
||||
__version__ = "0.4.37"
|
||||
@@ -5,7 +5,7 @@ from langgraph_sdk.encryption.types import DecryptResult, EncryptionContext
|
||||
from langgraph_sdk.priority import Priority
|
||||
from langgraph_sdk.priority.types import PriorityContext, PrioritySettings
|
||||
|
||||
__version__ = "0.4.9"
|
||||
__version__ = "0.4.8"
|
||||
|
||||
__all__ = [
|
||||
"Auth",
|
||||
|
||||
@@ -217,7 +217,7 @@ class HttpClient:
|
||||
current_headers = dict(
|
||||
request_headers if reconnect_path is None else reconnect_headers
|
||||
)
|
||||
if last_event_id is not None:
|
||||
if last_event_id:
|
||||
current_headers["Last-Event-ID"] = last_event_id
|
||||
|
||||
current_method = method if reconnect_path is None else "GET"
|
||||
|
||||
@@ -219,7 +219,7 @@ class SyncHttpClient:
|
||||
current_headers = dict(
|
||||
request_headers if reconnect_path is None else reconnect_headers
|
||||
)
|
||||
if last_event_id is not None:
|
||||
if last_event_id:
|
||||
current_headers["Last-Event-ID"] = last_event_id
|
||||
|
||||
current_method = method if reconnect_path is None else "GET"
|
||||
|
||||
@@ -79,18 +79,27 @@ class SSEDecoder:
|
||||
def __init__(self) -> None:
|
||||
self._event = ""
|
||||
self._data = bytearray()
|
||||
self._last_event_id = ""
|
||||
self._last_event_id: str | None = None
|
||||
self._retry: int | None = None
|
||||
self._at_stream_start = True
|
||||
|
||||
@property
|
||||
def last_event_id(self) -> str | None:
|
||||
"""Return the last event identifier that was seen."""
|
||||
"""Return the last event identifier that was seen.
|
||||
|
||||
return self._last_event_id or None
|
||||
`None` until the stream sends an `id` field, and `""` after an empty
|
||||
one, which resets the reconnect cursor.
|
||||
"""
|
||||
|
||||
return self._last_event_id
|
||||
|
||||
def decode(self, line: bytes) -> StreamPart | None:
|
||||
# See: https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation
|
||||
|
||||
if self._at_stream_start:
|
||||
self._at_stream_start = False
|
||||
line = line.removeprefix(b"\xef\xbb\xbf")
|
||||
|
||||
if not line:
|
||||
if (
|
||||
not self._event
|
||||
@@ -103,7 +112,7 @@ class SSEDecoder:
|
||||
sse = StreamPart(
|
||||
event=self._event,
|
||||
data=orjson.loads(self._data) if self._data else None, # ty: ignore[invalid-argument-type]
|
||||
id=self.last_event_id,
|
||||
id=self._last_event_id or None,
|
||||
)
|
||||
|
||||
# NOTE: as per the SSE spec, do not reset last_event_id.
|
||||
@@ -124,6 +133,8 @@ class SSEDecoder:
|
||||
if fieldname == b"event":
|
||||
self._event = value.decode()
|
||||
elif fieldname == b"data":
|
||||
if self._data:
|
||||
self._data.extend(b"\n")
|
||||
self._data.extend(value)
|
||||
elif fieldname == b"id":
|
||||
if b"\0" in value:
|
||||
|
||||
@@ -5,6 +5,7 @@ from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
import orjson
|
||||
import pytest
|
||||
from typing_extensions import assert_type
|
||||
|
||||
@@ -110,6 +111,25 @@ def test_stream_sse():
|
||||
assert len(parts) == 79
|
||||
|
||||
|
||||
def test_sse_decoder_joins_data_lines_with_a_newline():
|
||||
decoder = SSEDecoder()
|
||||
for line in (b"data: [1,", b"data: 2]"):
|
||||
decoder.decode(line)
|
||||
assert decoder.decode(b"") == StreamPart(event="", data=[1, 2], id=None) # ty: ignore[invalid-argument-type]
|
||||
|
||||
for line in (b"data: [1", b"data: 0]"):
|
||||
decoder.decode(line)
|
||||
with pytest.raises(orjson.JSONDecodeError):
|
||||
decoder.decode(b"")
|
||||
|
||||
|
||||
def test_sse_decoder_strips_a_leading_bom():
|
||||
decoder = SSEDecoder()
|
||||
for line in (b"\xef\xbb\xbfevent: values", b"data: 1"):
|
||||
decoder.decode(line)
|
||||
assert decoder.decode(b"") == StreamPart(event="values", data=1, id=None) # ty: ignore[invalid-argument-type]
|
||||
|
||||
|
||||
# --- HTTP client streaming ---
|
||||
|
||||
|
||||
@@ -292,6 +312,52 @@ async def test_http_client_stream_recovers_after_disconnect():
|
||||
]
|
||||
|
||||
|
||||
def _drop_after_an_empty_id(stream_cls: type, reconnects: list[httpx.Request]):
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
if request.method == "POST":
|
||||
return httpx.Response(
|
||||
200,
|
||||
headers={"Content-Type": "text/event-stream", "Location": "/reconnect"},
|
||||
stream=stream_cls(
|
||||
[
|
||||
b"id: 1\nevent: values\ndata: 1\n\n",
|
||||
b"id:\nevent: values\ndata: 2\n\n",
|
||||
],
|
||||
httpx.RemoteProtocolError("incomplete chunked read"),
|
||||
),
|
||||
)
|
||||
reconnects.append(request)
|
||||
return httpx.Response(
|
||||
200,
|
||||
headers={"Content-Type": "text/event-stream"},
|
||||
stream=stream_cls([b"event: end\ndata: null\n\n"]),
|
||||
)
|
||||
|
||||
return handler
|
||||
|
||||
|
||||
def test_sync_stream_reconnects_without_a_cursor_after_an_empty_id():
|
||||
reconnects: list[httpx.Request] = []
|
||||
transport = httpx.MockTransport(_drop_after_an_empty_id(ListByteStream, reconnects))
|
||||
with httpx.Client(transport=transport, base_url="https://example.com") as client:
|
||||
list(SyncHttpClient(client).stream("/stream", "POST", json={}))
|
||||
|
||||
assert "last-event-id" not in reconnects[0].headers
|
||||
|
||||
|
||||
async def test_async_stream_reconnects_without_a_cursor_after_an_empty_id():
|
||||
reconnects: list[httpx.Request] = []
|
||||
transport = httpx.MockTransport(
|
||||
_drop_after_an_empty_id(AsyncListByteStream, reconnects)
|
||||
)
|
||||
async with httpx.AsyncClient(
|
||||
transport=transport, base_url="https://example.com"
|
||||
) as client:
|
||||
[part async for part in HttpClient(client).stream("/stream", "POST", json={})]
|
||||
|
||||
assert "last-event-id" not in reconnects[0].headers
|
||||
|
||||
|
||||
# --- _sse_to_v2_dict conversion ---
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user