mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-10-04 18:22:10 +00:00
fix: stream agent tokens to the Agents UI (#1329)
* fix(dashboard): stream agent tokens to the Agents UI
The dashboard SSE endpoint was serializing the SDK's `StreamPart`
namedtuple via `str(part)`, so every event arrived at the browser
with `event = "StreamPart(event='values', data=..., id=None)"`. The
frontend's `event.startsWith("messages")` branch never fired and
every event fell through to a query refetch — which is why tool
calls appeared to stream (state refetch) but assistant text did not.
Also opt the run into `messages-tuple` stream mode so the server
actually buffers per-token `AIMessageChunk` events, and forward
`Last-Event-ID` so reconnects resume the in-flight run instead of
restarting from `seq=0`.
* fix(dashboard): drop unsupported stream_mode arg from join_stream
threads.join_stream's stream_mode is a ThreadStreamMode literal
(run_modes/lifecycle/state_update); passing run-level modes is
invalid. The default 'run_modes' already replays the modes the
run was created with, which we set correctly on runs.create.
---------
Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
This commit is contained in:
parent
ccea80b887
commit
72252e48c9
2 changed files with 30 additions and 5 deletions
|
|
@ -627,10 +627,15 @@ async def api_delete_thread(
|
||||||
@router.get("/threads/{thread_id}/stream")
|
@router.get("/threads/{thread_id}/stream")
|
||||||
async def api_stream_thread(
|
async def api_stream_thread(
|
||||||
thread_id: str,
|
thread_id: str,
|
||||||
|
request: Request,
|
||||||
session: dict[str, Any] = _SESSION_DEP,
|
session: dict[str, Any] = _SESSION_DEP,
|
||||||
) -> StreamingResponse:
|
) -> StreamingResponse:
|
||||||
|
last_event_id = request.headers.get("last-event-id")
|
||||||
|
|
||||||
async def event_generator():
|
async def event_generator():
|
||||||
async for chunk in stream_dashboard_thread(thread_id, session["sub"]):
|
async for chunk in stream_dashboard_thread(
|
||||||
|
thread_id, session["sub"], last_event_id=last_event_id
|
||||||
|
):
|
||||||
yield chunk
|
yield chunk
|
||||||
|
|
||||||
return StreamingResponse(
|
return StreamingResponse(
|
||||||
|
|
|
||||||
|
|
@ -26,6 +26,7 @@ logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
_ASSISTANT_ID = "agent"
|
_ASSISTANT_ID = "agent"
|
||||||
_DASHBOARD_SOURCE = "dashboard"
|
_DASHBOARD_SOURCE = "dashboard"
|
||||||
|
_DASHBOARD_STREAM_MODES: tuple[str, ...] = ("values", "updates", "messages-tuple")
|
||||||
|
|
||||||
|
|
||||||
def _agent_version_metadata() -> dict[str, str]:
|
def _agent_version_metadata() -> dict[str, str]:
|
||||||
|
|
@ -292,6 +293,8 @@ async def _start_agent_run(
|
||||||
input={"messages": [{"role": "user", "content": prompt}]},
|
input={"messages": [{"role": "user", "content": prompt}]},
|
||||||
config={"configurable": configurable, "metadata": _agent_version_metadata()},
|
config={"configurable": configurable, "metadata": _agent_version_metadata()},
|
||||||
if_not_exists="create",
|
if_not_exists="create",
|
||||||
|
stream_mode=list(_DASHBOARD_STREAM_MODES),
|
||||||
|
stream_resumable=True,
|
||||||
)
|
)
|
||||||
run_id = run.get("run_id") if isinstance(run, dict) else getattr(run, "run_id", None)
|
run_id = run.get("run_id") if isinstance(run, dict) else getattr(run, "run_id", None)
|
||||||
await client.threads.update(
|
await client.threads.update(
|
||||||
|
|
@ -367,6 +370,8 @@ async def send_dashboard_message(
|
||||||
_ASSISTANT_ID,
|
_ASSISTANT_ID,
|
||||||
input={"messages": [{"role": "user", "content": prompt}]},
|
input={"messages": [{"role": "user", "content": prompt}]},
|
||||||
config={"configurable": configurable, "metadata": _agent_version_metadata()},
|
config={"configurable": configurable, "metadata": _agent_version_metadata()},
|
||||||
|
stream_mode=list(_DASHBOARD_STREAM_MODES),
|
||||||
|
stream_resumable=True,
|
||||||
)
|
)
|
||||||
run_id = run.get("run_id") if isinstance(run, dict) else getattr(run, "run_id", None)
|
run_id = run.get("run_id") if isinstance(run, dict) else getattr(run, "run_id", None)
|
||||||
await client.threads.update(
|
await client.threads.update(
|
||||||
|
|
@ -426,7 +431,9 @@ async def delete_dashboard_thread(thread_id: str, login: str) -> None:
|
||||||
await client.threads.delete(thread_id)
|
await client.threads.delete(thread_id)
|
||||||
|
|
||||||
|
|
||||||
async def stream_dashboard_thread(thread_id: str, login: str) -> AsyncIterator[str]:
|
async def stream_dashboard_thread(
|
||||||
|
thread_id: str, login: str, *, last_event_id: str | None = None
|
||||||
|
) -> AsyncIterator[str]:
|
||||||
try:
|
try:
|
||||||
thread = await langgraph_client().threads.get(thread_id)
|
thread = await langgraph_client().threads.get(thread_id)
|
||||||
except Exception as exc: # noqa: BLE001
|
except Exception as exc: # noqa: BLE001
|
||||||
|
|
@ -435,7 +442,20 @@ async def stream_dashboard_thread(thread_id: str, login: str) -> AsyncIterator[s
|
||||||
metadata = thread.get("metadata") if isinstance(thread.get("metadata"), dict) else {}
|
metadata = thread.get("metadata") if isinstance(thread.get("metadata"), dict) else {}
|
||||||
_assert_thread_owner(metadata, login)
|
_assert_thread_owner(metadata, login)
|
||||||
|
|
||||||
stream = await langgraph_client().threads.join_stream(thread_id)
|
stream = await langgraph_client().threads.join_stream(
|
||||||
|
thread_id,
|
||||||
|
last_event_id=last_event_id,
|
||||||
|
)
|
||||||
async for part in stream:
|
async for part in stream:
|
||||||
payload = part if isinstance(part, dict) else {"event": str(part)}
|
event = getattr(part, "event", None) or (
|
||||||
yield f"data: {json.dumps(payload, default=str)}\n\n"
|
part.get("event") if isinstance(part, dict) else None
|
||||||
|
)
|
||||||
|
data = getattr(part, "data", None) if not isinstance(part, dict) else part.get("data")
|
||||||
|
event_id = getattr(part, "id", None) if not isinstance(part, dict) else part.get("id")
|
||||||
|
payload: dict[str, Any] = {"event": event, "data": data}
|
||||||
|
if event_id is not None:
|
||||||
|
payload["id"] = event_id
|
||||||
|
chunk = f"data: {json.dumps(payload, default=str)}\n\n"
|
||||||
|
if event_id is not None:
|
||||||
|
chunk = f"id: {event_id}\n{chunk}"
|
||||||
|
yield chunk
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue